大营销平台 —— 第二阶段初始化以及DB路由中间件的使用及基本原理

一、前言

在很早之前，我们结束了大营销的第一阶段，第一阶段主要聚焦于抽奖策略业务，其实说白了，本质还是一个扩展性很高、高度解耦的CRUD，其实和Agent中装配Client类似，都是通过抽象业务节点来将业务处理独立分隔开，同时用组合模式、责任链模式来对节点的顺序和分支进行预定义，当然，这个预定义我们交给了客户来做，因此是从数据库中拿出定义好了的顺序和分支，用于工厂装填规则树和责任链，这个部分是比较繁琐的，尤其是对于库表的设计能力要求很高。

这一节将开启第二阶段，这个阶段主要聚焦于抽奖活动业务，区别于第一阶段的“怎么抽”，这一阶段的目标是实现“抽什么”，例如签到领取抽奖次数、参与指定活动来指定奖池等。因此将新创一个子领域，用于实现这些业务。

二、DB路由中间件

1.库表设计（半成品版）

这里我先略讲，因为在没有熟悉实际的业务流程前，对于库表的设计其实不会太清楚其功能，因此这里主要看看有哪几个表，以及重点在于学习如何进行分库分表。

并且后续会对库表进行重新设计，这个版本并不是最优雅的版本。

可以看到，我们这里实现了分库+分表的混合策略：

分库体现在：我们将与用户操作相关的表单独分出来，即big_market_01、big_market_02。

分表体现在：一个库中有多个相同结构的表（副本），即raffle_activity_account_flow000、raffle_activity_account_flow001、raffle_activity_account_flow002、raffle_activity_account_flow003。

无论是分库还是分表，最终目的都是希望减轻单个数据库的压力，利用多组服务器实现数据分片，缓解单库单表的压力。

同时分表可以加快CRUD的效率（不会在一张表上大量上锁）。

2.中间件使用

这里的DB路由中间件是xfg自研的，Maven如下：

<dependency> <groupId>cn.bugstack.middleware</groupId> <artifactId>db-router-spring-boot-starter</artifactId> <version>1.0.2</version> </dependency>

接下来我们来剖析一下源码，看看这个中间件是如何实现无感分库分表的。

首先先看如何使用，很简单，在需要分库分表存储的方法上面加上注解 @DBRouter即可：

@Mapper @DBRouterStrategy(splitTable = true) public interface IRaffleActivityOrderDao { @DBRouter(key = "userId") void insert(RaffleActivityOrder raffleActivityOrder); @DBRouter List<RaffleActivityOrder> queryRaffleActivityOrderByUserId(String userId); }

这个注解是作为要进行分库分表的标记，key参数是指定分库分表的具体字段，其中，这个字段将自动去方法的参数对象中去找，例如这里的key是userId，那么这个注解就会自动去RaffleActivityOrder的对象中去找userId的成员变量。

同理，下面的查询方法也需要加上注解，这样就表示进行分库分表的查询。

当然，我们需要指定库的个数以及表的个数，并且用于进行后续的Hash分片存储，当然，我们也要去配置每个数据库的基本信息。

