三、基于可靠消息的最终一致性方案详解

（一）、消息发送一致性

消息中间件在分布式系统中的核心作用就是异步通讯、应用解耦和并发缓冲（也叫作流量削峰）。在分布式环境下，需要通过网络进行通讯，就引入了数据传输的不确定性，也就是CAP理论中的分区容错性。

消息发送一致性是指产生消息的业务动作与消息发送一致，也就是说如果业务操作成功，那么由这个业务操作所产生的消息一定要发送出去，否则就丢失。

处理方式一

public void completeOrderService() {
    // 处理订单
    order.process();
    // 发送会计原始凭证消息
    pipe.sendAccountingVouchetMessage();
}

在上面的情况中，如果业务操作成功，执行的消息发送之前应用发生故障，消息发送不出去，导致消息丢失，将会产生订单系统与会计系统的数据不一致。如果消息系统或者网络异常，也会导致消息发送不出去，也会造成数据不一致。

处理方式二

public void completeOrderService() {
    // 发送会计原始凭证消息
    pipe.sendAccountingVouchetMessage();
    // 处理订单
    order.process();
}

如果将上面的两个操作调换一下顺序，这种情况就会更加不可控了，消息发出去了业务订单可能会失败，会造成订单系统与业务系统的数据不一致。那么JMS标准中的XA协议是否可以保障发送的一致性？

• JMS协议标准的API中，有很多以XA开头的接口，其实就是前面讲到的支持XA协议（基于两阶段提交协议）的全局事务型接口。

XAConnection.class
XAConnectionFactory.class
XAQueueConnection.class
XAQueueConnectionFactory.class
XASession.class
XATopicConnection.class
XATopicConnectionFactory.class
XATopicSession.class

• JMS中的XA系列的接口可以提供分布式事务的支持。但是引用XA方式的分布式事务，就会带来很多局限性。
  

1. 要求业务操作的资源必须支持XA协议，但是并不是所有的资源都支持XA协议。
   
2. 两阶段提交协议的成本。
   
3. 持久化成本等DTP模型的局限性，例如：全局锁定、成本高、性能低。
   
4. 使用XA协议违背了柔性事务的初衷。

（二）、保证消息一致的变通做法

1. 发送消息：主动方现将应用把消息发给消息中间件，消息状态标记为“待确认”状态。
   
2. 消息中间件收到消息后，把消息持久化到消息存储中，但是并不影响被动方投递消息。
   
3. 消息中间件返回消息持久化结果，主动方根据返回的结果进行判断如何进行业务操作处理：
   

1. 失败：放弃执行业务操作处理，结束，必要时向上层返回处理结果。
   
2. 成功：执行业务操作处理。
   

4. 业务操作完成后，把业务操作结果返回给消息中间件。
   
5. 消息中间件收到业务操作结构后，根据业务结果进行处理：
   

1. 失败：删除消息存储中的消息，结束。
   
2. 成功：更新消息存储中的消息状态为“待发送”，然后执行消息投递。
   

6. 前面的正向流程都成功之后，向被动方应用投递消息。
   

但是在上面的处理流程中，任何一个环节都有可能出现问题。

（三）、常规MQ消息处理流程和特点

• 常规的MQ队列处理流程无法实现消息的一致性。
  
• 投递消息的本质就是消息消费，可以细化。

（四）、消息重复发送问题和业务接口幂等性设计

对于未确认的消息，采用按规则重新投递的方式进行处理。对于以上流程，消息重复发送会导致业务处理接口出现重复调用的问题。消息消费过程中消息重复发送的主要原因就是消费者成功接收处理完消息后，消息中间件没有及时更新投递状态导致的。如果允许消息重复发送，那么消费方应该实现业务接口的幂等性设计。

（五）、本地消息服务方案

• 实现思路：
  

1. 主动方应用系统通过业务操作完成业务数据的操作，在准备发送消息的时候将消息存储在主动方应用系统一份，另一份发送到实时消息服务
   
2. 被动方应用系统监听实时消息系统中的消息，当被动方完成消息处理后通过调用主动方接口完成消息确认
   
3. 主动方接收到消息确认以后删除消息数据。
   
4. 通过消息查询服务查询到消息被接收之后再规定的时间内没有返回ACK确认消息就通过消息恢复系统重新发送消息。
   

• 优点：
  

1. 消息的时效性比较高
   
2. 从应用设计的角度实现了消息数据的可靠性，消息数据的可靠性不依赖于MQ中间件，弱化了对MQ中间件特性的依赖。
   
3. 方案轻量级，容易实现。
   

• 缺点：
  

1. 与具体的业务场景绑定，耦合性强，不可以共用。
   
2. 消息数据与业务数据同步，占用业务系统资源。
   
3. 业务系统在使用关系型数据库的情况下消息服务性能会受到关系型数据库的并发性能限制。

（六）、独立消息服务方案

• 实现思路：
  

1. 预发送消息：主动方应用系统预发送消息，由消息服务子系统存储消息，如果存储失败，那么也就无法进行业务操作。如果返回存储成功，然后执行业务操作。
   
2. 执行业务操作：执行业务操作如果成功的时候，将业务操作执行成功的状态发送到消息服务子系统。消息服务子系统修改消息的标识为“可发送”状态。
   
3. 发送消息到实时消息服务：当消息的状态发生改变的时候，立刻将消息发送到实时消息服务中。接下来，消息将会被消息业务的消费端监听到，然后被消费。
   
4. 消息状态子系统：相当于定时任务系统，在消息服务子系统中定时查找确认超时的消息，在主动方应用系统中也去定时查找没有处理成功的任务，进行相应的处理。
   
5. 消息消费：当消息被消费的时候，向实时消息服务发送ACK，然后实时消息服务删除消息。同时调用消息服务子系统修改消息为“被消费”状态。
   
6. 消息恢复子系统：当消费方返回消息的时候，由于网络中断等其他原因导致消息没有及时确认，那么需要消息恢复子系统定时查找出在消息服务子系统中没有确认的消息。将没有被确认的消息放到实时消息服务中，进行重做，因为被动方应用系统的接口是幂等的。
   

• 优点：
  

1. 消息服务独立部署，独立维护，独立伸缩。
   
2. 消息存储可以按需选择不同的数据库来集成实现。
   
3. 消息服务可以被相同的的使用场景使用，降低重复建设服务的成本。
   
4. 从分布式服务应用设计开发角度实现了消息数据的可靠性，消息数据的可靠性不依赖于MQ中间件，弱化了对MQ中间件特性的依赖。
   
5. 降低了业务系统与消息系统之间的耦合，有利于系统的扩展维护。
   

• 缺点：
  

1. 一次消息发送需要两次请求。
   
2. 主动方应用系统需要实现业务操作状态的校验与查询接口。

（七）、消息服务子系统的设计实现

示例消息数据表：