大营销平台 —— 第二阶段初始化以及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 groupIdcn.bugstack.middleware/groupId artifactIddb-router-spring-boot-starter/artifactId version1.0.2/version /dependency接下来我们来剖析一下源码看看这个中间件是如何实现无感分库分表的。首先先看如何使用很简单在需要分库分表存储的方法上面加上注解 DBRouter即可Mapper DBRouterStrategy(splitTable true) public interface IRaffleActivityOrderDao { DBRouter(key userId) void insert(RaffleActivityOrder raffleActivityOrder); DBRouter ListRaffleActivityOrder 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?useUnicodetruecharacterEncodingutf8autoReconnecttruezeroDateTimeBehaviorconvertToNullserverTimezoneUTCuseSSLtrue username: root password: 123456 type-class-name: com.zaxxer.hikari.HikariDataSource pool: pool-name: Retail_HikariCP minimum-idle: 15 #最小空闲连接数量 idle-timeout: 180000 #空闲连接存活最大时间默认60000010分钟 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?useUnicodetruecharacterEncodingutf8autoReconnecttruezeroDateTimeBehaviorconvertToNullserverTimezoneUTCuseSSLtrue username: root password: 123456 type-class-name: com.zaxxer.hikari.HikariDataSource pool: pool-name: Retail_HikariCP minimum-idle: 15 #最小空闲连接数量 idle-timeout: 180000 #空闲连接存活最大时间默认60000010分钟 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?useUnicodetruecharacterEncodingutf8autoReconnecttruezeroDateTimeBehaviorconvertToNullserverTimezoneUTCuseSSLtrue username: root password: 123456 type-class-name: com.zaxxer.hikari.HikariDataSource pool: pool-name: Retail_HikariCP minimum-idle: 15 #最小空闲连接数量 idle-timeout: 180000 #空闲连接存活最大时间默认60000010分钟 maximum-pool-size: 25 #连接池最大连接数默认是10 auto-commit: true #此属性控制从池返回的连接的默认自动提交行为,默认值true max-lifetime: 1800000 #此属性控制池中连接的最长生命周期值0表示无限生命周期默认1800000即30分钟 connection-timeout: 30000 #数据库连接超时时间,默认30秒即30000 connection-test-query: SELECT 13.中间件原理这里你可能就会有疑惑这个路由是通过什么来实现的呢其实这里就必须去看源码了最终会发现是使用和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; ListRaffleActivityOrder 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; }