故障现场 | 消息发送居然有这么大的坑_Java教程

1. 问题&分析

基于 MQ 进行系统间的解耦真的是太香了，小艾还沉浸在喜悦中久久不能自拔。但，打脸的事已经在路上了。。。。

1.1. 案例

昨天下班，在电梯里，物流组的晨姐偶遇了小艾，就一个技术问题向小艾进行了反馈。具体来说，物流系统有一项功能，即实时监控订单的支付成功事件，一旦检测到，便会为顾客准备物资，进而安排快递发货。今天系统出现了几次空指针异常。查阅日志，似乎是在反查订单信息时，没有获取到预期的订单数据。但查询物流系统，物流单已经成功生成，对业务操作并未造成实际影响，但这个问题还是值得注意。由于这个问题并没有立即影响到业务流程，所以晨姐没有在第一时间联系小艾进行确认。

在小艾正准备启动IDEA寻找线索的时候，算法组的负责人龙哥急匆匆地走了过来，向小艾反映了他们团队遇到的一个重要问题。为了提升推荐效果，算法组也会实时监控订单支付成功事件，并以此为依据重新计算用户的推荐商品。然而，今天早上，他们突然收到了一系列的报警信息，问题同样是无法查询到订单信息，这个现象与物流系统的问题高度相似。

小艾随口问道：“这个问题会自己修复吗？”龙哥愣了一下，回答说：“以前会自动修复，但刚刚那条数据还在报错。”随后，龙哥提供了报错的订单ID，小艾立即去数据库中查询，却惊讶地发现，这条数据竟然不存在！

看到这种场景，小艾有些慌神，连龙哥什么时候走的都没有注意到。目光直勾勾的盯着电脑屏幕发呆：

@Transactional
public void paySuccess(String orderId, String token){
    // 验证 token，保障有效性
    checkToke(token);

    // 加载订单信息
    Order order = this.orderRepository.getById(orderId);
    if (order == null){
        throw new RuntimeException("订单不存在");
    }
    // 支付成功，更新订单状态
    order.paySuccess();

    // 将变更更新到数据库
    this.orderRepository.update(order);

    // 发送支付成功事件
    this.eventPublisher.publishEvent(new OrderPaidEvent(order));
    // 执行其他业务逻辑
    doSomething();
}

// 监听变更，发布 MQ
@EventListener
public void handle(OrderPaidEvent event){
    rocketMQTemplate.convertAndSend("order_event", event);
}

1.2. 问题分析

两个问题看起来一样，但又有区别。当下游在收到 MQ 消息时

无法查不到订单，但稍后能自我修复
一直查不到订单，数据库里还没有，无法进行自我修复

小艾无意间看到 paySuccess 方法上的 @Transactional 顿时茅塞顿开。

故障现场 | 消息发送居然有这么大的坑图片

正如上图所示：

在数据库更新完成后，系统会立即发送消息队列（MQ）消息，同时主流程会继续执行后续的耗时操作。
当下游接收到MQ消息时，会进行数据查询。然而，由于此时主流程尚未完成事务提交，因此无法查询到相关数据，导致下游出现错误。
如果MQ消息消费失败，系统会自动进行重试。如果在此期间主流程已经完成了事务提交，那么就能够成功查询到数据，从而使得业务流程得以恢复正常。

这就完美的解释了物流问题，那为什么算法组收到消息里的订单ID在数据库不存在呢？

故障现场 | 消息发送居然有这么大的坑图片

如上图所示：

在数据库更新完成后，系统会立即发送消息队列（MQ）消息，而主流程将同时继续执行后续的耗时操作。
若主流程在执行后续逻辑时发生异常，将导致整个事务回滚，进而中断处理过程。
下游系统接收到消息后，进行数据反查，但由于事务已回滚，因此无法查询到任何数据。
因为发生事务回滚，数据库中根本就没有这条记录，所以即使后面有自动重试机制，也无法恢复处理逻辑。

