小伙伴们可加入微信群讨论
本课程需具备rabbitmq基础，请先参考 《rabbitmq基础及spring-boot集成方案》

一 前言/大纲

微服务的流行带来了SLA上的极大提升，但是随之而来带来服务逻辑的复杂性，而分布式事物是微服务，分布式框架一个必须解决的难题，目前并没有统一的解决方案。

阿里《破解世界性技术难题！ GTS让分布式事务简单高效》一文，分布式系统中(SOA，微服务)中，分布式事务一致性是一个无法绕开的挑战。一个看起来简单的功能，内部可能需要调用多个“服务”并操作多个数据库或分片来实现，服务提供者分属与不同节点容器，一旦涉及到多系统，高并发，结果往往不可预料: 网络抖动怎么办？服务器突然挂了怎么办？第二阶段失败了要不要第一阶段回滚？万一回滚失败了呢？异常监控要如何来做？本讲座会介绍分布式事务通常的解决方案2PC(两阶段提交)，阿里RocketMQ解决方案，以及使用RabbitMQ实现消息最终一致性方案，包括项目源码的讲解。本讲座包括:

1.1 准备和学习

1. 什么是分布式事务
2. 目前的解决方案
3. 基于消息队列的解决方案

1.2 实战

基于RabbitMQ消息最终一致性解决方案(CoolMQ)原理及应用

1.3 答疑

回答大家问题

1.4 参考

RabbitMQ基础知识可阅读 http://rabbitmq.org.cn 上技术文档

二 什么是分布式事务

本地事务想必大家都了解。但什么是分布式事务呢？比如你在淘宝买东西，会有两个动作：一个是扣库存，第二个是更新订单状态。而在分布式系统中，比如说微服务，库存和订单是两个服务，那么假设第一步扣库存成功，而后发送请求到第二步时网络请求失败了，但是第一步无从感知，没法回滚，就会导致事务不一致了。

三 目前的解决方案

目前解决方案最常见的的有XA,2PC,3PC等，比如2阶段提交(2PC):

当一个事务跨越多个节点时，为了保持事务的ACID特性，需要引入一个作为协调者的组件来统一掌控所有节点(称作参与者)的操作结果并最终指示这些节点是否要把操作结果进行真正的提交(比如将更新后的数据写入磁盘等等)。因此，二阶段提交的算法思路可以概括为：参与者将操作成败通知协调者，再由协调者根据所有参与者的反馈情报决定各参与者是否要提交操作还是中止操作。

以开会为例：
甲乙丙丁四人要组织一个会议，需要确定会议时间，不妨设甲是协调者，乙丙丁是参与者。
投票阶段：
（1）甲发邮件给乙丙丁，周二十点开会是否有时间；
（2）甲回复有时间；
（3）乙回复有时间；
（4）丙迟迟不回复，此时对于这个活动，甲乙丙均处于阻塞状态，算法无法继续进行；
（5）丙回复有时间（或者没有时间）；
提交阶段：
（1）协调者甲将收集到的结果反馈给乙丙丁（什么时候反馈，以及反馈结果如何，在此例中取决与丙的时间与决定）；
（2）乙收到；
（3）丙收到；
（4）丁收到；
不仅要锁住参与者的所有资源，而且要锁住协调者资源，开销大。一句话总结就是：2PC效率很低，分布式事务很难做。

四 基于消息队列的解决方案

RocketMQ

目前基于消息队列的解决方案有阿里的RocketMQ，它提出了”半消息“,具体流程如下

第一阶段：上游应用执行业务并发送 MQ 消息

1. 上游应用发送待确认消息到可靠消息系统
2. 可靠消息系统保存待确认消息并返回
3. 上游应用执行本地业务
4. 上游应用通知可靠消息系统确认业务已执行并发送消息。
5. 可靠消息系统修改消息状态为发送状态并将消息投递到 MQ 中间件

第二阶段：下游应用监听 MQ 消息并执行业务
下游应用监听 MQ 消息并执行业务，并且将消息的消费结果通知可靠消息服务。

可靠消息的状态需要和下游应用的业务执行保持一致，可靠消息状态不是已完成时，确保下游应用未执行，可靠消息状态是已完成时，确保下游应用已执行。

下游应用和可靠消息服务之间的交互图如下：

1. 下游应用监听 MQ 消息组件并获取消息
2. 下游应用根据 MQ 消息体信息处理本地业务
3. 下游应用向 MQ 组件自动发送 ACK 确认消息被消费
4. 下游应用通知可靠消息系统消息被成功消费，可靠消息将该消息状态更改为已完成

这是阿里的实现方案，如果有兴趣的话可以去了解

RabbitMQ实现

我们在rabbitmq上肉身实战了一下可靠消息，rabbitmq的发送过程如下

1. 生产者发送消息到消息服务
2. 如果消息落地，则返回一个标志给消息发起端。
3. 消息队列将消息发送给消费者
4. 消息监听接受并处理消息，如果处理成功则手动确认。

我们来看看可能发送异常的四种

1 直接无法到达消息服务

网络断了，抛出异常，业务直接回滚即可。如果出现connection closed错误，直接增加 connection数即可

connectionFactory.setChannelCacheSize(100);

2 消息已经到达服务器，但返回的时候出现异常

rabbitmq提供了确认ack机制，可以用来确认消息是否有返回。因此我们可以在发送前在db中(内存或关系型数据库)先存一下消息，如果ack异常则进行重发

/**confirmcallback用来确认消息是否有送达消息队列*/     
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
    if (!ack) {
        //try to resend msg
    } else {
        //delete msg in db
    }
});
 /**若消息找不到对应的Exchange会先触发returncallback */
rabbitTemplate.setReturnCallback((message, replyCode, replyText, tmpExchange, tmpRoutingKey) -> {
    try {
        Thread.sleep(Constants.ONE_SECOND);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

    log.info("send message failed: " + replyCode + " " + replyText);
    rabbitTemplate.send(message);
});

如果消息没有到exchange,则confirm回调,ack=false
如果消息到达exchange,则confirm回调,ack=true
但如果是找不到exchange，则会先触发returncallback

3 消息送达后，消息服务自己挂了

如果设置了消息持久化，那么ack= true是在消息持久化完成后，就是存到硬盘上之后再发送的，确保消息已经存在硬盘上，万一消息服务挂了，消息服务恢复是能够再重发消息

4 未送达消费者

消息服务收到消息后，消息会处于"UNACK"的状态，直到客户端确认消息

channel.basicQos(1); // accept only one unack-ed message at a time (see below)
final Consumer consumer = new DefaultConsumer(channel) {
  @Override
  public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    String message = new String(body, "UTF-8");

    System.out.println(" [x] Received '" + message + "'");
    try {
      doWork(message);
    } finally {
       //确认收到消息
      channel.basicAck(envelope.getDeliveryTag(), false);
    }
  }
};
boolean autoAck = false;
channel.basicConsume(TASK_QUEUE_NAME, autoAck, consumer);

5 确认消息丢失

消息返回时假设确认消息丢失了，那么消息服务会重发消息。注意，如果你设置了autoAck= false，但又没应答 channel.baskAck也没有应答 channel.baskNack，那么会导致非常严重的错误：消息队列会被堵塞住，所以，无论如何都必须应答

6 消费者业务处理异常

消息监听接受消息并处理，假设抛异常了，第一阶段事物已经完成，如果要配置回滚则过于麻烦，即使做事务补偿也可能事务补偿失效的情况，所以这里可以做一个重复执行，比如guava的retry，设置一个指数时间来循环执行，如果n次后依然失败，发邮件、短信，用人肉来兜底。