版本: 5.0

事务消息

事务消息是 Apache RocketMQ 提供的一种高级消息类型。本文介绍了事务消息的应用场景、工作机制、限制、使用方法以及使用注意事项。

适用场景

分布式事务

在分布式系统中执行核心业务逻辑时，通常需要同时调用多个下游业务来处理相关逻辑。因此，确保核心业务与下游业务之间执行结果的一致性是分布式事务需要解决的最大挑战。事务消息诉求

在电商场景中，当用户下单时，会触发下游系统进行相应的变更。例如，物流系统必须启动发货，积分系统必须更新用户的积分，购物车系统必须清空用户的购物车。处理分支包括：

订单系统将订单状态从“未支付”变更为“已支付”。
物流系统添加待发货记录并创建订单物流记录。
积分系统更新用户的积分。
购物车系统清空购物车并更新用户的购物车记录。

传统的基于 XA 的方案：性能较差

确保分支间结果一致性的典型方法是使用基于 XA（eXtended Architecture）协议的分布式事务系统。该系统将四个调用分支封装为一个包含四个独立事务分支的大事务。虽然此解决方案可以确保结果的一致性，但需要锁定大量资源才能实现。锁定的资源数量随着分支数量的增加而增加，导致系统并发度降低。随着下游分支的增加，系统性能会下降。

基于普通消息的方案：结果一致性较差

一种基于 XA 方案的简化方案是将订单系统的变更视为本地事务，将下游系统的变更视为下游任务。事务分支被视为带有附加订单表事务的普通消息。该方案通过异步处理消息来缩短处理生命周期，并提高了系统并发度。普通消息方案

然而，该方案容易导致核心事务和事务分支之间的结果不一致，例如：

消息已发送，但订单未执行。结果，整个事务需要回滚。
订单已执行，但消息未发送。在这种情况下，必须重新发送消息以供消费。
超时错误无法可靠地检测到，这使得难以确定订单是需要回滚还是需要提交订单变更。

基于 Apache RocketMQ 分布式事务消息的方案：彻底的一致性

上述方案无法保证一致性的原因是，普通消息不具备独立数据库事务的提交、回滚和统一协调能力。

Apache RocketMQ 的事务消息特性在普通消息方案的基础上支持两阶段提交。该特性结合了两阶段提交和本地事务，以实现提交结果的全局一致性。

Apache RocketMQ 的事务消息方案功能强大、可扩展且易于开发。有关事务消息工作机制和流程的更多信息，请参阅“工作机制”。

工作原理

定义

事务消息是 Apache RocketMQ 提供的一种高级消息类型，旨在确保消息生产与本地事务之间的最终一致性。处理流程

下图显示了事务消息的交互过程。

生产者向 Apache RocketMQ Broker 发送消息。
Apache RocketMQ Broker 保存消息并将其标记为不可投递。处于此状态的消息称为“半消息”。随后，Broker 向生产者返回确认消息（ACK）。
生产者执行本地事务。
生产者向 Broker 发送第二个 ACK，以提交本地事务的执行结果。执行结果可以是 Commit（提交）或 Rollback（回滚）。
- 如果 Broker 接收到的消息状态为 Commit，则 Broker 将半消息标记为可投递，并将消息投递给消费者。
- 如果 Broker 接收到的消息状态为 Rollback，则 Broker 将回滚事务，并且不会将半消息投递给消费者。
如果网络断开或生产者应用程序重启，导致 Broker 未收到第二个 ACK，或者半消息的状态为 Unknown，则 Broker 会等待一段时间，并向生产者集群中的某个生产者发送请求，查询半消息的状态。注意：有关周期时长和最大查询次数的更多信息，请参阅参数限制。

生产者收到请求后，检查与该半消息对应的本地事务的执行结果。
生产者根据本地事务的执行结果向 Apache RocketMQ Broker 发送另一个 ACK。然后，Broker 按照步骤 4 处理该半消息。

事务消息的生命周期

初始化：生产者构建并初始化消息，准备发送至 Broker。
事务待处理（Transaction pending）：半消息被发送到 Broker。但是，它不会立即写入磁盘进行永久存储。相反，它存储在事务存储系统中。在系统验证本地事务的第二阶段成功之前，该消息不会被提交。在此期间，下游消费者无法看到该消息。
回滚（Rollback）：在第二阶段，如果事务的执行结果为回滚，则 Broker 会回滚半消息并终止工作流程。
就绪（Ready）：消息已发送至 Broker，对消费者可见并可消费。
处理中（Inflight）：消息被消费者获取，并按照本地业务逻辑进行处理。
在此期间，Broker 等待消费者提交结果。若超时未收到反馈，Apache RocketMQ 会对该消息进行重试。详情请参考消费重试。
确认（Acked）：消费者完成消费并提交结果，Broker 标记消息消费状态。
默认情况下，Apache RocketMQ 保留所有消息。消费确认后仅进行逻辑标记，不会立即物理删除。因此，在消息过期或空间不足被清理前，消费者可以回溯消费。
删除：当消息过期或存储空间不足时，Apache RocketMQ 会按滚动方式删除最早的物理文件。详情请参考消息存储与清理。

使用限制

消息类型一致性

事务消息只能用于 MessageType 为 Transaction 的主题（Topic）。

以事务为中心的消费

Apache RocketMQ 的事务消息特性保证了本地核心事务与下游分支之间可以处理同一个事务。但是，它无法保证消息消费结果与上游执行结果之间的一致性。因此，下游业务必须确保消息得到正确处理。我们建议消费者正确使用消费重试，以确保在发生故障时消息能够被正确处理。

中间状态可见性

