RocketMQ（三）：消息类型（普通、定时、顺序和事务消息）—

1 普通消息

普通消息为Apache RocketMQ最基础的消息，区别于有特性的顺序消息、定时/延时消息和事务消息。本节介绍普通消息的应用场景、功能原理、使用方法和使用建议。

1.1 应用场景

普通消息一般应用于微服务解耦、事件驱动、数据集成等场景，这些场景大多数要求数据传输通道具有可靠传输的能力，且对消息的处理时机、处理顺序没有特别要求。

典型场景一：微服务异步解耦

如上图所示，以在线的电商交易场景为例，上游订单系统将用户下单支付这一业务事件封装成独立的普通消息并发送至 Apache RocketMQ服务端，下游按需从服务端订阅消息并按照本地消费逻辑处理下游任务。每个消息之间都是相互独立的，且不需要产生关联。

典型场景二：数据集成传输

如上图所示，以离线的日志收集场景为例，通过埋点组件收集前端应用的相关操作日志，并转发到Apache RocketMQ。每条消息都是一段日志数据，Apache RocketMQ不做任何处理，只需要将日志数据可靠投递到下游的存储系统和分析系统即可，后续功能由后端应用完成。

1.2 功能原理

什么是普通消息

定义：普通消息是Apache RocketMQ基本消息功能，支持生产者和消费者的异步解耦通信。

普通消息生命周期

初始化：消息被生产者构建并完成初始化，待发送到服务端的状态。

待消费：消息被发送到服务端，对消费者可见，等待消费者消费的状态。

消费中：消息被消费者获取，并按照消费者本地的业务逻辑进行处理的过程。此时服务端会等待消费者完成消费并提交消费结果，如果一定时间后没有收到消费者的响应，Apache RocketMQ会对消息进行重试处理。

消息提交：消费者完成消费处理，并向服务端提交消费结果，服务端标记当前消息已被处理（包括消费成功和失败）。Apache RocketMQ默认支持保留所有消息，此时消息数据并不会立即删除，只是逻辑标记已消费。消费在保存时间到期或存储空间不足时被删除，消费者仍然可以回溯消息重新消费。

消息删除：Apache RocketMQ按照消息保存机制滚动清理最早的消息数据，将消息从物理文件中删除。

1.3 使用限制

普通消息仅支持使用Message Type 为Normal主题，即普通消息只能发送至类型为普通消息的主题中，发送的消息的类型必须和主题的类型一样。

1.4 使用示例

public class NormalMessage {
    private Producer producer;
    private SimpleConsumer simpleConsumer;
    public static void main(String[] args) {

    }
    public void test01() throws ClientException {
        //创建生产者（仅用于测试demo）
        ProducerBuilderImpl producerBuilder = new ProducerBuilderImpl();
        producer = producerBuilder.build();
        //创建消费者（仅用于测试demo）
        SimpleConsumerBuilderImpl simpleConsumerBuilder = new SimpleConsumerBuilderImpl();
        simpleConsumer = simpleConsumerBuilder.build();
        //普通消息发送
        MessageBuilderImpl messageBuilder = new MessageBuilderImpl();
        Message message = messageBuilder.setTopic("topic")
                //设置消息索引键，可根据关键字精确查找某条消息
                .setKeys("messageKey")
                //设置消息Tag,用于消费端根据指定Tag过滤消息
                .setTag("messageTag")
                //设置消息体
                .setBody("messageBody".getBytes())
                .build();
        try {
            //发送消息，需要关注发送结果，并捕获失败等异常
            SendReceipt sendReceipt = producer.send(message);
            System.out.println(sendReceipt.getMessageId());
        }catch (ClientException e){
            e.printStackTrace();
        }
        //消费示例1：使用pushConsumer消费者普通消息，只需要在消费监听器中处理
        MessageListener messageListener = new MessageListener() {
            @Override
            public ConsumeResult consume(MessageView messageView) {
                System.out.println(messageView);
                //根据消息结果返回状态
                return ConsumeResult.SUCCESS;
            }
        };
        //消费示例2：使用SimpleConsumer消费普通消息，主动获取信息进行消费处理并提交消费结果
        List<MessageView> messageViewList = null;
        try {
            messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
            messageViewList.forEach(messageView -> {
                System.out.println(messageView);
                //消费处理完成后，需要主动调用ACK提交消费结果
                try {
                    simpleConsumer.ack(messageView);
                }catch (ClientException e){
                    e.printStackTrace();
                }
            });
        }catch (ClientException e){
            //如果遇到系统流控等原因造成拉取失败，需要重新发送获取消息请求
            e.printStackTrace();
        }
    }
}

