RocketMQ 第二章

7、SpringBoot整合RocketMQ

SpringBoot 提供了快捷操作 RocketMQ 的 RocketMQTemplate 对象。

7.1、引入依赖

注意依赖的版本需要和 RocketMQ 的版本相同。

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.1.1</version>
</dependency>

7.2、编写配置文件

# 应用名称
spring.application.name=my-boot-producer-demo
# 应用服务 WEB 访问端口
server.port=8080
# nameserver地址
rocketmq.name-server=119.91.21.45:9876;1.117.115.99:9876
# 生产者组
rocketmq.producer.group=my-boot-producer-group

7.3、编写⽣产者发送普通消息

@Component
public class MyProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void sendMessage(String topic, String message) {
        /**
         * convertAndSend：将字符串转换成Message对象
         */
        rocketMQTemplate.convertAndSend(topic, message);
    }

}

7.4、编写JUnit单元测试发送消息

		@Test
    void testSendMessage() {
        String topic = "my-boot-topic";
        String message = "hello rocket mq springboot message";
        producer.sendMessage(topic, message);
        System.out.println("消息发送成功！");
    }

7.5、创建消费者程序

@Component
@RocketMQMessageListener(consumerGroup = "my-boot-consumer-group1", topic = "my-boot-topic")
public class MyConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        System.out.println("收到的消息:" + message);
    }

}

7.6、发送事务消息

编写生产者方法

		/**
     * 发送事务消息
     *
     * @param topic
     * @param msg
     * @throws InterruptedException
     */
    public void sendMessageInTransaction(String topic, String msg) throws InterruptedException {
        String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};
        for (int i = 0; i < 10; i++) {
            Message<String> message = MessageBuilder.withPayload(msg).build();
            // topic和tag整合在一起
            String destination = topic + ":" + tags[i % tags.length];
            // 第一个destination是消息要发送的目的地topic，第二个destination是消息携带的业务数据
            TransactionSendResult sendResult = rocketMQTemplate.sendMessageInTransaction(destination, message, destination);
            System.out.println(sendResult);
            Thread.sleep(10);
        }
    }

编写事务监听器类

@RocketMQTransactionListener(rocketMQTemplateBeanName = "rocketMQTemplate")
public class MyTransactionListener implements RocketMQLocalTransactionListener {