mini-db-router: jdbc: datasource: dbCount: 2 tbCount: 4 default: db00 routerKey: userId list: db01,db02 db00: driver-class-name: com.mysql.cj.jdbc.Driver url: jdbc:mysql://192.168.242.130:3306/big_market?useUnicode=true&characterEncoding=utf8&autoReconnect=true&zeroDateTimeBehavior=convertToNull&serverTimezone=UTC&useSSL=true username: root password: 123456 type-class-name: com.zaxxer.hikari.HikariDataSource pool: pool-name: Retail_HikariCP minimum-idle: 15 #最小空闲连接数量 idle-timeout: 180000 #空闲连接存活最大时间，默认600000（10分钟） maximum-pool-size: 25 #连接池最大连接数，默认是10 auto-commit: true #此属性控制从池返回的连接的默认自动提交行为,默认值：true max-lifetime: 1800000 #此属性控制池中连接的最长生命周期，值0表示无限生命周期，默认1800000即30分钟 connection-timeout: 30000 #数据库连接超时时间,默认30秒，即30000 connection-test-query: SELECT 1 db01: driver-class-name: com.mysql.cj.jdbc.Driver url: jdbc:mysql://192.168.242.130:3306/big_market_01?useUnicode=true&characterEncoding=utf8&autoReconnect=true&zeroDateTimeBehavior=convertToNull&serverTimezone=UTC&useSSL=true username: root password: 123456 type-class-name: com.zaxxer.hikari.HikariDataSource pool: pool-name: Retail_HikariCP minimum-idle: 15 #最小空闲连接数量 idle-timeout: 180000 #空闲连接存活最大时间，默认600000（10分钟） maximum-pool-size: 25 #连接池最大连接数，默认是10 auto-commit: true #此属性控制从池返回的连接的默认自动提交行为,默认值：true max-lifetime: 1800000 #此属性控制池中连接的最长生命周期，值0表示无限生命周期，默认1800000即30分钟 connection-timeout: 30000 #数据库连接超时时间,默认30秒，即30000 connection-test-query: SELECT 1 db02: driver-class-name: com.mysql.cj.jdbc.Driver url: jdbc:mysql://192.168.242.130:3306/big_market_02?useUnicode=true&characterEncoding=utf8&autoReconnect=true&zeroDateTimeBehavior=convertToNull&serverTimezone=UTC&useSSL=true username: root password: 123456 type-class-name: com.zaxxer.hikari.HikariDataSource pool: pool-name: Retail_HikariCP minimum-idle: 15 #最小空闲连接数量 idle-timeout: 180000 #空闲连接存活最大时间，默认600000（10分钟） maximum-pool-size: 25 #连接池最大连接数，默认是10 auto-commit: true #此属性控制从池返回的连接的默认自动提交行为,默认值：true max-lifetime: 1800000 #此属性控制池中连接的最长生命周期，值0表示无限生命周期，默认1800000即30分钟 connection-timeout: 30000 #数据库连接超时时间,默认30秒，即30000 connection-test-query: SELECT 1

3.中间件原理

这里你可能就会有疑惑，这个路由是通过什么来实现的呢？

其实这里就必须去看源码了，最终会发现是使用和Java原生的HashMap一样的扰动函数，通过指定字段的hashCode来对操作对象进行分片，所以指定的key内容相同的方法，最终存入的片是一样的，例如前面的key是userId，那么如果存入两个userId内容都相同的对象是会存入同一个库的同一个表中去的。

核心的路由代码如下：

