【RocketMQ】RocketMQ应用难点

🎯 导读：本文探讨了RocketMQ中消息重复消费的问题及其解决方案，尤其是在CLUSTERING模式下的扩容影响。文章分析了重复消费的原因，如广播模式、负载均衡模式下的多consumerGroup消费、消费者组内的动态变化及网络延迟等，并提出了利用唯一标识进行去重的方法。此外，还讨论了如何选择合适的存储机制以高效处理大量消息标识，如HashMap、MySQL、Redis以及推荐的布隆过滤器等方案。对于防止消息堆积与确保消息不丢失，也给出了相应的策略和技术措施。

文章目录

消息重复消费问题（去重）
- 为什么出现重复消费
- 解决方案
- - MySQL去重
  - 布隆过滤器
  - 加锁（适合短时间的去重）
  - 过程
- 那些操作需要控制幂等
- 布隆过滤器实现
- - 生产者
  - 添加 hutool 的依赖
  - 消费者
如何解决消息堆积问题？
- 什么情况下会出现堆积
- - 生产速度远大于消费速度
  - 消费者消费出现问题
- 跳过堆积
- 重置消费点位
如何确保消息不丢失？
- 方式一：将mq的刷盘机制设置为同步刷盘
- 方式二：生产者做日志
- 方式三：集群主备模式
- 方式四：开启mq的trace机制，消息跟踪机制

消息重复消费问题（去重）

为什么出现重复消费

官网说明：RocketMQ确保所有消息至少传递一次。在大多数情况下，消息不会重复。

BROADCASTING（广播）模式下，所有注册的消费者都会消费，这些消费者通常是集群部署的一个个微服务，这样就会多台机器重复消费
CLUSTERING（负载均衡）模式下，如果一个topic被多个consumerGroup消费，也会重复消费
扩容影响：在 CLUSTERING 模式下，只有一个 consumerGroup ，一个队列只会分配给一个消费者，看起来好像是不会重复消费。但有个特殊情况：一个消费者新上线后，同组的所有消费者要重新负载均衡（反之，一个消费者掉线后也一样）。一个队列所对应的新的消费者要获取之前消费的offset，可能此时之前的消费者已经消费了一条消息，但并没有把 offset 提交给broker，那么新的消费者可能会重新消费一次。虽然orderly模式是前一个消费者先解锁，后一个消费者加锁再消费的模式，比起concurrently要严格了，但是加锁的线程和提交offset的线程不是同一个，所以还是会出现极端情况下的重复消费
在发送批量消息的时候，会被当做一条消息进行处理，那么如果批量消息中有一条业务处理成功，其他失败了，还是会被重新消费一次
网络卡顿，生产者多次发一样的消息（例如买一个东西，发送了两次减库存）

【扩容影响说明】

一开始只有一个消费者，4个队列都归C1管，C1已经开始消费a、b、c、d了，但是还没有消费完，没有返回让offset偏移

在这里插入图片描述

突然C2上线了，看到 c、d 还没有被消费完，然后自己又拿去消费一次

在这里插入图片描述

那么如果在CLUSTERING（负载均衡）模式下，并且在同一个消费者组中，不希望一条消息被重复消费，该怎么办呢？我们可以想到去重操作，找到消息唯一的标识，可以是msgId，也可以是你自定义的唯一的key，

解决方案

官网：RocketMQ 无法避免消息重复(Exactly-Once)，所以如果业务对消费重复非常敏感，务必在业务层面进行去重处理。可以借助关系数据库进行去重。首先需要确定消息的唯一键，可以是 msgld，也可以是消息内容中的唯一标识字段，例如订单 id 等。在消费之前判断唯一键是否在关系数据库中存在。如果不存在则插入，并消费，否则跳过。(实际过程要考虑原子性问题，判断不存在可以尝试插入，如果报主键冲突，则插入失败，直接跳过)

msgld 一定要是全局唯一标识符，但实际使用中，可能会存在相同的消息有两个不同 msgld 的情况（消费者主动重发、因客户端重投机制导致的重复等），这种情况需要使用业务字段进行重复消费。 使用自己的Key更安全

使用去重方案解决，例如将消息的唯一标识存起来，然后每次消费之前先判断是否存在这个唯一标识，如果存在则不消费，如果不存在则消费，并且消费以后将这个标记保存。消息的体量是非常大的，可能在生产环境中会到达上千万甚至上亿条，那么我们该如何选择一个容器来保存所有消息的标识，并且又可以快速的判断是否存在呢？