1.5 使用建议

设置全局唯一业务索引键，方便问题追踪

Apache RocketMQ 支持自定义索引键（消息的key），在消息查询和轨迹查询时，可以通过索引键高效精确地查询到消息。

因此，发送消息时，建议设置业务上唯一的信息作为索引，方便后续快速定位消息。例如，订单ID、用于ID等。

2 定时/延时消息

定时/延时消息为Apache RocketMQ中的高级特性消息，本节介绍定时/延时消息的应用场景、功能原理、使用限制、使用方法和使用建议。

【备注】定时消息和延时消息本质相同，都是服务端根据消息设置的定时时间在某一个固定时刻将消息投递给消费者消费。因此，下文统一用定时消息描述。

2.1 使用场景

在分布式定时调度触发、任务超时处理等场景，需要实现精准、可靠的定时事件触发。使用Apache RocketMQ的定时消息可以简化定时调度任务的开发逻辑，实现高性能、可扩展、高可靠的定时触发能力。

典型场景1：分布式定时调度

在分布式定时调度场景下，需要实现各类精度的定时任务，例如每天5点执行文件清理，每隔2分钟触发一次消息推送等需求。传统基于数据库的定时调度方案在分布式场景下，性能不高，实现复杂。基于Apache RocketMQ的定时消息可以封装出多种类型的定时触发器。

典型场景2：任务超时处理

以电商交易场景为例，订单下单后暂未支付，此时不可以直接关闭订单，而是需要等待一段时间后才能关闭订单。使用Apache RocketMQ定时消息可以实现超时任务的检查触发。

基于定时消息的超时任务处理具备如下优势：

· 精度高、开发门槛低：基于消息通知方式不存在定时阶梯间隔。可以轻松实现任意精度事件触发，无需业务去重。

· 高性能可扩展：传统的数据库扫描方式较为复杂，需要频繁调用接口扫描，容易产生性能瓶颈。Apache RocketMQ的定时消息具有高并发和水平扩展的能力。

2.2 功能原理

什么是定时消息

定时消息是Apache RocketMQ提供的一种高级消息类型，消息被发送至服务端后，在指定时间后才能被消费者消费。通过设置一定的定时时间可以实现分布式场景的延时调度触发效果。

定时时间设置原则

1、Apache RocketMQ定时消息设置的定时时间是一个预期触发的系统时间，延时时间也需要转换成当前系统时间后的某一个时间戳，而不是一段延时时长。

2、定时时间的格式为毫米级的Unix时间戳，需要将要设置的时刻转换成时间戳形式。具体方式，请参考Unix时间戳转换工具。

3、定时时间必须设置在定时时长范围内，超过范围则定时不生效，服务端会立即投递消息。

4、定时时长最大默认值为24小时，不支持自定义修改。

5、定时时间必须设置为当前时间之后，若设置当前时间之前，则定时不生效，服务端会立即投递消息。

定时消息声明周期

1、初始化：消息被生产者构建完成初始化，待发送到服务端的状态。

2、定时中：消息被发送到服务端，和普通消息不同的是，服务端不会直接构建消息索引，而是会将定时消息单独存储到在定时存储系统中，等待定时时刻到达。

3、待消费：定时时刻到达后，服务端将消息重新写入普通存储引擎，对下游消费者可见，等待消费者消费的状态。

4、消费中：消息被消费者获取，并按照消费者本地的业务逻辑进行处理的过程。此时服务端会等待消费者完成消费并提交消费结果，如果一定时间后没有收到消费者的响应， Apache RocketMQ 会对消息进行重试处理。

5、消费提交：消费者完成消费处理，并向服务端提交消费结果，服务端标记当前消息已被处理（包括处理成功和失败）。Apache RocketMQ 默认支持保留所有消息，此时消息数据并不会立即被删除，只是逻辑标识已消费。消息在保存时间到期或存储空间不足被删除前，消费者仍然可以回溯消息重新消费。

6、消息删除：Apache RocketMQ 按照消息保存机制滚动清理最早的消息数据，将消息从物理文件中删除。

2.3 使用限制

消息类型一致性