    /**
     * 执行本地事务
     *
     * @param msg
     * @param arg 业务参数
     * @return
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        String destination = (String) arg;
        // 把spring类型的message转换成Rocketmq类型的message
        org.apache.rocketmq.common.message.Message message = RocketMQUtil.convertToRocketMessage(new StringMessageConverter(), "utf-8", destination, msg);
        // 如何得到message上的tag的内容
        String tags = message.getTags();
        if (StringUtils.contains(tags, "TagA")) {
            return RocketMQLocalTransactionState.COMMIT;
        } else if (StringUtils.contains(tags, "TagB")) {
            return RocketMQLocalTransactionState.ROLLBACK;
        } else {
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        return null;
    }

}

编写单元测试发送事务消息

		@Test
    void testSendMessageInTransaction() throws InterruptedException {
        String topic = "my-boot-topic";
        String message = "hello rocket mq transaction springboot message";
        producer.sendMessageInTransaction(topic, message);
        System.out.println("事务消息发送成功");
    }

8、Spring Cloud Stream 整合 RocketMQ

8.1、Spring Cloud Stream介绍

Spring Cloud Stream 是⼀个框架，⽤于构建与共享消息系统连接的⾼度可扩展的事件驱动微服务。
该框架提供了⼀个灵活的编程模型，该模型基于已经建⽴和熟悉的 Spring 习惯⽤法和最佳实践，包括对持久 pub/sub 语义、消费者组和有状态分区的⽀持。

Untitled

Spring Cloud Stream 的核⼼模块：

Destination Binders：负责提供与外部消息传递系统集成的组件。
Destination Bindings：外部消息系统和最终⽤户提供的应⽤程序代码（⽣产者/消费者）之间的桥梁。
Message：⽣产者和消费者⽤来与⽬标绑定器（以及通过外部消息系统的其他应⽤程序）进⾏通信的规范数据结构。

8.2、编写生产者

引入依赖

<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
    <exclusions>
        <exclusion>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
        </exclusion>
        <exclusion>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-acl</artifactId>
        </exclusion>
    </exclusions>
</dependency>

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.7.1</version>
</dependency>
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-acl</artifactId>
    <version>4.7.1</version>
</dependency>

编写配置文件

# 应用名称
spring.application.name=my-spring-cloud-rocketmq-producer
# 应用服务 WEB 访问端口
server.port=8080
# output 生产者的目的地topic
spring.cloud.stream.bindings.output.destination=TopicTest
# 配置rocketmq连接的nameserver地址
spring.cloud.stream.rocketmq.binder.name-server=119.91.21.45:9876;1.117.115.99:9876

在启动类上打上注解

@EnableBinding(Source.class)
@SpringBootApplication
public class MySpringCloudRocketmqProducerApplication {

    public static void main(String[] args) {
        SpringApplication.run(MySpringCloudRocketmqProducerApplication.class, args);
    }

}

其中 @EnableBinding(Source.class) 是指向配置⽂件的 output 参数。

编写生产者程序

@Component
public class MyProducer {

    @Resource
    private Source source;

    public void sendMessage(String msg) {
        // 封装消息头
        Map<String, Object> headers = new HashMap<>();
        headers.put(MessageConst.PROPERTY_TAGS, "TagA");
        MessageHeaders messageHeaders = new MessageHeaders(headers);
        // 创建消息对象
        Message<String> message = MessageBuilder.createMessage(msg, messageHeaders);
        // 发送消息
        source.output().send(message);
    }

}

编写单元测试发送消息

@SpringBootTest
class MySpringCloudRocketmqProducerApplicationTests {
    
    @Autowired
    private MyProducer producer;

    @Test
    void testSendMessage() {
        producer.sendMessage("hello spring cloud stream message");
    }

}

8.3、编写消费者

引入依赖

<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
    <exclusions>
        <exclusion>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
        </exclusion>
        <exclusion>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-acl</artifactId>
        </exclusion>
    </exclusions>
</dependency>

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.7.1</version>
</dependency>
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-acl</artifactId>
    <version>4.7.1</version>
</dependency>

编写配置文件

# 应用名称
spring.application.name=my-spring-cloud-rocketmq-consumer
# 应用服务 WEB 访问端口
server.port=8081
# input配置消费者的信息
spring.cloud.stream.bindings.input.destination=TopicTest
# 配置消费者的消费组
spring.cloud.stream.bindings.input.group=spring-cloud-stream-consumer-group
# 配置nameserver的地址
spring.cloud.stream.rocketmq.binder.name-server=119.91.21.45:9876;1.117.115.99:9876

在启动类上打上注解

@EnableBinding(Sink.class)
@SpringBootApplication
public class MySpringCloudRocketmqConsumerApplication {

    public static void main(String[] args) {
        SpringApplication.run(MySpringCloudRocketmqConsumerApplication.class, args);
    }

}

其中 @EnableBinding(Sink.class) 是指向配置⽂件的 input 参数。

编写消费者程序

@Component
public class MyConsumer {

    @StreamListener(Sink.INPUT)
    public void processMessage(String message) {
        System.out.println("收到的消息：" + message);
    }

}

9、RocketMQ核心概念

9.1、消息模型（Message Model）

RocketMQ主要由 Producer、Broker、Consumer 三部分组成，其中 Producer 负责⽣产消息，Consumer 负责消费消息，Broker 负责存储消息。Broker 在实际部署过程中对应⼀台服务器，每个 Broker 可以存储多个 Topic 的消息，每个 Topic 的消息也可以分⽚存储于不同的 Message Queue。Message Queue ⽤于存储消息的物理地址，每个 Topic 中的消息地址存储于多个 Message Queue 中。ConsumerGroup 由多个 Consumer 实例构成。

9.2、消息⽣产者（Producer）

负责⽣产消息，⼀般由业务系统负责⽣产消息。⼀个消息⽣产者会把业务应⽤系统⾥产⽣的消息发送到 broker 服务器。RocketMQ提供多种发送⽅式，同步发送、异步发送、顺序发送、单向发送。同步和异步⽅式均需要 Broker 返回确认信息，单向发送不需要。

⽣产者组将多个⽣产者归为⼀组。⽤于保证⽣产者的⾼可⽤，⽐如在事务消息中回查本地事务状态，需要⽣产者具备⾼可⽤的特性，才能完成整个任务。

9.3、消息消费者（Consumer）

负责消费消息，⼀般是后台系统负责异步消费。⼀个消息消费者会从 Broker 服务器拉取消息、并将其提供给应⽤程序。从⽤户应⽤的⻆度⽽⾔提供了两种消费形式：拉取式消费、推动式消费。

消费者组将多个消息消费者归为⼀组，⽤于保证消费者的⾼可⽤和⾼性能。

9.4、主题（Topic）

表示⼀类消息的集合，每个主题包含若⼲条消息，每条消息只能属于⼀个主题，是 RocketMQ 进⾏消息订阅的基本单位。

9.5、代理服务器（Broker Server）

消息中转⻆⾊，负责存储消息、转发消息。代理服务器在 RocketMQ 系统中负责接收从⽣产者发送来的消息并存储、同时为消费者的拉取请求作准备。代理服务器也存储消息相关的元数据，包括消费者组、消费进度偏移量、主题和队列消息等。

9.6、名字服务（Name Server）

名称服务充当路由消息的提供者。⽣产者或消费者能够通过名字服务查找各主题相应的 Broker IP 列表。多个 Namesrv 实例组成集群，但彼此之间相互独⽴，没有信息交换。

9.7、拉取式消费（Pull Consumer）

Consumer 消费的⼀种类型，应⽤通常主动调⽤ Consumer 的拉消息⽅法从 Broker 服务器拉消息、主动权由应⽤程序控制。⼀旦获取了批量消息，应⽤就会启动消费过程。

9.8、推动式消费（Push Consumer）

Consumer 消费的⼀种类型，该模式下 Broker 收到数据后会主动推送给消费端，该消费模式⼀般实时性较⾼。

9.9、⽣产者组（Producer Group）

同⼀类 Producer 的集合，这类 Producer 发送同⼀类消息且发送逻辑⼀致。如果发送的是事务消息且原始⽣产者在发送之后崩溃，则 Broker 服务器会联系同⼀⽣产者组的其他⽣产者实例以提交或回溯消费。

9.10、消费者组（Consumer Group）

同⼀类 Consumer 的集合，这类 Consumer 通常消费同⼀类消息且消费逻辑⼀致。消费者组使得在消息消费⽅⾯，实现负载均衡和容错的⽬标变得⾮常容易。要注意的是，消费者组的消费者实例必须订阅完全相同的Topic。RocketMQ ⽀持两种消息模式：集群消费（Clustering）和⼴播消费（Broadcasting）。

9.11、集群消费（Clustering）

集群消费模式下，相同 Consumer Group 的每个 Consumer 实例平均分摊消息。

Untitled

9.12、⼴播消费（Broadcasting）

⼴播消费模式下，相同 Consumer Group 的每个 Consumer 实例都能接收到全量的消息。

9.13、普通顺序消息（Normal Ordered Message）

普通顺序消费模式下，消费者通过同⼀个消费队列收到的消息是有顺序的，不同消息队列收到的消息则可能是⽆顺序的。

9.14、严格顺序消息（Strictly Ordered Message）

严格顺序消息模式下，消费者收到的所有消息均是有顺序的。

9.15、消息（Message）

消息系统所传输信息的物理载体，⽣产和消费数据的最⼩单位，每条消息必须属于⼀个主题。RocketMQ 中每个消息拥有唯⼀的 Message ID，且可以携带具有业务标识的 Key。系统提供了通过 Message ID 和 Key 查询消息的功能。

9.16、标签（Tag）

为消息设置的标志，⽤于在同⼀主题下区分不同类型的消息。来⾃同⼀业务单元的消息，可以根据不同业务⽬的在同⼀主题下设置不同标签。标签能够有效地保持代码的清晰度和连贯性，并优化 RocketMQ 提供的查询系统。消费者可以根据 Tag 实现对不同⼦主题的不同消费逻辑，实现更好的扩展性。

10、消息存储机制

Untitled

消息存储是 RocketMQ 中最为复杂和最为重要的⼀部分，本节将分别从 RocketMQ 的消息存储整体架构、PageCache 与 Mmap 内存映射以及 RocketMQ 中两种不同的刷盘⽅式三⽅⾯来分别展开叙述。

10.1、消息存储整体架构

Untitled

消息存储架构图中主要由下⾯三个跟消息存储相关的⽂件构成。

CommitLog

消息主体以及元数据的存储主体，存储 Producer 端写⼊的消息主体内容，消息内容不是定⻓的。单个⽂件⼤⼩默认 1G ，⽂件名⻓度为 20 位，左边补零，剩余为起始偏移量，⽐如 00000000000000000000 代表了第⼀个⽂件，起始偏移量为 0，⽂件⼤⼩为 1G=1073741824；当第⼀个⽂件写满了，第⼆个⽂件为 00000000001073741824，起始偏移量为 1073741824，以此类推。消息主要是顺序写⼊⽇志⽂件，当⽂件写满了，接着写⼊下⼀个⽂件；

ConsumeQueue

消息消费队列，引⼊的⽬的主要是提⾼消息消费的性能，由于 RocketMQ 是基于主题 topic 的订阅模式，消息消费是针对主题进⾏的，如果要遍历CommitLog ⽂件中根据 topic 检索消息是⾮常低效的。Consumer 即可根据 ConsumeQueue 来查找待消费的消息。其中，ConsumeQueue（逻辑消费队列）作为消费消息的索引，保存了指定 Topic 下的队列消息在 CommitLog 中的起始物理偏移量 offset，消息⼤⼩ size 和消息 Tag 的 HashCode 值。ConsumeQueue ⽂件可以看成是基于 topic 的 CommitLog 索引⽂件，故 ConsumeQueue ⽂件夹的组织⽅式如下：topic/queue/file 三层组织结构，具体存储路径为：$HOME/store/consumequeue/{topic}/{queueId}/{fileName}。同样 ConsumeQueue ⽂件采取定⻓设计，每⼀个条⽬共 20 个字节，分别为 8 字节的 CommitLog 物理偏移量、4 字节的消息⻓度、8字节 Tag HashCode，单个⽂件由 30W 个条⽬组成，可以像数组⼀样随机访问每⼀个条⽬，每个ConsumeQueue ⽂件⼤⼩约为 5.72M；

IndexFile

IndexFile（索引⽂件）提供了⼀种可以通过 key 或时间区间来查询消息的⽅法。Index ⽂件的存储位置是：$HOME \store\index${fileName}，⽂件名fileName 是以创建时的时间戳命名的，固定的单个 IndexFile ⽂件⼤⼩约为 400 M，⼀个 IndexFile 可以保存 2000W 个索引，IndexFile 的底层存储设计为在⽂件系统中实现 HashMap 结构，故 RocketMQ 的索引⽂件其底层实现为 hash 索引。

在上⾯的 RocketMQ 的消息存储整体架构图中可以看出，RocketMQ 采⽤的是混合型的存储结构，即为 Broker 单个实例下所有的队列共⽤⼀个⽇志数据⽂件（CommitLog）来存储。RocketMQ 的混合型存储结构（多个Topic的消息实体内容都存储在⼀个 CommitLog 中）针对 Producer 和 Consumer 分别采⽤了数据和索引部分相分离的存储结构，Producer 发送消息⾄ Broker 端，然后 Broker 端使⽤同步或者异步的⽅式对消息进行刷盘持久化，保存⾄ CommitLog 中。只要消息被刷盘持久化到磁盘⽂件 CommitLog 中，那么 Producer 发送的消息就不会丢失。正因为如此，Consumer 也就肯定有机会去消费这条消息。当⽆法拉取到消息后，可以等下⼀次消息拉取，同时服务端也⽀持⻓轮询模式，如果⼀个消息拉取请求未拉取到消息，Broker 允许等待 30s 的时间，只要这段时间内有新消息到达，将直接返回给消费端。这⾥，RocketMQ 的具体做法是，使⽤ Broker 端的后台服务线程— ReputMessageService 不停地分发请求并异步构建 ConsumeQueue（逻辑消费队列）和 IndexFile（索引⽂件）数据。

10.2、⻚缓存与内存映射

⻚缓存（PageCache）是 OS 对⽂件的缓存，⽤于加速对⽂件的读写。⼀般来说，程序对⽂件进⾏顺序读写的速度⼏乎接近于内存的读写速度，主要原因就是由于 OS 使⽤ PageCache 机制对读写访问操作进⾏了性能优化，将⼀部分的内存⽤作 PageCache。对于数据的写⼊，OS 会先写⼊⾄ Cache 内，随后通过异步的⽅式由 pdflush 内核线程将 Cache 内的数据刷盘⾄物理磁盘上。对于数据的读取，如果⼀次读取⽂件时出现未命中 PageCache 的情况，OS 从物理磁盘上访问读取⽂件的同时，会顺序对其他相邻块的数据⽂件进⾏预读取。

Untitled

在 RocketMQ 中，ConsumeQueue 逻辑消费队列存储的数据较少，并且是顺序读取，在 page cache 机制的预读取作⽤下，Consume Queue⽂件的读性能⼏乎接近读内存，即使在有消息堆积的情况下也不会影响性能。⽽对于 CommitLog 消息存储的⽇志数据⽂件来说，读取消息内容时会产⽣较多的随机访问读取，严重影响性能。如果选择合适的系统 IO 调度算法，⽐如设置调度算法为“Deadline”（此时块存储采⽤ SSD 的话），随机读的性能也会有所提升。

另外，RocketMQ 主要通过 MappedByteBuffer 对⽂件进⾏读写操作。其中，利⽤了 NIO 中的 FileChannel 模型将磁盘上的物理⽂件直接映射到⽤户态的内存地址中（这种 Mmap 的⽅式减少了传统 IO 将磁盘⽂件数据在操作系统内核地址空间的缓冲区和⽤户应⽤程序地址空间的缓冲区之间来回进⾏拷⻉的性能开销），将对⽂件的操作转化为直接对内存地址进⾏操作，从⽽极⼤地提⾼了⽂件的读写效率（正因为需要使⽤内存映射机制，故 RocketMQ 的⽂件存储都使⽤定⻓结构来存储，⽅便⼀次将整个⽂件映射⾄内存）。

10.3、消息刷盘

Untitled

同步刷盘

Untitled

如上图所示，只有在消息真正持久化到磁盘后，RocketMQ 的 Broker 端才会真正返回给 Producer 端⼀个成功的 ACK 响应。同步刷盘对 MQ 消息可靠性来说是⼀种不错的保障，但是性能上会有较⼤影响，⼀般适⽤于⾦融业务应⽤该模式较多。

异步刷盘

能够充分利⽤ OS 的 PageCache 的优势，只要消息写⼊ PageCache 即可将成功的 ACK 返回给 Producer 端。消息刷盘采⽤后台异步线程提交的⽅式进⾏，降低了读写延迟，提⾼了 MQ 的性能和吞吐量。

11、集群核⼼概念

11.1、消息主从复制

RocketMQ 官⽅提供了三种集群搭建⽅式。

2主2从异步通信⽅式

Untitled

使⽤异步⽅式进⾏主从之间的数据复制，吞吐量⼤，但可能会丢消息。

使⽤ conf/2m-2s-async ⽂件夹内的配置⽂件做集群配置。

2主2从同步通信⽅式

Untitled

使⽤同步⽅式进⾏主从之间的数据复制，保证消息安全投递，不会丢失，但影响吞吐量

使⽤ conf/2m-2s-sync ⽂件夹内的配置⽂件做集群配置。

2主⽆从⽅式

不存在复制消息，会存在单点故障，且读的性能没有前两种⽅式好。

使⽤ conf/2m-noslave ⽂件夹内的配置⽂件做集群配置。

11.2、负载均衡

RocketMQ 中的负载均衡都在 Client 端完成，具体来说的话，主要可以分为 Producer 端发送消息时候的负载均衡和 Consumer 端订阅消息的负载均衡。

Producer的负载均衡

Untitled

Producer端在发送消息的时候，会先根据 Topic 找到指定的 TopicPublishInfo，在获取了 TopicPublishInfo 路由信息后，RocketMQ 的客户端在默认⽅式下
selectOneMessageQueue( )⽅法会从 TopicPublishInfo 中的 messageQueueList 中选择⼀个队列（MessageQueue）进⾏发送消息。具体的容错策略均在MQFaultStrategy 这个类中定义。这⾥有⼀个 sendLatencyFaultEnable 开关变量，如果开启，在随机递增取模的基础上，再过滤掉 not available 的 Broker 代理。所谓的"latencyFaultTolerance"，是指对之前失败的，按⼀定的时间做退避。例如，如果上次请求的 latency 超过 550 Lms，就退避 3000 Lms；超过1000L，就退避60000L；如果关闭，采⽤随机递增取模的⽅式选择⼀个队列（MessageQueue）来发送消息，latencyFaultTolerance机制是实现消息发送⾼可⽤的核⼼关键所在。

Untitled

Consumer的负载均衡

在RocketMQ中，Consumer端的两种消费模式（Push/Pull）都是基于拉模式来获取消息的，⽽在Push模式只是对pull模式的⼀种封装，其本质实现为消息拉取线程在从服务器拉取到⼀批消息后，然后提交到消息消费线程池后，⼜“⻢不停蹄”的继续向服务器再次尝试拉取消息。如果未拉取到消息，则延迟⼀下⼜继续拉取。在两种基于拉模式的消费⽅式（Push/Pull）中，均需要Consumer端在知道从Broker端的哪⼀个消息队列—队列中去获取消息。因此，有必要在Consumer端来做负载均衡，即Broker端中多个MessageQueue分配给同⼀个ConsumerGroup中的哪些Consumer消费。

Consumer 端的负载均衡可以通过 consumer 的 api 进⾏设置：

consumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueAveragelyByCircle());

AllocateMessageQueueStrategy 接⼝的实现类实现了不同的负载均衡策略：

AllocateMachineRoomNearby：基于就近机房优先级的代理分配策略。可以指定实际的分配策略。如果任何使⽤者在机房中活动，则部署在同⼀台机器中的代理的消息队列应该分配给这些使⽤者。否则，这些消息队列可以与所有消费者共享，因为没有活着的消费者可以垄断它们
AllocateMessageQueueAveragely：平均哈希队列算法
AllocateMessageQueueAveragelyByCircle：轮询哈希队列算法
AllocateMessageQueueByConfig：不分配，通过指定 MessageQueue 列表来消费
AllocateMessageQueueByMachineRoom：机房哈希队列算法，如⽀付宝逻辑机房
AllocateMessageQueueConsistentHash：⼀致性哈希队列算法，带有虚拟节点的⼀致性哈希环。

注意，在 MessageQueue 和 Consumer 之间⼀旦发⽣对应关系的改变，就会触发 rebalance，进⾏重新分配。

Untitled

11.3、消息重试

⾮⼴播模式下，Consumer消费消息失败后，需要提供⼀种重试机制，令消息再消费⼀次。Consumer 消费消息失败通常可以认为有以下⼏种情况：

由于消息本身的原因，例如反序列化失败，消息数据本身⽆法处理（例如话费充值，当前消息的⼿机号被注销，⽆法充值）等。这种错误通常需要跳过这条消息，再消费其它消息，⽽这条失败的消息即使⽴刻重试消费，99% 也不成功，所以最好提供⼀种定时重试机制，即过 10 秒后再重试。
由于依赖的下游应⽤服务不可⽤，例如 db 连接不可⽤，外系统⽹络不可达等。遇到这种错误，即使跳过当前失败的消息，消费其他消息同样也会报错。这种情况建议应⽤sleep 30s，再消费下⼀条消息，这样可以减轻 Broker 重试消息的压⼒。

在代码层⾯，如果消费者返回的是以下三种情况，则消息会重试消费：

		consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("收到的消息：" + msg);
                }
                // 过段时间再重试消费
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                // 下面这种等价于上面的
                // return null;
                // 或者抛出异常，也会进行重试消费
            }
        });

消费者返回 null，或者返回 ConsumeConcurrentlyStatus.RECONSUME_LATER，或者抛出异常，都会触发消息重试。

关于重试次数

RocketMQ 会为每个消费组都设置⼀个 Topic 名称为 “%RETRY%+consumerGroup” 的重试队列（这⾥需要注意的是，这个 Topic 的重试队列是针对消费组，⽽不是针对每个 Topic 设置的），⽤于暂时保存因为各种异常⽽导致 Consumer 端⽆法消费的消息。考虑到异常恢复起来需要⼀些时间，会为重试队列设置多个重试级别，每个重试级别都有与之对应的重新投递延时，重试次数越多投递延时就越⼤。RocketMQ 对于重试消息的处理是先保存⾄ Topic 名称为 “SCHEDULE_TOPIC_XXXX” 的延迟队列中，后台定时任务按照对应的时间进⾏ Delay 后重新保存⾄ “%RETRY%+consumerGroup” 的重试队列中。

与延迟队列的设置相同，消息默认会重试 16 次，每次重试的时间间隔如下：

10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

注意：重试超过指定次数的消息后，将会进⼊到死信队列中 %DLQ%my-consumer-group1。

11.4、死信队列

死信队列⽤于处理⽆法被正常消费的消息。当⼀条消息初次消费失败，消息队列会⾃动进⾏消息重试；达到最⼤重试次数后，若消费依然失败，则表明消费者在正常情况下⽆法正确地消费该消息，此时，消息队列不会⽴刻将消息丢弃，⽽是将其发送到该消费者对应的特殊队列中。

RocketMQ 将这种正常情况下⽆法被消费的消息称为死信消息（Dead-LetterMessage），将存储死信消息的特殊队列称为死信队列（Dead-Letter Queue）。在 RocketMQ 中，可以通过使⽤ console 控制台对死信队列中的消息进⾏重发来使得消
费者实例再次进⾏消费。

死信队列具备以下特点：

RocketMQ 会⾃动为需要死信队列的 ConsumerGroup 创建死信队列。
死信队列与 ConsumerGroup 对应，死信队列中包含该 ConsumerGroup 所有相关 topic 的死信消息。
死信队列中消息的有效期与正常消息相同，默认48⼩时。
若要消费死信队列中的消息，需在控制台将死信队列的权限设置为 6，即可读可写。

Untitled

11.5、幂等消息

幂等性：多次操作造成的结果是⼀致的。对于⾮幂等的操作，幂等性如何保证？

1、在请求方式中的幂等性体现：

get：多次get 结果是⼀致的
post：添加，⾮幂等
put：修改：幂等，根据 id 修改
delete：根据 id 删除，幂等

对于⾮幂等的请求，我们在业务⾥要做幂等性保证。

2、在消息队列中的幂等性体现：

在消息队列中，很可能⼀条消息被冗余部署的多个消费者收到，对于⾮幂等的操作，⽐如⽤户注册，就需要做幂等性保证，否则消息将会被重复消费。可以将情况概括为以下⼏种：

⽣产者重复发送：由于⽹络抖动，导致⽣产者没有收到 broker 的 ack 确认⽽再次重发消息，实际上 broker 收到了多条重复的消息，造成消息重复。
消费者重复消费：由于⽹络抖动，消费者没有返回 ack 给 broker，导致消费者重试消费。
rebalance 时的重复消费：由于⽹络抖动，在 rebalance 重分配时也可能出现消费者重复消费某条消息。

3、如何保证幂等性消费？

mysql 插⼊业务 id 作为主键，主键是唯⼀的，所以⼀次只能插⼊⼀条
使⽤ redis 或 zk 的分布式锁（主流的⽅案）

12、RocketMQ最佳实践

12.1、保证消息顺序消费

为什么要保证消息有序？

⽐如有这么⼀个物联⽹的应⽤场景，IOT 中的设备在初始化时需要按顺序接收这样的消息：

设置设备名称
设置设备的⽹络
重启设备使配置⽣效

如果这个顺序颠倒了，可能就没有办法让设备的配置⽣效，因为只有重启设备才能让配置⽣效，但重启的消息却在设置设备消息之前被消费。

如何保证消息顺序消费？

全局有序：消费的所有消息都严格按照发送消息的顺序进⾏消费
局部有序：消费的部分消息按照发送消息的顺序进⾏消费

Untitled

12.2、快速处理积压消息

在 rocketmq 中，如果消费者消费速度过慢，⽽⽣产者⽣产消息的速度⼜远超于消费者消费消息的速度，那么就会造成⼤量消息积压在 mq 中。

Untitled

如何查看消息积压的情况？

在 console 控制台中可以查看：

Untitled

如何解决消息积压？

在这个消费者中，使⽤多线程，充分利⽤机器的性能进⾏消费消息。
通过业务的架构设计，提升业务层⾯消费的性能。
创建⼀个消费者，该消费者在 RocketMQ 上另建⼀个主题，该消费者将 poll 下来的消息，不进⾏消费，直接转发到新建的主题上。新建的主题配上多个
MessageQueue，多个 MessageQueue 再配上多个消费者。此时，新的主题的多个分区的多个消费者就开始⼀起消费了。

Untitled

12.3、保证消息可靠性投递

Untitled

保证消息可靠性投递，⽬的是消息不丢失，可以顺利抵达消费者并被消费。要想实现可靠性投递，需要完成以下⼏个部分。

1、⽣产者发送事务消息

2、broker 集群使⽤ Dledger ⾼可⽤集群

dledger 集群的数据同步由两阶段完成：

第⼀阶段：同步消息到 follower，此时的消息状态是 uncommitted。follower 在收到消息以后，返回⼀个 ack 给 leader，leader ⾃⼰也会返回 ack 给⾃⼰。leader在收到集群中的半数以上的 ack 后开始进⼊到第⼆阶段。
第⼆阶段：leader 发送 committed 命令，集群中的所有 broker 把消息写⼊到⽇志⽂件中，此时该消息才表示接收完毕，允许消费者进行消费。

3、保证消费者的同步消费

消费者使⽤同步的⽅式，在消费完后返回 ack。

4、使⽤基于缓存中间件的 MQ 降级⽅案

当 MQ 整个服务不可⽤时，为了防⽌服务雪崩，消息可以暂存于缓存中间件中，⽐如 redis。待 MQ 恢复后，将 redis 中的数据重新刷进 MQ 中。