HashMap：单机部署可以使用，无法解决分布式问题
MySQL去重表，加唯一索引，插入成功就执行业务：太慢了，数据库压力大
Redis(setnx)：占用内存大，成本高
布隆过滤器（推荐）：内存小，效率高

MySQL去重

@SpringBootTest
class ARocketmqDemoApplicationTests {@Autowiredprivate JdbcTemplate jdbcTemplate;@Testvoid repeatProducer() throws Exception {DefaultMQProducer producer = new DefaultMQProducer("repeat-producer-group");producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);producer.start();String key = UUID.randomUUID().toString();System.out.println(key);// 测试 发两个key一样的消息Message m1 = new Message("repeatTopic", null, key, "扣减库存-1".getBytes());Message m1Repeat = new Message("repeatTopic", null, key, "扣减库存-1".getBytes());producer.send(m1);producer.send(m1Repeat);System.out.println("发送成功");producer.shutdown();}/*** mysql的唯一索引实现消费幂等性* ---------------------* 我们设计一个去重表 对消息的唯一key添加唯一索引* 每次消费消息的时候 先插入数据库 如果成功则执行业务逻辑 [如果业务逻辑执行报错 则删除这个去重表记录]* 如果插入失败 则说明消息来过了,直接签收了** @throws Exception*/@Testvoid repeatConsumer() throws Exception {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("repeat-consumer-group");consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);consumer.subscribe("repeatTopic", "*");consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {// 先拿keyMessageExt messageExt = msgs.get(0);String keys = messageExt.getKeys();// 原生方式操作Connection connection = null;try {connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/test?serverTimezone=GMT%2B8&useSSL=false", "root", "123456");} catch (SQLException e) {e.printStackTrace();}PreparedStatement statement = null;try {// 插入数据库 因为我们 key做了唯一索引statement = connection.prepareStatement("insert into order_oper_log(`type`, `order_sn`, `user`) values (1,'" + keys + "','123')");} catch (SQLException e) {e.printStackTrace();}// 新增 要么成功 要么报错   修改 要么成功,要么返回0 要么报错try {// 执行新增statement.executeUpdate();} catch (SQLException e) {System.out.println("executeUpdate");if (e instanceof SQLIntegrityConstraintViolationException) {// 唯一索引冲突异常// 说明消息来过了System.out.println("该消息来过了");// 签收消息，从队列中删除消息return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}e.printStackTrace();}// 如果业务报错 则删除掉这个去重表记录 delete order_oper_log where order_sn = keys;// 不删除的话，每次重试就会报错，重试失败System.out.println(new String(messageExt.getBody()));System.out.println(keys);return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.in.read();}}

布隆过滤器

布隆过滤器（Bloom Filter）是1970年由布隆提出的。它实际上是一个很长的二进制向量和一系列随机映射函数。布隆过滤器可以用于检索一个元素是否在一个集合中。它的优点是空间效率和查询时间都比一般的算法要好的多，缺点是有一定的误识别率和删除困难。

在hutool的工具中我们可以直接使用，使用redis的bitmap类型手写一个也可以的，Redisson中也有布隆过滤器的实现 https://hutool.cn/docs/#/bloomFilter/%E6%A6%82%E8%BF%B0

在这里插入图片描述

布隆过滤器判断key不在，说明消息没有被消费过，执行消费
布隆过滤器判断key存在，消息可能被消费过，需要做进一步判断（可以结合MySQL做进一步判断）

加锁（适合短时间的去重）

添加一个限时锁
如果可以加锁成功，执行消费，消费失败，释放锁；如果加锁失败，说明消息被消费了，或者正在消费中，直接返回即可，后续如果消费失败，消息会被再次投递

过程

生产者发消息带唯一标记
消费者控制消息消费幂等性（多次操作的影响均和第一次操作产生的影响相同）
- 方式一：查询key是否存在（存在就返回、不存在就新增并执行业务）
- 方式二：直接插入（插入成功就执行业务，否则返回）

那些操作需要控制幂等

新增：普通的新增操作是非幂等的（字段有唯一约束除外）

修改：看情况（++不是幂等，i=i+1是幂等）

查询：幂等操作

删除：幂等操作

布隆过滤器实现

生产者

@Test
public void testRepeatProducer() throws Exception {// 创建默认的生产者DefaultMQProducer producer = new DefaultMQProducer("test-group");// 设置nameServer地址producer.setNamesrvAddr("localhost:9876");// 启动实例producer.start();// 我们可以使用自定义key当做唯一标识String keyId = UUID.randomUUID().toString();System.out.println(keyId);Message msg = new Message("TopicTest", "tagA", keyId, "我是一个测试消息".getBytes());SendResult send = producer.send(msg);System.out.println(send);// 关闭实例producer.shutdown();
}

添加 hutool 的依赖

<dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>5.7.11</version>
</dependency>

消费者

/*** 在boot项目中可以使用@Bean在整个容器中放置一个单例对象*/
public static BitMapBloomFilter bloomFilter = new BitMapBloomFilter(100);@Test
public void testRepeatConsumer() throws Exception {// 创建默认消费者组DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group");consumer.setMessageModel(MessageModel.BROADCASTING);// 设置nameServer地址consumer.setNamesrvAddr("localhost:9876");// 订阅一个主题来消费   表达式，默认是*consumer.subscribe("TopicTest", "*");// 注册一个消费监听 MessageListenerConcurrently是并发消费// 默认是20个线程一起消费，可以参看 consumer.setConsumeThreadMax()consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context) {// 拿到消息的keyMessageExt messageExt = msgs.get(0);String keys = messageExt.getKeys();// 判断是否存在布隆过滤器中if (bloomFilter.contains(keys)) {// 执行进一步的精确判断boolean isConsumed = MySQL查询(keys);if (isConsumed == true) return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}执行业务处理（假如执行完业务，宕机了，key没有被添加到布隆过滤器中，还是重复消费）bloomFilter.add(keys);return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.in.read();
}

上面的方案已经可以解决绝大多数的重复消费问题了，像宕机这种极端场景，我还不知道完美的解决方案，有知道的大佬请告诉我，谢谢

如何解决消息堆积问题？

一般认为单条队列消息差值>=10w时算消息堆积

什么情况下会出现堆积

生产速度远大于消费速度

生产方可以做业务限流
增加消费者数量，但是消费者数量<=队列数量，适当增加消费线程数量
- ```
@RocketMQMessageListener(topic = "modeTopic",consumerGroup = "mode-consumer-group-a", messageModel = MessageModel.CLUSTERING, consumeThreadNumber = 40)
public class MyConsumer implements RocketMQConsumer {// ...
}
```
- 核心线程数 < CPU核心数
- 最大线程数
  - IO密集，读文件、操作数据库，CPU快，磁盘慢，CPU空闲很多，建议设置为2n，n是CPU的最大处理器数量，如下图所示，建议通过Runtime.getRuntime().availableProcessors()来获取，因为不同服务器这个是不一样的，不建议写死
  - CPU密集，CPU使用频繁，如果开太多，CPU核数不够用，会频繁切换，效率低，建议设置(n+1)
动态扩容队列数量（一般由运维工程师来决定），从而增加消费者数量。动态扩容之后，程序不是一下就感知到的，刚扩容的时候，新来的队列还是收不到消息，要过一段时间才会收到
- 读队列数量不要设置大于写队列数量（读多没有用），直接设置相等就可以
- perm：设置为2，这个主题的消息只能读不能写；设置为4，只能写，不能读；设置为6，可写可读
- 当队列的最大位点不是全为0的话（如下图所示），不可以缩容，不然会出问题，有的队列的消息会丢失；全是0的时候，可以缩

消费者消费出现问题

排查消费者程序的问题

跳过堆积

跳过堆积，这个组的消息都不要了，偏移量会自动往后面移动，表示已经消费过。跳过之后，无法回滚，要谨慎

在这里插入图片描述

重置消费点位

从这个时间开始至今被消费过的消息，会重新被消费

在这里插入图片描述

如何确保消息不丢失？

硬盘读写方式：

随机读写：将数据不固定存储到不同的扇区，查询数据较慢
顺序读写：提前申请一片空间，将数据顺序存储（MQ使用的是这种）

方式一：将mq的刷盘机制设置为同步刷盘

消息持久化：

同步刷盘：生产者给broker发送消息，broker先把消息持久化到磁盘之后，再返回成功给生产者。安全，性能降低，但其实性能还是不错的
异步刷盘：先把数据存储到内存的buffer中，到达一定的量，再存储到磁盘中。不安全，性能好

方式二：生产者做日志

不用同步刷盘，生产者可以自己做消息日志，写到文件或者数据库中，不占用MQ的性能

生产者使用同步发送模式，收到mq的返回确认以后，顺便往自己的数据库里面写key createTime status=1。消费者消费以后，修改数据这条消息的状态 = 2，记录消息的handleTime
写一个定时任务，间隔两天去查询数据，如果有status = 1 and createTime < day-2，将这个消息补发一次