浅析分布式业务一致性方案

欢迎大家关注公众号「JAVA前线」查看更多精彩分享文章，主要包括源码分析、实际应用、架构思维、职场分享、产品思考等等，同时欢迎大家加我微信「java_front」一起交流学习

1 场景分析

现在有一种业务场景：A作为消息发送方，处理业务成功后，投递消息。B作为消息接收方，接收消息，处理业务。在这种业务场景中，我们希望A业务处理成功后，B业务也处理成功。这种抽象场景可以体现在具体业务场景：

下单：用户支付成功，订单中心投递消息，物流中心发货
报名：用户报名成功，活动中心投递消息，下游准备物料
买券：用户支付成功，订单中心投递消息，营销中心发券

A和B作为各自独立系统，如果想要保证业务一致性，并不像单体应用那么简单。我认为从三个维度考虑，而不是单点思考这个问题：

业务发送方
业务消费方
监控

2 业务发送方

业务发送方需要保证一件事情：当本业务处理成功后，一定可以投递成功业务消息。通常有两个方案：本地消息表、事务消息。

2.1 本地消息表

本地消息表思想是在业务表的同一个库中，引入消息表，通过数据库本地事务保证业务表和消息表操作强一致性。使用步骤：

第一步：业务表和消息表强一致更新，消息状态为【待发送】
第二步：发送消息至消息队列，修改为消息状态为【已发送】或【发送失败】
第三步：定时任务查询X时间前【待发送】和【发送失败】消息，重新推送至消息队列

2.2 事务消息

RocketMQ事务消息特性可以满足本文业务场景，事务消息原理如下图：

06 rocketmq事务消息原理.jpg

事务消息使用步骤：

第一步：业务方发送半消息至RocketMQ
第二步：RocketMQ返回半消息发送成功结果
第三步：执行业务代码
第四步：业务执行成功，则提交半消息至RocketMQ，此时消息才算真正提交成功。业务执行失败，回滚半消息
第五步：如果业务执行完成后，由于各种原因（例如网络原因）未返回结果，导致半消息无法确定提交还是回滚
第六步：业务需要提供查询接口，RocketMQ回调这个接口，根据结果决定提交还是回滚半消息

3 业务接收方

业务接收方需要注意以下几个维度：

维度一：接收到消息，如果处理成功需要告知RocketMQ消息处理成功，避免重复消息
维度二：接收到消息，如果处理失败需要告知RocketMQ消息处理失败，等待重试消费
维度三：因为RocketMQ重试机制存在，消息可能会被重复消费，所以必须做业务幂等

3.1 维度一

接收到消息后进行业务处理，如果处理成功则告知RocketMQ成功：

public class MessageListenerImpl implements MessageListener {

    @Override
    public Action consume(Message message, ConsumeContext context) {
        boolean result = doBusiness(message);
        if(result) {
            // 业务处理成功
            return Action.CommitMessage;
        } else {
            return Action.ReconsumeLater;
        }
    }
}

3.2 维度二

3.2.1 代码编写

接收到消息后进行业务处理，如果处理则告诉RocketMQ当前消息消费失败，并希望在稍后重新尝试消费：

public class MessageListenerImpl implements MessageListener {

    @Override
    public Action consume(Message message, ConsumeContext context) {
        boolean result = doBusiness(message);
        if(result) {
            return Action.CommitMessage;
        } else {
            // 业务处理失败
            return Action.ReconsumeLater;
        }
    }
}

消费者有三种方式告知RocketMQ消费失败，均会触发重试机制：

public class MessageListenerImpl implements MessageListener {

    @Override
    public Action consume(Message message, ConsumeContext context) {
        boolean result = doBusiness(message);
        if(result) {
            return Action.CommitMessage;
        } else {
            // 方式一
            return Action.ReconsumeLater;
            // 方式二
            throw new RuntimeException("doBusiness fail");
            // 方式三
            return null;
        }
    }
}

3.2.2 重试机制

(1) 顺序消息

如果消费者在处理过程中失败，RocketMQ 消息队列会自动触发重试机制，每隔1秒尝试重新投递该消息。在此期间由于重试机制的运行，应用可能会遇到消息消费被暂时阻塞的现象。

(2) 无序消息

无序消息类型
- 普通消息
- 定时消息
- 延时消息
- 事务消息
重试次数
- 默认16次
- 自定义重试次数，超过16次后重试间隔均为2小时
重试次数：与上一次时间间隔
- 第1次：10秒
- 第2次：30秒
- 第3次：1分钟
- 第4次：2分钟
- 第5次：3分钟
- 第6次：4分钟
- 第7次：5分钟
- 第8次：6分钟
- 第9次：7分钟
- 第10次：8分钟
- 第11次：9分钟
- 第12次：10分钟
- 第13次：20分钟
- 第14次：30分钟
- 第15次：1小时
- 第16次：2小时
无序消息重试功能只有在集群消费模式下有效。广播模式下消费失败，失败消息将不会被重试，系统会继续消费下一条新消息