定时消息仅在支持MessageType为Delay的主题内使用，即定时消息只能发送至类型为定时消息的主题中，发送的消息的类型必须和主题的类型一致。

定时精度约束

Apache RocketMQ 定时消息的定时时长参数精度到毫秒级，但是默认精度为1000ms，即定时消息为秒级精度。

Apache RocketMQ 定时消息的状态支持持久化存储，系统由于故障重启后，仍支持按照原来设置的定时时间触发消息投递。若存储系统异常重启，可能会导致定时消息投递出现一定延迟。

2.4 使用实例

public class DelayMessage {
    private Producer producer;
    private SimpleConsumer simpleConsumer;

    public void test() throws ClientException {
        ProducerBuilderImpl producerBuilder = new ProducerBuilderImpl();
        producer = producerBuilder.build();
        SimpleConsumerBuilderImpl consumerBuilder = new SimpleConsumerBuilderImpl();
        simpleConsumer = consumerBuilder.build();
        //定时/延时发送消息
        MessageBuilder messageBuilder = new MessageBuilderImpl();
        //以下示例表示：延迟时间为10分钟之后的unix时间戳
        long deliverTimeStamp = System.currentTimeMillis() + 10L * 60 * 1000;
        Message message = messageBuilder.setTopic("topic")
                //设置消息索引键，可根据关键字精确查找某条消息
                .setKeys("messageKey")
                //设置消息Tag,用于消费端根据Tag过滤消息
                .setTag("messageTag")
                .setDeliveryTimestamp(deliverTimeStamp)
                //消息体
                .setBody("messageBody".getBytes())
                .build();
        try {
            //发送消息，需要关注发送结果，并捕获失败等异常
            SendReceipt sendReceipt = producer.send(message);
            System.out.println(sendReceipt.getMessageId());
        }catch (ClientException e){
            e.printStackTrace();
        }
        //消费示例1：使用pushConsumer消费定时消息，只需要在消费监听器处理
        MessageListener messageListener = new MessageListener() {
            @Override
            public ConsumeResult consume(MessageView messageView) {
                System.out.println(messageView.getDeliveryTimestamp());
                //根据消费结果返回状态
                return ConsumeResult.SUCCESS;
            }
        };
        //消费示例2：使用simpleConsumer消费定时消息，主动获取消息进行消费处理并提交消费 结果
        List<MessageView> messageViewList = null;
        try {
            messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
            messageViewList.forEach(messageView -> {
                System.out.println(messageView);
                //消费处理完成后，需要主动调用ACK提交消费结果
                try {
                    simpleConsumer.ack(messageView);
                }catch (ClientException e){
                    e.printStackTrace();
                }
            });
        }catch (ClientException e){
            //如果遇到系统流控等原因造成拉取失败，需要重新发送获取消息请求
            e.printStackTrace();
        }

    }
}

2.5 使用建议

避免大量相同定时时刻的消息

定时消息的实现逻辑需要先经过定时存储等待触发，定时时间到达后才会被投递给消费者。因此，如果将大量定时消息的定时时间设置为同一时刻，则到达该时刻后会有大量消息同时需要被处理，会造成系统压力过大，导致消息分发延迟，影响定时精度。

3 顺序消息

顺序消息是Apache RocketMQ中的高级特性消息，本节介绍顺序消息的应用场景、功能原理、使用限制、使用方法和使用建议。

3.1 应用场景

在有序事件处理、撮合交易、数据实时增量同步等场景下，异构系统间需要维持强一致的状态同步，上游的事件变更需要按照顺序传递到下游进行处理。在这类场景下使用Apache RocketMQ的顺序消息可以有效保证数据传输的顺序性。

典型场景1：撮合交易

以证券、股票交易撮合场景为例，对于出价相同的交易单，坚持按照先出价先交易的原则，下游处理订单的系统需要严格按照出价顺序来处理订单。

典型场景2：数据实时增量同步

普通消息

顺序消息

以数据库变更增量同步场景为例，上游源端数据库按需执行增删改查操作，将二进制操作日志作为消息，通过Apache RocketMQ传输到下游搜索系统，下游系统按照顺序还原消息数据，实现状态数据按序刷新。如果是普通消息则可能会导致状态混乱，和预期操作结果不符，基于顺序消息可以实现下游状态和上游操作结果一致。

3.2 功能原理

什么是顺序消息

