分布式事务之可靠消息最终一致性方案实践

﹏ヽ暗。殇╰゛Y 2023-10-11 09:44 31阅读 0赞

## 摘要 ##

如果你选用的消息中间件事rocketMQ,可以直接使用它的事务消息方便的实现分布式事务。但如果你用的是rabbitMQ或kafka这些没有事务消息的中间件，就需要自己来设计分布式事务的实现方案了。一种经典的方案是[本地消息表][Link 1]，通过本地事务+定时任务来实现最终一致。本文介绍另一种方案，通过自己实现可靠性消息服务，来保证消息发布方和消费方的最终一致。

## 方案设计 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xhbWJfSVQ_size_16_color_FFFFFF_t_70]  
该设计方案来自于[石杉架构笔记][Link 2].

#### 1，上游服务发送消息到可靠消息服务 ####

上游服务告诉可靠消息服务保存一条待确认的消息，然后就执行业务逻辑操作本地数据库，如果业务操作成功就告诉可靠消息服务更新消息为“已发送”并投递消息到MQ；如果业务操作失败了，就告诉可靠消息服务删除之前的“待确认”状态的消息。

如果操作本地数据库成功了（步骤3），但通知可靠消息服务更新消息状态（步骤4）时失败了怎么办？  
如果发生了这种情况，可靠消息服务中的消息会继续保持“待确认”的状态下去，相当于消息丢了，也没有别的影响。如果想做的更好些，可以在可靠消息服务开一个后台线程运行定时任务，检查状态为“待确认”且已过期的消息，询问上游服务相关业务操作是否成功了，如果成功就更新消息状态为“已发送”同时投递消息到MQ；否则删除消息。

#### 2，可靠消息服务投递消息到MQ ####

在可靠消息服务中更新消息为“已发送”和将消息投递到MQ这两个操作要放到一个本地事务中，保证：

*  如果数据库里更新消息的状态失败了，那么就抛异常退出了，就别投递到MQ；
 *  如果投递MQ失败报错了，那么就要抛异常让本地数据库事务回滚。

这俩操作必须得一起成功，或者一起失败。

#### 3，下游服务消费MQ消息 ####

下游服务消费MQ消息后，执行业务操作，如果成功了告知MQ消费成功，并告知可靠消息服务更新消息状态为“已完成”。

那如果下游服务消费消息出了问题，没消费到？或者是下游服务对消息的处理失败了，怎么办？  
仍然可以在可靠消息服务后台开一个线程执行定时任务，检查状态为“已发送”且超时的消息，再次将此消息投递到MQ让下游消费，需要下游保证消费消息的幂等性。

## 业务场景 ##

现在有电话营销系统（后面简称电销系统）和调度平台两个系统。电销系统处理具体电销案件作业逻辑，依赖于调度平台将案件分配给人处理。电销系统中有marketing\_task表维护电销案件，调度平台有dispatching\_task表维护包括电销系统、催收系统、审批系统等业务系统在内的多种业务案件。  
电销系统生成电销案件后，是通过MQ中间件将案件信息推送调度平台，后者异步将案件保存并完成后续分配。

下面我们就来看下如何使用可靠消息服务来保证电销系统和调度平台之间，案件状态的最终一致。

## 方案实践 ##

首先，定义可靠消息服务接口

public interface ReliableMsgService { 
        // 保存一条消息，状态位“待确认”，消息体包含上游要发送给下游的信息
        public ReliableMsgResponse prepare(ReliableMsgRequest request);
        // 供上游数据库事务成功时调用，消息确认做两件事：1，更新消息状态为“已发送”；2，将消息内容投递到MQ中间件。
        public ReliableMsgResponse confirm(ReliableMsgRequest request);
        // 供上游数据库事务失败时调用，删除消息
        public ReliableMsgResponse delete(ReliableMsgRequest request);
        // 供下游在消息消费成功后调用，更新消息状态为“已完成”
       public ReliableMsgResponse finish(ReliableMsgRequest request);
    }

注意接口实现的confirm方法要定义本地事务，保证两件事同时成功或失败.

在电销推送案件到调度的业务场景中，电销作为上游系统与可靠消息服务交互如下：

public class TaskService{ 
        @Autowired
        private ReliableMsgService reliableMsg;
        
        public void pushTask(String caseNo){ 
            // 省略将消息体放入request代码
            ...
            try{ 
                // 发送待确认消息
                reliableMsg.prepare(request);
            }catch(Exception e){ 
                log.error('调用可靠消息服务prepare出现异常',e)
                return;
            }
            
            try{ 
                // 更新电销案件表，将案件设为处理中
                updateTaskStatus(caseNo, HANDLING);
                // 更新成功，发送确认消息
                reliable.confirm(request);
            }catch(Exception e){ 
                // 更新出现异常，发送删除消息
                reliable.delete(request);
            }
        }
    }

作为下游服务的调度平台消费MQ消息逻辑如下：

public void CreateTaskConsumer{ 
        @Autowired
        private ReliableMsgService reliableMsg;   
         
        public void consume(Message msg){ 
            // 生成电销案件
            insertDispachingTask(msg);
            // 此处省略准备request代码
            ...
            // 通知可靠消息服务，消息处理完毕
            reliableMsg.finish(request);
        }
    }

上面的都是主流程中的代码，我们再看一下出现异常时的处理，主要是方案里提到的可靠消息服务中定义的两个定时任务。

定时任务1

// 查询“待确认”且超时的消息
    public class HandleOverTimePrepareJob { 
        @Autowire
        private TaskStatusQueryFacade taskStatusQuery;
        @Autowire
        private ReliableMsgService reliableMsg;
        
        public void execute(JobContext context){ 
            // 查找状态为“待确认”同时超时了的消息
            List<RMessage> list = queryOverTimePrepareMsg();
            for(RMessage msg:list){ 
              Boolean res = taskStatusQuery.updateSucceed(msg.caseNo);
              //省略准备request代码
              ...
              if(res){ 
                  reliableMsg.confirm(request);
              }else{ 
                  reliableMsg.delete(request);
              }
            }
        }
    }

对应电销系统接口定义：

// 电销系统接口
    public interface TaskStatusQueryFacade{ 
    	// 电销系统更新案件状态是否成功
        public Boolean updateSucceed(String caseNo);
    }

定时任务2

// 查询“已发送”且超时的消息
    public class HandleOverTimeConfirmJob { 
        @Autowire
        private ReliableMsgService reliableMsg;
        
        public void execute(JobContext context){ 
            // 查找状态为“已发送”同时超时了的消息
            List<RMessage> list = queryOverTimeConfirmedMsg();
            for(RMessage msg:list){ 
              //省略准备request代码
              ...
              // 再次投递消息到MQ
              reliableMsg.confirm(request);
            }
        }
    }

## 总结 ##

对比本地消息表方案可以看出，可靠消息服务方案其实就是把消息表的维护从发送方的数据库独立出来作为单独的服务。优点是可以使上游系统更加专注于业务代码。其实RocketMQ中的事务消息的思路与本文描述的方案类似，如果你使用的是RocketMQ直接使用事务消息更方便，如果是其他不支持事务消息的MQ中间件，可以考虑本方案。

[Link 1]: https://blog.csdn.net/Lamb_IT/article/details/101145963
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xhbWJfSVQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20190922145412357.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xhbWJfSVQ=,size_16,color_FFFFFF,t_70
[Link 2]: https://juejin.im/post/5bf2c6b6e51d456693549af4