分布式事务控制方案
本篇文章给出一种要求高可用性（AP思想）的分布式事务控制方案
下篇新鲜出炉：点我查看
- 分布式事务控制方案
- 1、业务背景
- 2、本地消息表的设计
- 3、对消息表的操作
- 4、任务调度
- 5、任务流程控制的抽象类
- 6、课程发布的实现类
- 7、总结
1、业务背景
业务背景：在线学习平台，教学机构在上传课程时，需要将课程内容同步到数据库，缓存，文件系统，搜索系统，这里需要用到分布式事务，来确保四个组件的业务顺利完成。
在CAP理论中，分布式系统只能满足一致性C、可用性A、分区容错性P三者中的两个，由于分布式系统天然要求分区容错，否则就是单体项目，所以只能选CP或AP
其中CP可以使用Seata框架基于AT和TCC模式去实现，AP也有多种实现方式。
我们的业务背景中这四个组件并不要求强一致性，而是要求高可用性，如果其中某个组件没有完成数据同步，那之前已经完成的组件不必回退到事务开始前的状态，所以我们实现AP思想，采用本地消息表+任务调度完成最终一致性
具体的项目环境：SpringBoot框架，数据库MySQL，使用MyBatis-Plus快速开发，缓存Redis，分布式文件系统MinIO，搜索系统ES
2、本地消息表的设计
在业务背景中，如果用户要进行课程发布，我们向MySQL中的消息表里插入一条记录，记录中应当包含四个组件（MySQL，Redis，MinIO，ES）完成的状态，如果四个组件全部完成，就删除这条记录，向历史消息表中插入一条记录，如果四个组件有哪个没有完成，通过查询记录就可以从未完成的地方重新进行数据同步，从而实现最终一致性。
我们除了课程这个业务场景，还会在其他业务场景执行相似的业务，所以我们要考虑如何进行代码的抽象、封装和复用。
我们发现在消息表中没有必要注明每个具体的组件，而是通过小任务一，小任务二…的方式设计数据表，具体的业务逻辑由具体的业务代码实现，而具体的业务代码通过继承抽象的类，来实现对数据库的控制。
设计一个抽象的类，这个类应当实现对数据表的处理，并提供一个接口，让具体的业务代码实现这个接口。
3、对消息表的操作
首先创建数据库和数据表，表中字段包括业务相关字段（消息类型代码，关联业务信息、代码等等），小任务的状态、上一次成功失败时间、重试次数（暂定五个小任务，提高适用性）。
其次创建一个微服务模块，添加MyBatis-Plus依赖和配置
<!-- MySQL 驱动 -->
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<scope>runtime</scope>
</dependency>
<!-- mybatis plus的依赖 -->
<dependency>
<groupId>com.baomidou</groupId>
<artifactId>mybatis-plus-boot-starter</artifactId>
</dependency>
spring:
application:
name: service
datasource:
driver-class-name: com.mysql.cj.jdbc.Driver
url: jdbc:mysql://localhost:3306/table?serverTimezone=UTC&userUnicode=true&useSSL=false&
username: xxxx
password: xxxx
实现DAO和service层开发
在实现流程控制的抽象类之前我们要思考一个问题，具体的业务代码在实现了任务后，如何开启调用任务的执行？
首先肯定不是在发布课程的方法中直接调用方法，这是同步调用，并且只能执行一次，不适合AP思想的分布式事务。对于数据同步实时性要求不高的技术解决方案有很多，例如MQ、Canal、Logstash、任务调度等等
我们可以在插入数据表后，向消息队列添加一条消息，消费者收到消息后检查数据库是否存在对应记录，没有就执行一次任务，如果任务执行失败，就像消息队列添加一条消息。
我们也可以使用中间件canal，解决耦合性问题，canal通过模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 请求，得到 binglog 日志，
我们也可以通过任务调度的方式执行任务，在任务流程代码中查询数据库，根据数据库中的记录执行任务。
这里我们采用任务调度的方式执行任务。
4、任务调度
任务调度是指对计算任务进行合理安排和调度的过程。分布式任务调度是指在分布式系统中，将任务分割成若干份，根据调度规则交由不同的实例并行执行。
XXL-JOB，是一个轻量级分布式任务调度平台，开发迅速，学习简单，易扩展。包含调度中心（管理执行器、任务、日志，监控运维），执行器（注册服务，执行服务，执行结果上报，记录日志）和任务（具体的业务代码）。
如何确保任务不重复执行？任务调度采用分片广播的方式，查询数据表得到任务的id（自增id），模上分片总数，如果等于当前执行器的任务号，就执行该任务，否则不执行，对于任务分配超出执行器执行能力的情况，通过合理设置任务广播频率，以及设置任务拒绝策略为丢弃任务来确保没有任务被重复执行。
执行流程：
启动XXL-JOB调度中心，创建执行器和任务，任务执行实现通过cron表达式设置为每小时一次，设置分片广播和丢弃策略。
在课程发布微服务中添加XXL-JOB的依赖、配置文件和配置类，创建任务方法，在该方法前添加注解@XxlJob(“JobHandler”)
@XxlJob("JobHandler")
public void coursePublishJobHandler() throws Exception {
// 分片参数
int shardIndex = XxlJobHelper.getShardIndex();
int shardTotal = XxlJobHelper.getShardTotal();
// 在下一节中实现process，这里仅是测试方法
System.out.println("XXL-JOB任务调度测试成功");
// process(......);
}
启动微服务，可以在物理地址中看到一个实例，执行一次任务，在输出窗口看到
我们可以发现，任务的触发是根据创建任务时设置的执行时间来完成的，但是从用户的角度出发，有些用户允许在一段时间以内完成数据同步，例如一到两个工作日内完成，但是有的用户希望在发布课程后能及时的完成数据同步，例如一小时内完成数据同步，这就需要在代码端对xxl-job的控制中心进行通知，xxl-job也提供了这个接口
打开从github下载的xxl-job项目，在JobInfoController的接口中有体现，通过调用start方法，传入任务的id，来触发一次任务。
下面是代码逻辑
@Override
public ReturnT<String> start(int id) {
XxlJobInfo xxlJobInfo = xxlJobInfoDao.loadById(id);
// valid
ScheduleTypeEnum scheduleTypeEnum = ScheduleTypeEnum.match(xxlJobInfo.getScheduleType(), ScheduleTypeEnum.NONE);
if (ScheduleTypeEnum.NONE == scheduleTypeEnum) {
return new ReturnT<String>(ReturnT.FAIL_CODE, (I18nUtil.getString("schedule_type_none_limit_start")) );
}
// next trigger time (5s后生效，避开预读周期)
long nextTriggerTime = 0;
try {
Date nextValidTime = JobScheduleHelper.generateNextValidTime(xxlJobInfo, new Date(System.currentTimeMillis() + JobScheduleHelper.PRE_READ_MS));
if (nextValidTime == null) {
return new ReturnT<String>(ReturnT.FAIL_CODE, (I18nUtil.getString("schedule_type")+I18nUtil.getString("system_unvalid")) );
}
nextTriggerTime = nextValidTime.getTime();
} catch (Exception e) {
logger.error(e.getMessage(), e);
return new ReturnT<String>(ReturnT.FAIL_CODE, (I18nUtil.getString("schedule_type")+I18nUtil.getString("system_unvalid")) );
}
xxlJobInfo.setTriggerStatus(1);
xxlJobInfo.setTriggerLastTime(0);
xxlJobInfo.setTriggerNextTime(nextTriggerTime);
xxlJobInfo.setUpdateTime(new Date());
xxlJobInfoDao.update(xxlJobInfo);
return ReturnT.SUCCESS;
}
上述代码逻辑：
首先根据传入的任务id获取任务，判断任务的触发方式，如果为空，就报错，返回异常码500。
然后设置任务下一次的触发时间，为当前时间的5s后，更新触发状态，返回success。
我们可以在课程发布的逻辑中中调用这个方法，来实现及时同步课程内容。
注意，需要在调用方法头添加注解@PermissionLimit(limit = false)，来绕开登录验证，但是这增加了代码的不安全性，需要对这种权限的使用进行限制。
@RequestMapping("/startJob")
@ResponseBody
@PermissionLimit(limit = false)
public ReturnT<String> startJob(@RequestBody XxlJobInfo jobInfo) {
return xxlJobService.start(jobInfo.getId());
}
5、任务流程控制的抽象类
接下来实现抽象类，在这个类中需要提供任务执行的流程，而非具体的代码，提供一个抽象方法，业务代码通过实现这个抽象方法，在这个方法中实现具体的业务执行代码
此时我们已经得到分片总数和分片号，通过查询数据库中记录，自增id模上分片总数等于分片号的方式判断是否由当前执行实例实行
MyBatis-Plus没有提供查询方法，在Mapper中进行实现
@Select("SELECT t.* FROM mq_message t WHERE t.id % #{shardTotal} = #{shardindex} and t.state='0' and t.message_type=#{messageType} limit #{count}")
List<MqMessage> selectListByShardIndex(@Param("shardTotal") int shardTotal, @Param("shardindex") int shardindex, @Param("messageType") String messageType,@Param("count") int count);
在得到消息记录后，这是一个列表的形式，我们开启线程池，使用newFixedThreadPool，线程总数就是任务数，没有临时线程，使用CountDownLatch控制线程完成情况，每个线程中执行process方法，process是一个抽象方法，由具体的实现类进行实现，返回一个boolean变量，表示任务是否完成，并记录日志。
public abstract class MessageProcessAbstract {
@Autowired
MqMessageService mqMessageService;
/**
* @param mqMessage 执行任务内容
* @return boolean true:处理成功，false处理失败
* @description 任务处理
* @author zkp15
* @date 2023/9/21 19:47
*/
public abstract boolean execute(MqMessage mqMessage);
/**
* @description 扫描消息表多线程执行任务
* @param shardIndex 分片序号
* @param shardTotal 分片总数
* @param messageType 消息类型
* @param count 一次取出任务总数
* @param timeout 预估任务执行时间,到此时间如果任务还没有结束则强制结束 单位秒
* @return void
* @author zkp15
* @date 2023/9/21 20:35
*/
public void process(int shardIndex, int shardTotal, String messageType,int count,long timeout) {
try {
//扫描消息表获取任务清单
List<MqMessage> messageList = mqMessageService.getMessageList(shardIndex, shardTotal,messageType, count);
//任务个数
int size = messageList.size();
if(size<=0){
return ;
}
//创建线程池
ExecutorService threadPool = Executors.newFixedThreadPool(size);
//计数器
CountDownLatch countDownLatch = new CountDownLatch(size);
messageList.forEach(message -> {
threadPool.execute(() -> {
//处理任务
try {
boolean result = execute(message);
if(result){
//更新任务状态,删除消息表记录,添加到历史表
int completed = mqMessageService.completed(message.getId());
if (completed>0){
log.debug("任务执行成功:{}",message);
}else{
log.debug("任务执行失败:{}",message);
}
}
} catch (Exception e) {
e.printStackTrace();
log.debug("任务出现异常:{},任务:{}",e.getMessage(),message);
}
//计数
countDownLatch.countDown();
});
});
//等待,给一个充裕的超时时间,防止无限等待，到达超时时间还没有处理完成则结束任务
countDownLatch.await(timeout,TimeUnit.SECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
6、课程发布的实现类
在这个实现类中，继承上一节的抽象类，以及抽象方法execute，在execute中，分别执行数据库操作，建立缓存，上传分布式文件系统，建立搜索索引。
@Component
public class CoursePublishTask extends MessageProcessAbstract {
......
//任务调度入口
@XxlJob("CoursePublishJobHandler")
public void coursePublishJobHandler() throws Exception {
// 分片参数
int shardIndex = XxlJobHelper.getShardIndex();
int shardTotal = XxlJobHelper.getShardTotal();
log.debug("shardIndex="+shardIndex+",shardTotal="+shardTotal);
//参数:分片序号、分片总数、消息类型、一次最多取到的任务数量、一次任务调度执行的超时时间
process(shardIndex,shardTotal,"course_publish",30,60);
}
//课程发布任务处理
@Override
public boolean execute(MqMessage mqMessage) {
//获取消息相关的业务信息
String businessKey1 = mqMessage.getBusinessKey1();
long courseId = Integer.parseInt(businessKey1);
// 课程发布表
saveCourseToMQ(mqMessage, courseId);
// 课程缓存
saveCourseCache(mqMessage, courseId);
// 课程静态化
generateCourseHtml(mqMessage, courseId);
// 课程索引
saveCourseIndex(mqMessage, courseId);
return true;
}
......
}
7、总结
本文在实际开发业务场景的基础上，给出了一种遵循AP思想的分布式事务控制方案，通过本地消息表+任务调度的方式实现。
项目亮点有：
-
本地消息表通过任务123代替具体的任务，结合流程控制抽象类，只给出流程控制的代码，具体的业务实现由具体的实现类完成，从而实现解耦合，提高代码复用。
-
任务流程控制中开启多实例和多线程，并行高效的执行任务。
-
使用任务调度XXL-JOB进行任务执行，采用分片广播的方式，保证了任务执行的幂等性。其中控制中心提供了两种任务调度的规则，按照Cron的定时执行策略，和非登录任务执行通知的及时执行策略，为用户提供了多样化的体验
由于篇幅原因，四个小任务的实现，数据库、缓存、文件系统、搜索系统的数据同步，我们放在下一篇继续论述