顺序消息是Apache RocketMQ 提供的一种高级消息类型，支持消费者按照发送消息的先后顺序获取消息，从而实现业务场景中的顺序处理。相比其他类型消息，顺序消息在发送、存储和投递的处理过程中，更多强调多条消息间的先后顺序关系。

Apache RocketMQ 顺序消息的顺序关系通过消息组（MessageGroup）判定和识别的，发送顺序消息时需要为每条消息设置归属的消息组，相同消息组的多条消息之间遵循先进先出的顺序关系，不同消息组、无消息组的消息之间不涉及顺序性。

基于消息组的顺序判定逻辑，支持按照业务逻辑做细粒度拆分，可以在满足业务局部顺序的前提下提高系统的并行度和吞吐能力。

如何保证消息的顺序性

Apache RocketMQ 的消息的顺序性分为两部分，生产顺序性和消费顺序性。

生产顺序性：

Apache RocketMQ 通过生产者和服务端的协议保障单个生产者串行地发送消息，并按序存储和持久化。

如何保证消息生产的顺序性，必须满足以下条件：

1、单一生产者：消息生产的顺序性仅支持单一生产者，不同生产者分布在不同的系统，即使设置相同的消息组，不同生产者之间产生的消息也无法判定其先后顺序。

2、串行发送：Apache RocketMQ 生产者客户端支持多线程安全访问，但如果生产者使用多线程并行发送，则不同线程之间产生的消息将无法判定其先后顺序。

满足以上条件的生产者，将顺序消息发送至Apache RocketMQ后，会保证设置了同一个消息组的消息，按照发生顺序存储在同一队列中。服务端顺序存储逻辑如下：

1、相同消息组的消息按照先后顺序被存储在同一个队列。

2、不同消息组的消息客户混合在同一个队列中，且不保证连续。

如上图所示，消息组1和消息组4的消息混合在存储队列中，Apache RocketMQ保证消息组1中的消息G1-M1、G1-M2、G1-M3是按照发送顺序存储，且消息组4的G4-M1、G4-M2也是按顺序存储，但消息组1和消息组4中的消息不涉及顺序关系。

消费顺序性：

Apache RocketMQ通过消费者和服务端的协议保障消息严格按照存储的先后顺序处理。

如需保证消息消费的顺序性，则必须满足以下条件：

1、投递顺序

Apache RocketMQ 通过客户端SDK和服务端通信协议保障消息按照服务端存储顺序投递，但业务方消费消息时需要严格按照接收——处理——应答的语义处理消息，避免因异步处理导致消息乱序。

【备注】消费者类型为PushConsumer时，Apache RocketMQ保证消息按照存储顺序一条一条投递给消费者，若消费者类型为SimpleConsumer，则消费者有可能一次拉取多条消息。此时，消息消费的顺序性需要由业务方自行保证。

2、有限重试

Apache RocketMQ 顺序消息投递仅在重试次数限定范围内，即一条消息如果一致重试失败，超过最大重试次数后将不再重试，跳过这条消息消费，不会一直阻塞后续消息处理。

对于需要严格保证消费顺序的场景，请务必设置合理的重试次数，避免参数不合理导致消息乱序。

生产顺序和消费顺序组合

如果消息需要严格按照先进先出（FIFO）的原则处理，即先发送的先消费、后发送的后消费，则必须要同时满足生产顺序性和消费顺序性。

一般业务场景下，同一个生产者可能对接多个下游消费者，不一定所有的消费者业务都需要顺序消费，则可以将生产顺序和消费顺序进行差异化组合，应用于不同的业务场景。例如发送顺序消息，但使用非顺序的并发消费方式来提高吞吐能力。更多组合方式如下所示：

生产顺序	消费顺序	顺序效果
设置消息组，保证消息顺序发送	顺序消费	按照消息组粒度，严格保证消息顺序。同一消息组内的消息顺序和发送顺序完全一致
设置消息组，保证消息顺序发送	并发消费	并发消费，尽可能按照时间顺序处理
未设置消息组，消息乱序发送	顺序消费	按队列存储粒度，严格顺序。基于Apache RocketMQ本身队列属性，消费顺序和队列存储的顺序一致，但不保证和发送顺序一致。
未设置消息组，消息乱序发送	并发消费	尽可能按照时间顺序处理

顺序消息声明周期

初始化： 消息被生产者构建并完成初始化，待发送到服务端的状态。