package cn.bugstack.middleware.db.router; /** * @description: 数据路由切面，通过自定义注解的方式，拦截被切面的方法，进行数据库路由 * @author: 小傅哥，微信：fustack * @date: 2021/9/22 * @github: https://github.com/fuzhengwei * @Copyright: 公众号：bugstack虫洞栈 | 博客：https://bugstack.cn - 沉淀、分享、成长，让自己和他人都能有所收获！ */ @Aspect public class DBRouterJoinPoint { private Logger logger = LoggerFactory.getLogger(DBRouterJoinPoint.class); private DBRouterConfig dbRouterConfig; private IDBRouterStrategy dbRouterStrategy; public DBRouterJoinPoint(DBRouterConfig dbRouterConfig, IDBRouterStrategy dbRouterStrategy) { this.dbRouterConfig = dbRouterConfig; this.dbRouterStrategy = dbRouterStrategy; } @Pointcut("@annotation(cn.bugstack.middleware.db.router.annotation.DBRouter)") public void aopPoint() { } /** * 所有需要分库分表的操作，都需要使用自定义注解进行拦截，拦截后读取方法中的入参字段，根据字段进行路由操作。 * 1. dbRouter.key() 确定根据哪个字段进行路由 * 2. getAttrValue 根据数据库路由字段，从入参中读取出对应的值。比如路由 key 是 uId，那么就从入参对象 Obj 中获取到 uId 的值。 * 3. dbRouterStrategy.doRouter(dbKeyAttr) 路由策略根据具体的路由值进行处理 * 4. 路由处理完成比，就是放行。 jp.proceed(); * 5. 最后 dbRouterStrategy 需要执行 clear 因为这里用到了 ThreadLocal 需要手动清空。关于 ThreadLocal 内存泄漏介绍 https://t.zsxq.com/027QF2fae */ @Around("aopPoint() && @annotation(dbRouter)") public Object doRouter(ProceedingJoinPoint jp, DBRouter dbRouter) throws Throwable { String dbKey = dbRouter.key(); if (StringUtils.isBlank(dbKey) && StringUtils.isBlank(dbRouterConfig.getRouterKey())) { throw new RuntimeException("annotation DBRouter key is null！"); } dbKey = StringUtils.isNotBlank(dbKey) ? dbKey : dbRouterConfig.getRouterKey(); // 路由属性 String dbKeyAttr = getAttrValue(dbKey, jp.getArgs()); // 路由策略 dbRouterStrategy.doRouter(dbKeyAttr); // 返回结果 try { return jp.proceed(); } finally { dbRouterStrategy.clear(); } } private Method getMethod(JoinPoint jp) throws NoSuchMethodException { Signature sig = jp.getSignature(); MethodSignature methodSignature = (MethodSignature) sig; return jp.getTarget().getClass().getMethod(methodSignature.getName(), methodSignature.getParameterTypes()); } public String getAttrValue(String attr, Object[] args) { if (1 == args.length) { Object arg = args[0]; if (arg instanceof String) { return arg.toString(); } } String filedValue = null; for (Object arg : args) { try { if (StringUtils.isNotBlank(filedValue)) { break; } // filedValue = BeanUtils.getProperty(arg, attr); // fix: 使用lombok时，uId这种字段的get方法与idea生成的get方法不同，会导致获取不到属性值，改成反射获取解决 filedValue = String.valueOf(this.getValueByName(arg, attr)); } catch (Exception e) { logger.error("获取路由属性值失败 attr：{}", attr, e); } } return filedValue; } /** * 获取对象的特定属性值 * * @author tang * @param item 对象 * @param name 属性名 * @return 属性值 */ private Object getValueByName(Object item, String name) { try { Field field = getFieldByName(item, name); if (field == null) { return null; } field.setAccessible(true); Object o = field.get(item); field.setAccessible(false); return o; } catch (IllegalAccessException e) { return null; } } /** * 根据名称获取方法，该方法同时兼顾继承类获取父类的属性 * * @author tang * @param item 对象 * @param name 属性名 * @return 该属性对应方法 */ private Field getFieldByName(Object item, String name) { try { Field field; try { field = item.getClass().getDeclaredField(name); } catch (NoSuchFieldException e) { field = item.getClass().getSuperclass().getDeclaredField(name); } return field; } catch (NoSuchFieldException e) { return null; } } }

对于MyBatis进行拦截，反射修改SQL语句，从而实现无感分库分表：

/** * @description: Mybatis 拦截器，通过对 SQL 语句的拦截处理，修改分表信息 * @author: 小傅哥，微信：fustack * @date: 2021/9/30 * @github: https://github.com/fuzhengwei * @Copyright: 公众号：bugstack虫洞栈 | 博客：https://bugstack.cn - 沉淀、分享、成长，让自己和他人都能有所收获！ */ @Intercepts({@Signature(type = StatementHandler.class, method = "prepare", args = {Connection.class, Integer.class})}) public class DynamicMybatisPlugin implements Interceptor { private Pattern pattern = Pattern.compile("(from|into|update)[\\s]{1,}(\\w{1,})", Pattern.CASE_INSENSITIVE); @Override public Object intercept(Invocation invocation) throws Throwable { // 获取StatementHandler StatementHandler statementHandler = (StatementHandler) invocation.getTarget(); MetaObject metaObject = MetaObject.forObject(statementHandler, SystemMetaObject.DEFAULT_OBJECT_FACTORY, SystemMetaObject.DEFAULT_OBJECT_WRAPPER_FACTORY, new DefaultReflectorFactory()); MappedStatement mappedStatement = (MappedStatement) metaObject.getValue("delegate.mappedStatement"); // 获取自定义注解判断是否进行分表操作 String id = mappedStatement.getId(); String className = id.substring(0, id.lastIndexOf(".")); Class<?> clazz = Class.forName(className); DBRouterStrategy dbRouterStrategy = clazz.getAnnotation(DBRouterStrategy.class); if (null == dbRouterStrategy || !dbRouterStrategy.splitTable()){ return invocation.proceed(); } // 获取SQL BoundSql boundSql = statementHandler.getBoundSql(); String sql = boundSql.getSql(); // 替换SQL表名 USER 为 USER_03 Matcher matcher = pattern.matcher(sql); String tableName = null; if (matcher.find()) { tableName = matcher.group().trim(); } assert null != tableName; String replaceSql = matcher.replaceAll(tableName + "_" + DBContextHolder.getTBKey()); // 通过反射修改SQL语句 Field field = boundSql.getClass().getDeclaredField("sql"); field.setAccessible(true); field.set(boundSql, replaceSql); field.setAccessible(false); return invocation.proceed(); } }

4.测试

写单元测试来对路由中间件测试，看看是否成功分库分表：

主库测试：

@Slf4j @RunWith(SpringRunner.class) @SpringBootTest public class RaffleActivityDaoTest { @Resource private IRaffleActivityDao raffleActivityDao; @Test public void test_queryRaffleActivityByActivityId() { RaffleActivity raffleActivity = raffleActivityDao.queryRaffleActivityByActivityId(100301L); log.info("测试结果：{}", JSON.toJSONString(raffleActivity)); } }

分库分表测试：

@Slf4j @RunWith(SpringRunner.class) @SpringBootTest public class RaffleActivityOrderDaoTest { @Resource private IRaffleActivityOrderDao raffleActivityOrderDao; private final EasyRandom easyRandom = new EasyRandom(); @Test public void test_insert_random() { for (int i = 0; i < 5; i++) { RaffleActivityOrder raffleActivityOrder = new RaffleActivityOrder(); // EasyRandom 可以通过指定对象类的方式，随机生成对象值。如；easyRandom.nextObject(String.class)、easyRandom.nextObject(RaffleActivityOrder.class) raffleActivityOrder.setUserId(easyRandom.nextObject(String.class)); raffleActivityOrder.setActivityId(100301L); raffleActivityOrder.setActivityName("测试活动"); raffleActivityOrder.setStrategyId(100006L); raffleActivityOrder.setOrderId(RandomStringUtils.randomNumeric(12)); raffleActivityOrder.setOrderTime(new Date()); raffleActivityOrder.setState("not_used"); // 插入数据 raffleActivityOrderDao.insert(raffleActivityOrder); } } @Test public void test_insert() { RaffleActivityOrder raffleActivityOrder = new RaffleActivityOrder(); raffleActivityOrder.setUserId("xiaofuge"); raffleActivityOrder.setActivityId(100301L); raffleActivityOrder.setActivityName("测试活动"); raffleActivityOrder.setStrategyId(100006L); raffleActivityOrder.setOrderId(RandomStringUtils.randomNumeric(12)); raffleActivityOrder.setOrderTime(new Date()); raffleActivityOrder.setState("not_used"); // 插入数据 raffleActivityOrderDao.insert(raffleActivityOrder); } @Test public void test_queryRaffleActivityOrderByUserId() { String userId = "xiaofuge"; List<RaffleActivityOrder> raffleActivityOrders = raffleActivityOrderDao.queryRaffleActivityOrderByUserId(userId); log.info("测试结果：{}", JSON.toJSONString(raffleActivityOrders)); } }

五、第二阶段初始化

1.库表设计（正式版）

前面的库表是一个半成品，是可以实现功能，但是效率不够高，并且比较难以维护的版本，下面的就是正式版本了，后续活动流程就主要依赖这个版本了。

半成品和正式版对比如下：

具体的库表设计如下：

其中sku表示库存表，作为整个活动配置的中间表，关联活动次数表和活动信息表。sku表的关联图如下：

至于活动订单表和活动账户表由于都是和用户相关的，会频繁操作，因此采用分库分表。

2.活动领域初始化

整体架构如下：

其实初始化并不是很复杂，主要就是补出仓储相关的类，然后补出需要的实体类。

其实这里我认为他讲解的顺序不太好，按照我的理解，应该以模板抽象类作为骨架，后续使用到不同的实体类才一个一个慢慢去补出来，实体类并不是一开始就能设计好的。

抽象类如下，这个抽象类实现一个抽奖订单的接口，因此需要实现创建活动订单的方法，这里主要目的是拿到（创建）活动订单实体类，为了将指定活动订单实体对象中的成员变量都拿到，我们还需要拿到其他三个实体对象，而这三个实体对象我们是直接从仓储拿的，因此在仓储中需要自行补三个方法。

/** * @author 印东升 * @description * @create 2026-07-16 16:12 */ @Slf4j public abstract class AbstractRaffleActivity implements IRaffleOrder{ protected IActivityRepository activityRepository; public AbstractRaffleActivity(IActivityRepository activityRepository) { this.activityRepository = activityRepository; } @Override public ActivityOrderEntity createRaffleActivityOrder(ActivityShopCartEntity activityShopCartEntity) { //用sku查询活动信息 ActivitySkuEntity activitySkuEntity = activityRepository.queryActivitySku(activityShopCartEntity.getSku()); //查询活动信息 ActivityEntity activityEntity = activityRepository.queryRaffleActivityByActivityId(activitySkuEntity.getActivityId()); //查询次数信息（用户在活动上可参与的次数） ActivityCountEntity activityCountEntity = activityRepository.queryRaffleActivityCountByActivityCountId(activitySkuEntity.getActivityCountId()); log.info("查询结果：{} {} {}", JSON.toJSONString(activitySkuEntity),JSON.toJSONString(activityEntity),JSON.toJSONString(activityCountEntity)); return ActivityOrderEntity.builder().build(); } }

仓储如下，实现类这里我就不写了，就是简单的dao操作：

/** * @author Fuzhengwei bugstack.cn @小傅哥 * @description 活动仓储接口 * @create 2024-03-16 10:31 */ public interface IActivityRepository { ActivitySkuEntity queryActivitySku(Long sku); ActivityEntity queryRaffleActivityByActivityId(Long activityId); ActivityCountEntity queryRaffleActivityCountByActivityCountId(Long activityCountId); }

3.聚合根

这里新增了一个聚合根，由于是第一次接触聚合根（前面只在Agent项目中的装配过程中接触到了聚合的概念），因此我们这里将着重讲一下聚合根这个DDD的概念。

聚合根的本质就是将几个类放在一个类中，这个过程就是聚合，聚合根本身就是个充血模型。

使用聚合根的目的：

1.通过事务来保证聚合根内的对象数据一致

比如这里面我们聚合了账户和订单，原因就是这两个 Entity对象 必须同生共死——如果扣了次数但订单没落库，用户次数白白浪费；如果订单插入了但次数没扣，用户可以无限下单。

2.校验数据

比如这个聚合根中的账户对象和订单对象，他们都要保证：

• totalCountSurplus > 0（总次数没花完）
• dayCountSurplus > 0（今天的额度还有）
• monthCountSurplus > 0（这个月的额度还有）

对于这些变量的数值校验，如果在Service中自己写校验流程就会很麻烦，因为每次更改某个实体对象都要去校验，但是如果我们在聚合根中写一个校验方法，在Service中不去操作某个具体的实体对象，而是调用聚合根的方法，那么这个校验就只需要调用一次聚合根内的方法就可以了。甚至可以将扣减方法直接写在聚合根中，内含校验+扣减业务，对于Service就只需要调用一次聚合根的扣减方法就能直接实现校验+扣减了。

这也是充血模型的最大优势——减少繁琐且具有流程化的操作。

/** * @author Fuzhengwei bugstack.cn @小傅哥 * @description 下单聚合对象 聚合类保证事务的统一 * @create 2024-03-16 10:32 */ @Data @Builder @AllArgsConstructor @NoArgsConstructor public class CreateOrderAggregate { /** * 活动账户实体 */ private ActivityAccountEntity activityAccountEntity; /** * 活动订单实体 */ private ActivityOrderEntity activityOrderEntity; }