Apache RocketMQ 的事务消息特性仅保证最终一致性，这意味着在消息投递给消费者之前，上游事务与下游分支之间的状态不保证一致。因此，事务消息仅适用于接受异步执行的事务场景。

事务超时机制

Apache RocketMQ 为事务消息实现了超时机制。在收到半消息后，如果 Broker 在一段时间后无法确定是提交还是回滚事务，则默认回滚该消息。有关超时周期的更多信息，请参阅参数限制。

示例

创建主题（Topic）

在 Apache RocketMQ 5.0 中，建议使用 mqadmin 工具创建主题。注意，需要添加消息类型作为属性参数。以下是示例：

sh mqadmin updateTopic -n <nameserver_address> -t <topic_name> -c <cluster_name> -a +message.type=Transaction

发送消息

发送事务消息与发送普通消息在以下方面有所不同：

在发送事务消息之前，必须启用事务检查器并将其与本地事务执行相关联。
创建生产者时，必须设置事务检查器并绑定要发送消息的主题列表。这些操作使客户端内置的事务检查器能够在发生异常时恢复主题。

创建 TRANSACTION 主题

NORMAL 主题不支持投递 TRANSACTION 消息，如果您向 NORMAL 主题发送 TRANSACTION 消息，将会报错。

./bin/mqadmin updatetopic -n localhost:9876 -t TestTopic -c DefaultCluster -a +message.type=TRANSACTION

-c 集群名称
-t 主题名称
-n nameserver 地址
-a 额外属性：我们添加了一个值为 TRANSACTION 的 message.type 属性，以支持投递 TRANSACTION 消息。

以下示例以 Java 为例，展示了如何发送事务消息：

    // The demo is used to simulate the order table query service to check whether the order transaction is submitted. 
    private static boolean checkOrderById(String orderId) {
        return true;
    }
    // The demo is used to simulate the execution result of a local transaction. 
    private static boolean doLocalTransaction() {
        return true;
    }
    public static void main(String[] args) throws ClientException {
        ClientServiceProvider provider = new ClientServiceProvider();
        MessageBuilder messageBuilder = new MessageBuilder();
        // Build a transaction producer: The transactional message requires the producer to build a transaction checker to check the intermediate status of an exceptional half message. 
        Producer producer = provider.newProducerBuilder()
                .setTransactionChecker(messageView -> {
                    /**
                     * The transaction checker checks whether the local transaction is correctly committed or rolled back based on the business ID, for example, an order ID. 
                     * If this order is found in the order table, the order insertion action is committed correctly by the local transaction. If no order is found in the order table, the local transaction has been rolled back. 
                     */
                    final String orderId = messageView.getProperties().get("OrderId");
                    if (Strings.isNullOrEmpty(orderId)) {
                        // Message error. Rollback is returned. 
                        return TransactionResolution.ROLLBACK;
                    }
                    return checkOrderById(orderId) ? TransactionResolution.COMMIT : TransactionResolution.ROLLBACK;
                })
                .build();
        // Create a transaction branch. 
        final Transaction transaction;
        try {
            transaction = producer.beginTransaction();
        } catch (ClientException e) {
            e.printStackTrace();
            // If the transaction branch fails to be created, the transaction is terminated. 
            return;
        }
        Message message = messageBuilder.setTopic("topic")
                // Specify the message index key so that the system can use a keyword to accurately locate the message. 
                .setKeys("messageKey")
                // Specify the message tag so that consumers can use the tag to filter the message. 
                .setTag("messageTag")
                // For transactional messages, a unique ID associated with the local transaction is created to verify the query of the local transaction status. 
                .addProperty("OrderId", "xxx")
                // Message body. 
                .setBody("messageBody".getBytes())
                .build();
        // Send a half message.
        final SendReceipt sendReceipt;
        try {
            sendReceipt = producer.send(message, transaction);
        } catch (ClientException e) {
            // If the half message fails to be sent, the transaction can be terminated and the message is rolled back. 
            return;
        }
        /**
         * Execute the local transaction and check the execution result. 
         * 1. If the result is Commit, deliver the message. 
         * 2. If the result is Rollback, roll back the message. 
         * 3. If an unknown exception occurs, no action is performed until a response is obtained from a half message status query. 
         *
         */
        boolean localTransactionOk = doLocalTransaction();
        if (localTransactionOk) {
            try {
                transaction.commit();
            } catch (ClientException e) {
                // You can determine whether to retry the message based on your business requirements. If you do not want to retry the message, you can use the half message status query to submit the transaction status. 
                e.printStackTrace();
            }
        } else {
            try {
                transaction.rollback();
            } catch (ClientException e) {
                // We recommend that you record the exception information. This enables you to submit the transaction status based on the half message status query in the event of a rollback exception. Otherwise, you have to retry the message. 
                e.printStackTrace();
            }
        }
    }

使用说明

避免大量半消息超时。

Apache RocketMQ 允许您在事务提交过程中发生异常时检查事务，以确保事务一致性。但是，生产者应尽量避免本地事务返回未知结果。大量的事务检查会导致系统性能下降并延迟事务处理。

正确处理正在进行中的事务。

在半消息状态查询期间，对于正在进行中的事务，不要返回 Rollback 或 Commit。相反，应保持事务的 Unknown 状态。

通常，事务处于进行中状态的原因是事务执行缓慢且查询频繁。建议采用以下两种解决方案：

将第一次查询的间隔设置为较大的值。但是，这可能会导致依赖查询结果的消息出现较大的延迟。
使程序能够正确识别正在进行中的事务。

事务消息

适用场景​

工作原理​

使用限制​

示例​

使用说明​

适用场景

工作原理

使用限制

示例

使用说明