待消费：消息被发送到服务端，对消费者可见，等待消费者消费的状态。

消费中：消息被消费者获取，并按照消费者本地的业务逻辑处理的过程。此时服务端会等待消费者完成消费并提交消费结果，如果一定时间后没有收到消费者的响应，Apache RocketMQ会对消息进行重试处理。

消费提交：消费者完成消费处理，并向服务端提交消费结果，服务端标记当前消息已被处理（包括消费成功和失败）。Apache RocketMQ默认支持保留所有消息，此时消息数据并不会立即删除，只是逻辑标记已消费。消息在保存时间到期或存储空间不足被删除前，消费者仍然可以回溯消费重新消费。

消息删除：Apache RocketMQ按照消息保存机制滚动清理最早的消息数据，将消息从物理文件中删除。

【备注】消息消费失败或消费超时，会触发服务端重试逻辑，重试消息属于新的消息，原消息的声明周期已结束。顺序消息消费失败进行消息重试时，为保障消息的顺序性，后续消息不可被消费，必须等待前面的消息消费完成后才能被处理。

3.3 使用限制

顺序消息仅支持使用MessageType为FIFO的主题，即顺序消息只能发送至类型为顺序消息的主题中，发送的消息的类型必须和主题类型一致。

3.4 使用示例

public class OrderMessage {
    private Producer producer;
    private SimpleConsumer simpleConsumer;
    
    public void test() throws ClientException {
        ProducerBuilderImpl producerBuilder = new ProducerBuilderImpl();
        producer = producerBuilder.build();
        SimpleConsumerBuilderImpl consumerBuilder = new SimpleConsumerBuilderImpl();
        simpleConsumer = consumerBuilder.build();
        //顺序消息发送
        MessageBuilderImpl messageBuilder = new MessageBuilderImpl();
        Message message = messageBuilder.setTopic("FIFO")
                //设置消息索引键，可根据关键字精确查找某条消息
                .setKeys("messageKey")
                //设置消息Tag,用于消费端根据Tag过滤嘻嘻
                .setTag("messageTag")
                //设置顺序消息的排序分组，该分组尽量保持离散，避免热点排序分组
                .setMessageGroup("fifoGroup001")
                .setBody("messageBody".getBytes())
                .build();
        try {
            //发送消息，需要关注发送结果，并捕获失败等异常
            SendReceipt sendReceipt = producer.send(message);
            System.out.println(sendReceipt.getMessageId());
        }catch (ClientException e){
            e.printStackTrace();
        }
        //消费顺序消息时，需要确保当前消费者分组是顺序投递模式，否则仍然按并发乱序投递
        //消费示例1：使用pushConsumer消费定时消息，只需要在消费监听器处理
        MessageListener messageListener = new MessageListener() {
            @Override
            public ConsumeResult consume(MessageView messageView) {
                System.out.println(messageView.getDeliveryTimestamp());
                //根据消费结果返回状态
                return ConsumeResult.SUCCESS;
            }
        };
        //消费示例2：使用simpleConsumer消费定时消息，主动获取消息进行消费处理并提交消费结果
        //需要注意的是，同一个MessageGroup的消息，如果前序消息没有消费完成，再次调用Receive是获取不到后续消息的
        List<MessageView> messageViewList = null;
        try {
            messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
            messageViewList.forEach(messageView -> {
                System.out.println(messageView);
                //消费处理完成后，需要主动调用ACK提交消费结果
                try {
                    simpleConsumer.ack(messageView);
                }catch (ClientException e){
                    e.printStackTrace();
                }
            });
        }catch (ClientException e){
            //如果遇到系统流控等原因造成拉取失败，需要重新发送获取消息请求
            e.printStackTrace();
        }

    }
}

3.5 使用建议

串行消费，避免批量消费导致乱序

消息消费建议串行处理，避免一次消费多条消息，否则可能出现乱序情况。

消息组尽可能打散，避免集中导致热点

Apache RocketMQ 保证相同消息组的消息存储到同一个队列中，如果不同业务场景的消息都集中在少量或一个消息组中，则这些消息存储压力都会集中到服务端的少量队列或一个队列中。容易导致性能热点，且不利于扩展。一般建议的消息组设计会采用订单ID、用户ID作为顺序参考，即同一个终端用户的消息保证顺序，不同用户的消息无需保证顺序。