小艾终于锁定了问题所在，深深地吸了一口气，释放了紧绷的神经。就在这时，晨姐的电话打了进来。小艾喃喃自语：“毫无疑问，和算法部门遇到的情况一样，被XXX订单给堵住了。”说罢，她信心满满地接起了电话…

本质：该问题根本原因是==没有保障更新数据库操作与发送消息操作这两个业务单元之间的一致性。==

2. 解决方案

定位后，解决方案就变的非常清晰。

2.1. 方案1：使用 @TransactionalEventListener

最简的方案就是将 @EventListener 注解换成 @TransactionalEventListener。

2.1.1. EventListener 和 TransactionalEventListener

EventListener 和 TransactionalEventListener 都是 Spring 中用于处理事件的监听器。它们之间的主要区别在于它们处理事件的方式和事务管理。

EventListener：这是一个通用的事件监听器，当事件发布时，它会立即执行相应的处理方法。它不会参与到事务管理中，也就是说，即使在事务执行过程中发生异常，EventListener 依然会执行。
TransactionalEventListener：这是一个具有事务管理功能的事件监听器。当事件发布时，它会等待当前事务完成后再执行相应的处理方法。这意味着，如果在事务执行过程中发生异常，TransactionalEventListener 将不会执行，从而确保事务的一致性。

总之，EventListener 和 TransactionalEventListener 的主要区别在于它们处理事件的方式和事务管理。在选择使用哪种监听器时，需要根据实际需求和事务一致性的要求来决定。

2.1.2. 源码示例

了解两者的区别后，只需做一点调整便可以解决这个问题，调整如下：

/**
 * 使用 @TransactionalEventListener 替代 @EventListener 监听订单支付事件，然后发送消息到 RocketMQ
 * @param event
 */
@TransactionalEventListener
public void handle(OrderPaidEvent event){
    rocketMQTemplate.convertAndSend("order_event", event);
}

如果没有使用 Spring 的 Event 机制，但仍想实现 @TransactionEventlistner 的效果，可以直接使用 Spring API：

private void doIfCommitted(Runnable task) {
        if (TransactionSynchronizationManager.isSynchronizationActive()) {
            TransactionSynchronization transactionSynchronization = new TransactionSynchronizationAdapter(){
                @Override
                public void afterCommit() {
                    task.run();
                }
            };
            TransactionSynchronizationManager.registerSynchronization(transactionSynchronization);

        }else {
            task.run();
        }
    }

2.1.3. 问题&挑战

这个方案确实解决了上述问题，但从一致性角度分析，还是存在设计缺陷，只是发生的概率变低而已，没有从根本上解决问题。

在事务提交后发送 MQ 时，可能会遇到以下几种情况，导致两个操作（数据库操作和 MQ 发送操作）之间的一致性问题：

数据库事务提交成功，但在发送 MQ 消息时发生网络故障。此时，数据库操作已经完成，但 MQ 消息未能成功发送。
数据库事务提交成功，但在发送 MQ 消息时发生 MQ 服务器故障。此时，数据库操作已经完成，但 MQ 消息未能成功发送。
数据库事务提交成功，但在发送 MQ 消息时发生应用程序故障。此时，数据库操作已经完成，但 MQ 消息未能成功发送。
数据库事务提交成功，但在发送 MQ 消息时发生消息丢失。此时，数据库操作已经完成，但 MQ 消息未能成功发送。

这个方案极为简单，但大幅降低了错误概率，主要应用于要求并不严格的业务场景。

2.2 方案2：RocketMQ事务消息

RocketMQ 的事务消息就是针对这个问题设计的，可以非常高效的解决这个问题。

2.2.1. 半消息以及工作原理

RocketMQ事务消息是一种支持分布式事务的消息模型，将消息生产和消费与业务逻辑绑定在一起，确保消息发送和事务执行的原子性，保证消息的可靠性。