因此建议将业务以消息组粒度进行拆分，例如，将订单ID、用户ID作为消息组关键字，可实现同一终端用户的消息按照顺序处理，不同用户的消息无需保证顺序。

4 事务消息

事务消息是Apache RocketMQ中的高级特性消息，本节将介绍事务消息的应用场景、功能原理、使用限制、使用方法和使用建议。

4.1 使用场景

分布式事务的诉求

分布式系统调用的特点为一个核心业务逻辑的执行，同时需要调用多个下游业务进行处理。因此，如何保证核心业务和多个下游业务的执行结果完全一致，是分布式事务需要解决的主要问题。

以电商交易场景为例，用户支付订单这一核心操作的同时会涉及到下游物流发货、积分变更、购物车状态清空等多个子系统的变更，当前业务的处理分支包括：

1、主分支订单系统状态更新：有未支付变更为支付成功。

2、物流系统状态新增：新增待发货物流记录，创建订单物流记录。

3、积分系统状态变更：变更用户积分，更新用户积分表。

4、购物车系统状态变更：清空购物车，更新用户购物车记录。

传统XA事务方案：性能不足

为了保证上述四个分支的执行结果一致性，典型方案是基于XA协议的分布式事务来实现。将四个调用分支封装成包含4个独立事务分支的大事务。基于XA分布式事务的方案可以满足业务处理结果的正确性，但最大的缺点是多分支环境下资源锁定范围大，并发度低下，随着下游分支的增加，系统性能会越来越差。

基于普通消息方案：一致性保障困难

将上述基于XA事务的方案进行简化，将订单系统变更作为本地事务，剩下的系统变更作为普通消息的下游来执行，事务分支简化成普通消息+订单表事务，充分利用消息异步化的能力缩短链路，提高并发速度。

该方案中消息下游分支和订单系统变更的主分支很容易出现不一致的线程，例如：

1、消息发送成功，订单没有执行成功，需要回滚整个事务。

2、订单执行成功，消息没有发送成功，需要额外补偿才能发现不一致。

3、消息发送超时未知，此时无法判断需要回滚订单还是提交订单变更。

基于Apache RocketMQ分布式事务消息：支持最终一致性

上述普通消息方案中，普通消息和订单事务无法保证一直的原因，本质上是由于普通消息无法像单机数据库事务一样，具备提交、回滚和统一协调的能力。

而基于Apache RocketMQ实现的分布式事务消息功能，在普通消息基础上，支持二阶段的提交能力。将二阶段提交和本地事务绑定，实现全局提交结果的一致性。

Apache RocketMQ 事务消息的方案，具备高性能、可扩展、业务开发简单的优势。

4.2 功能原理

什么是事务消息

事务消息是Apache RocketMQ提供的一种高级消息类型，支持在分布式场景下保障消息生产和本地事务的最终一致性。

事务消息处理流程

事务消息交互流程如下图所示

1、生产者将消息发送至Apache RocketMQ服务端。

2、Apache RocketMQ服务端将消息持久化成功之后，向生产者返回Ack确认消息已经发送成功，此时消息被标记为“暂不能投递”，这种状态下的消息即为半事务消息。

3、生产者开始执行本地事务逻辑。

4、生产者根据本地事务执行结果向服务端提交二次确认结果（Commit或是Rollback），服务端收到确认结果后处理逻辑如下：

（1）二次确认结果是Commit：服务端将半事务消息标记为可投递，并投递给消费者。

（2）二次确认结果是RollBack：服务端将回滚事务，不会将半事务消息投递给消费者。

5、在断网或者是生产者应用重启的特殊情况下，若服务端未收到发送者提供的二次确认结果，或服务端收到的二次确认结果是Unknown位置装填，经过固定时间后，服务端将对消息生产者即生产者集群中任一生产者实例发送消息回查。说明服务端回查的间隔时间和最大回查次数，请参见参数限制。

6、生产者收到消息回查后，需要检查对应消息的本地事务执行的最终结果。

7、生产者根据检查到本地事务的最终状态再次提交二次确认，服务端仍按照步骤4对半事务消息进行处理。

事务消息声明周期

初始化：半事务消息被生产者构建并完成初始化，待发送到服务端的状态。

事务待提交：半事务消息被发送到服务端，和普通消息不同并不会直接被服务端持久化，而是会被单独存储到事务存储系统中，等待第二阶段本地事务返回执行结果后再提交。此时消息对下游消费者不可见。