事务消息分为两个阶段：发送消息和确认消息，确认消息分为提交和回滚两个操作。在提交操作执行完毕后，消息才会被消费端消费，而在回滚操作执行完毕后，消息会被删除，从而达到了事务的一致性和可靠性。

事务消息的发生流程如下：

故障现场 | 消息发送居然有这么大的坑图片

生产者发送prepare消息到RocketMQ服务端，RocketMQ将消息存储到本地并返回结果；
生产者开始执行本地事务，并根据本地事务的结果将状态信息提交给RocketMQ服务端；
如果本地事务执行成功，生产者向RocketMQ服务端发送commit消息；
如果本地事务执行失败，生产者向RocketMQ服务端发送rollback消息；
RocketMQ接收到commit或rollback消息后，对消息进行投放或删除；

如果生成者发送 prepare 消息后，未在规定时间内发送 commit 或 rollback 消息，RocketMQ 将进入恢复流程，具体如下：

故障现场 | 消息发送居然有这么大的坑图片

如果在回查的时间之前没有收到相应的 commit 或 rollback 消息，则 RocketMQ 会将对该 prepare 消息进行回查；
应用程序接收到回查指令，从业务库中获取数据，并根据业务逻辑进行判断，最终是 commit 还是 rollback；
RocketMQ 接收到 commit 或 rollback 回复后，进行相应动作，从而实现业务操作和消息发送的一致性；

2.2.2. 源码示例

一个简单的示例代码如下：

// 编写事务监听器类
public class TransactionListenerImpl implements TransactionListener {
    private AtomicInteger transactionIndex = new AtomicInteger(0);

    // 执行本地事务
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        int value = transactionIndex.getAndIncrement();
        System.out.println("executeLocalTransaction " + value);
        // TODO 执行本地事务，并返回事务状态
        // 本例假定 index 为偶数的消息执行成功，奇数的消息执行失败
        if (value % 2 == 0) {
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        return LocalTransactionState.ROLLBACK_MESSAGE;
    }

    // 检查本地事务状态
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        System.out.println("checkLocalTransaction " + msg.getTransactionId());
        // 模拟检查本地事务状态，返回事务状态
        boolean committed = prepare(true);
        if (committed) {
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        return LocalTransactionState.UNKNOW;
    }

    // 模拟操作预处理逻辑
    private boolean prepare(boolean commit) {
        System.out.println("prepare " + (commit ? "commit" : "rollback"));
        return commit;
    }

}

// 编写发送消息的代码
public class Producer {
    private static final String NAME_SERVER_ADDR = "localhost:9876";

    public static void main(String[] args) throws Exception {
        TransactionMQProducer producer = new TransactionMQProducer("MyGroup");
        producer.setNamesrvAddr(NAME_SERVER_ADDR);
        // 注册事务监听器
        producer.setTransactionListener(new TransactionListenerImpl());
        producer.start();

        // 发送事务消息
        String[] tags = {"TagA", "TagB", "TagC"};
        for (int i = 0; i < 3; i++) {
            Message msg = new Message("TopicTest", tags[i], ("Hello RocketMQ " + i).getBytes(StandardCharsets.UTF_8));
            // 在消息发送时传递给事务监听器的参数
            SendResult sendResult = producer.sendMessageInTransaction(msg, null);
            System.out.printf("%s%n", sendResult);
        }

        // 关闭生产者
        producer.shutdown();
    }
}

单看代码很难理解，简单画了张图，具体如下：

故障现场 | 消息发送居然有这么大的坑图片

2.2.3. 问题&挑战

事务消息并不完美，存在一定的问题：

与 MQ 实现强相关，并不是每个 MQ 实现都对事务消息提供支持；
API 比较晦涩，存在一定的学习成本，同时需要对业务逻辑拆分到 Listener 中，增加理解成本；

2.3. 方案3：本地消息表