消息回滚：第二阶段如果事务执行结果明确为回滚，服务端会将半事务消息回滚，该事务消息流程终止。

提交待销费：第二阶段如果食物执行结果明确为提交，服务端会将半事务消息重新存储到普通存储系统中，此时消息对下游消费者可见，等待被消费者获取并消费。

消息删除：Apache RocketMQ按照消息保存机制滚动清理最早的消息数据，将消息从物理文件中删除。

4.3 使用限制

消息类型一致性

事务消息仅支持在MessageType为Transaction的主题内使用，即事务消息只能发送至类型为事务消息的主题中，发送的消息和类型必须和主题的类型一致。

消费事务性

Apache RocketMQ 事务消息保证本地分支事务和下游消息发送事务的一致性，但不保证消息消费结果和上游事务的一致性。因此需要下游业务分支自行保证消息正确处理，建议消费端做好消息重试，如果有短暂失败可以利用重试机制保证最终处理成功。

中间状态可见性

Apache RocketMQ 事务消息为最终一致性，即在消息提交到下游消费端处理完成之前，下游分支和上游事务之间的状态会不一致。因此，事务消息仅适合接收异步执行的事务场景。

事务超时机制

Apache RocketMQ 事务消息的声明周期存在超时机制，即半事务消息被生产者发送服务端后，如果在执行时间内服务端无法确认提交或者回滚状态，则消息默认会被回滚。

4.4 使用示例

public class TransactionMessage {
    //演示demo，模拟订单表查询服务，用来确认订单事务是否提交
    private static boolean checkOrderById(String orderId){
        return true;
    }
    //演示demo，模拟本地事务的执行结果
    private static boolean doLocalTransaction(){
        return true;
    }

    public static void main(String[] args) throws ClientException {
        ClientServiceProvider provider = new ClientServiceProviderImpl();
        MessageBuilderImpl messageBuilder = new MessageBuilderImpl();
        //构造事务生产者：事务消息需要生产者构建一个事务检查器，用来检查确认异常半事务的中间状态
        Producer producer = provider.newProducerBuilder()
                .setTransactionChecker(messageView -> {
                    /*
                        事务检查器一般是根据业务的ID去检查本地事务是否正确提交还是回滚，此处以订单ID属性为例。
                        在订单表找到了这个订单，说明本地事务查无订单的操作已经正确提交；如果订单表没有订单，说明本地事务已经回滚
                     */
                    final String orderId = messageView.getProperties().get("OrderId");
                    if (Strings.isNullOrEmpty(orderId)) {
                        //错误的消息，直接返回Rollback
                        return TransactionResolution.ROLLBACK;
                    }
                    return checkOrderById(orderId) ? TransactionResolution.COMMIT : TransactionResolution.ROLLBACK;
                })
                .build();
        //开始事务分支
        final Transaction transaction;
        try {
            transaction = producer.beginTransaction();
        }catch (ClientException e){
            e.printStackTrace();
            //事务分支开始失败，直接退出
            return;
        }
        Message message = messageBuilder.setTopic("topic")
                .setKeys("messageKey")
                .setTag("messageTag")
                //一般事务消息都会设置一个本地事务关联的唯一ID，用来做本地事务回查的检验
                .addProperty("OrderId", "xxx")
                .setBody("messageBody".getBytes())
                .build();
        //发送半事务消息
        final SendReceipt sendReceipt;
        try {
            sendReceipt = producer.send(message,transaction);
        }catch (ClientException e){
            e.printStackTrace();
            //半事务消息发送失败，事务可以直接退出并回滚
            return;
        }
        /**
         * 执行本地事务，并确定本地事务结果
         * 1、如果本地事务提交成功，即提交消息事务
         * 2、如果本地事务提交失败，则回滚消息事务
         * 3、如果本地事务未知异常，则不处理，等待事务消息回查
         */
        boolean localTransactionOK = doLocalTransaction();
        if(localTransactionOK){
            try {
                transaction.commit();
            }catch (ClientException e){
                //业务可以自身对实时性的要求选择是否重试，如果放弃重试，可以依赖事务消息回查机制进行事务状态的提交
                e.printStackTrace();
            }
        }else{
            try {
                transaction.rollback();
            }catch (ClientException e){
                //建议记录异常信息，回滚异常时可以无需重试，依赖事务消息回查机制进行事务状态的提交
                e.printStackTrace();
            }
        }

    }
}