事务消息表方案是一种常用的保证消息发送与业务操作一致性的方法。该方案基于数据库事务和消息队列，将消息发送和业务操作放入同一个事务中，并将业务操作和消息发送的状态记录在数据库的消息表中，以实现消息的可靠性和幂等性。

2.3.1. 设计&核心流程

整体如下图所示：

故障现场 | 消息发送居然有这么大的坑图片

image

核心流程如下：

应用程序开启一个数据库事务，并在事务中执行业务操作和消息发送；
在事务中，将业务操作和消息发送的状态记录到消息表中；
如果业务操作执行成功，并且消息发送成功，提交事务，否则回滚事务；
定时扫描消息表，并根据消息状态重新发送未被确认的消息。如果消息发送成功，更新消息状态；否则根据重试次数更新消息状态或者丢弃消息；

通过事务消息表方案，可以保证消息的可靠性。即使在消息发送失败或应用程序崩溃的情况下，也可以通过重新发送消息将业务操作和消息发送的状态同步。同时，该方案可以避免消息重复发送和漏发的情况。

2.3.2. 功能封装

清晰的流程为复用打下了基础，lego 对其做了封装。

2.3.2.1. 环境准备

首先，需要引入 lego 相关依赖：

<dependency>
    <groupId>com.geekhalo.lego</groupId>
    <artifactId>lego-starter</artifactId>
    <version>0.1.12 以上版本</version>
</dependency>

其次，在业务数据库上新建一张表用于存储消息，示例如下：

create table test_message
(
    id           bigint auto_increment primary key,

    orderly      tinyint      not null comment '是否为顺序消息',

    topic        varchar(64)  not null comment 'MQ topic',
    sharding_key varchar(128) not null comment 'ShardingKey，用于选择不同的 partition',
    tag          varchar(128) not null comment 'Message Tag 信息',

    msg_id       varchar(64)  not null comment 'Msg ID 只有发送成功后才有数据',
    msg_key      varchar(64)  not null comment 'MSG Key，用于查询数据',
    msg          longtext     not null comment '要发送的消息',

    retry_time   tinyint      not null comment '重试次数',
    status       tinyint      not null comment '发送状态:0-初始化，1-发送成功，2-发送失败',

    create_time  datetime     not null,
    update_time  datetime     not null,

    index idx_update_time_status(update_time, status)
);

为了兼容多种MQ类型，对发送者进行了抽象，因此需要实现自己的 MessageSender。

@Component
@Getter
@Slf4j
public class TestMessageSender implements MessageSender {

    @Override
    public String send(Message message) {
        // 发送消息
    }
}

最后，就是对所有的组件进行配置，示例代码如下：

@Configuration
@Slf4j
public class LocalTableBasedReliableMessageConfiguration
        extends LocalTableBasedReliableMessageConfigurationSupport {

    @Autowired
    private DataSource dataSource;

    @Autowired
    private MessageSender messageSender;

    @Override
    protected DataSource dataSource() {
        return this.dataSource;
    }

    @Override
    protected String messageTable() {
        return "test_message";
    }

    @Override
    protected MessageSender createMessageSend() {
        return this.messageSender;
    }
}

其中，包括：

继承自 LocalTableBasedReliableMessageConfigurationSupport，由父类完成基本配置；
实现 DataSource dataSource() 方法，返回业务数据源（备注：必须与业务使用同一个数据源）
实现 String messageTable() 方法，配置本地消息表表名；
实现 MessageSender createMessageSend() 方法，返回 MessageSender 实例，执行真正的消费发送；

2.2.3.2. 具体使用

ReliableMessageSender#send 在业务方法中使用，执行可靠消息发送；

@Transactional
public void testSuccess(){
    // 业务逻辑
    Message message = buildMessage();
    // 业务逻辑
    this.reliableMessageSender.send(message);
}

除发送流程外，还需要配置补充机制。

ReliableMessageCompensator#compensate 周期性调度，对未发送或发送失败的消息进行补充；