一、RocketMQ 运行架构回顾RocketMQ 的客户端编程模型相对固定核心是围绕 NameServer、Broker、Producer 和 Consumer 四个角色展开。客户端只需指定 NameServer 地址无需关心具体 Broker 地址由 NameServer 负责路由发现。二、RocketMQ 消息模型深入理解1. 客户端基本流程RocketMQ 基于 Maven 提供客户端核心依赖 rocketmq-client。生产者与消费者的编程步骤相对固定掌握固定步骤有助于学习更复杂的消息模型。生产者固定步骤创建消息生产者 producer并指定生产者组名指定 NameServer 地址启动 producer与服务端建立连接的过程创建消息对象指定主题 Topic、Tag 和消息体发送消息关闭生产者 producer释放资源public class Producer { public static void main(String[] args) throws MQClientException, InterruptedException { //初始化一个消息生产者 DefaultMQProducer producer new DefaultMQProducer(please_rename_unique_group_name); // 指定nameserver地址 producer.setNamesrvAddr(192.168.65.112:9876); // 启动消息生产者服务 producer.start(); for (int i 0; i 2; i) { try { // 创建消息。消息由Topic,Tag和body三个属性组成其中Body就是消息内容 Message msg new Message(TopicTest,TagA,(Hello RocketMQ i).getBytes(RemotingHelper.DEFAULT_CHARSET)); //发送消息获取发送结果 SendResult sendResult producer.send(msg); System.out.printf(%s%n, sendResult); } catch (Exception e) { e.printStackTrace(); Thread.sleep(1000); } } //消息发送完后停止消息生产者服务。 producer.shutdown(); } }消费者固定步骤创建消费者 Consumer必须指定消费者组名指定 NameServer 地址订阅主题 Topic 和 Tag设置回调函数处理消息启动消费者 consumer消费者会一直挂起持续处理消息public class Consumer { public static void main(String[] args) throws InterruptedException, MQClientException { //构建一个消息消费者 DefaultMQPushConsumer consumer new DefaultMQPushConsumer(please_rename_unique_group_name_4); //指定nameserver地址 consumer.setNamesrvAddr(192.168.65.112:9876); consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); // 订阅一个感兴趣的话题这个话题需要与消息的topic一致 consumer.subscribe(TopicTest, *); // 注册一个消息回调函数消费到消息后就会触发回调。 consumer.registerMessageListener(new MessageListenerConcurrently() { Override public ConsumeConcurrentlyStatus consumeMessage(ListMessageExt msgs,ConsumeConcurrentlyContext context) { msgs.forEach(messageExt - { try { System.out.println(收到消息:new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET)); } catch (UnsupportedEncodingException e) {} }); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); //启动消费者服务 consumer.start(); System.out.print(Consumer Started); } }指定 NameServer 有两种方式1、客户端直接指定优先级更高例如例如 consumer.setNameSrvAddr(127.0.0.1:9876)2、通过系统环境变量 NAMESRV_ADDR 指定。2. 消息确认机制消息安全是互联网金融服务的基础保障RocketMQ 从生产端和消费端两个维度保证消息安全。生产端三种发送方式单向发送sendOneway只管发送不关心 Broker 是否接收成功效率最高适用于允许丢失消息的场景如日志。同步发送send发送后阻塞等待 Broker 返回 SendResult能最大程度保证消息安全但效率较低。SendStatus 枚举包含 SEND_OK、FLUSH_DISK_TIMEOUT、FLUSH_SLAVE_TIMEOUT、SLAVE_NOT_AVAILABLE。异步发送send SendCallback注册回调函数不阻塞主线程兼顾安全性与高吞吐。注意回调触发前不能调用 shutdown()否则回调无法执行。超时时间默认 3 秒可通过 setSendMsgTimeout 定制。消费端状态确认机制消费者通过返回 ConsumeConcurrentlyStatus 枚举CONSUME_SUCCESS 或 RECONSUME_LATER告知 Broker 处理结果。返回 RECONSUME_LATER 时Broker 会进行消息重试。RocketMQ 默认最大重试次数为 16 次超过后消息进入死信队列。重试机制要点Broker 为每个消费者组自动生成对应的重试 Topic%RETRY%ConsumeGroup避免重试消息阻塞原队列。消费者组内任意实例均可接收重试消息因此同一消费者组内的订阅主题和消费逻辑必须保持一致。Broker 只根据消费者返回的状态判断处理结果因此消费者业务逻辑应尽量使用同步实现处理完成后再返回状态。消费起始位点消费者组可通过 ConsumeFromWhere 枚举指定消费起点包括 CONSUME_FROM_LAST_OFFSET从队列最后一条开始、CONSUME_FROM_FIRST_OFFSET从队列第一条开始、CONSUME_FROM_TIMESTAMP从某个时间点开始配合 ConsumerTimestamp 属性。3. 广播消息广播模式与集群模式是消费者端最基本的两种模式。集群模式下一个消息只会被消费者组中的一个实例消费一次广播模式下消息会推送给所有消费者实例处理。实现思路默认集群模式下Broker 为每个 ConsumerGroup 维护统一 Offset广播模式的本质是将 Offset 转移到 Consumer 端自行保管Broker 推送消息时不再关注 ConsumerGroup。注意点Broker 端不维护消费进度消费者处理失败无法进行消息重试。Consumer 端维护 Offset 用于服务重启时按上次消费进度继续处理若 Offset 丢失消费者仍可拉取消息但无法获取丢失 Offset 期间的消息。4. 过滤消息同一 Topic 下存在多种消息时消费者可通过过滤机制只关注自己感兴趣的消息。Tag 简单过滤生产者发送消息时设置 Tag 属性消费者通过 consumer.subscribe(Topic, TagA) 订阅指定 Tag。多个 Tag 可用双竖线||连接星号*匹配所有。String[] tags new String[] {TagA, TagB, TagC}; for (int i 0; i 15; i) { Message msg new Message(TagFilterTest,tags[i % tags.length], Hello world.getBytes(RemotingHelper.DEFAULT_CHARSET)); SendResult sendResult producer.send(msg); System.out.printf(%s%n, sendResult); }#消费者端就可以通过这个Tag属性订阅⾃⼰感兴趣的内容Consumer就只会出处理TagA的消息 consumer.subscribe(TagFilterTest, TagA);SQL 过滤通过 Tag 和用户自定义属性以标准 SQL 方式过滤。支持数值比较、、、、BETWEEN、、字符比较、、IN、IS NULL / IS NOT NULL、逻辑符号 AND / OR / NOT。使用自定义参数过滤需在 Broker 端将 enablePropertyFilter 设置为 true默认 false。RocketMQ 通过 ANTLR 引擎解析 SQL 语句。String[] tags new String[] {TagA, TagB, TagC}; for (int i 0; i 15; i) { Message msg new Message(SqlFilterTest,tags[i % tags.length], (Hello RocketMQ i).getBytes(RemotingHelper.DEFAULT_CHARSET) ); msg.putUserProperty(a, String.valueOf(i)); SendResult sendResult producer.send(msg); System.out.printf(%s%n, sendResult); }#消费者端在进⾏过滤时可以指定⼀个标准的SQL语句定制复杂的过滤规则 consumer.subscribe(SqlFilterTest, MessageSelector.bySql((TAGS is not null and TAGS in (TagA, TagB)) and (a is not null and a between 0 and 3))); #注意如果需要使⽤⾃定义参数进⾏过滤需要在Broker端将参数enablePropertyFilter设置成true。这个参数默认是false。Tag属性的处理⽐较简单就是直接匹配。⽽SQL语句的处理会⽐较麻烦⼀点。RocketMQ也是通过ANLTR引擎来解析SQL语句然后再进⾏消息过滤的。 ANLTR是⼀个开源的SQL语句解析框架。很多开源产品都在使⽤ANLTR来解析SQL语句。⽐如ShardingSphereFlink等。注意点消息过滤在 Broker 端完成只推送消费者感兴趣的消息减少网络 IO但会加大服务端压力因此过滤逻辑越简单越好。消费者不感兴趣的消息并不直接丢弃通常需要同一消费者组内其他消费者实例消费剩余消息若一直没有其他消费者Broker 仍会推进 Offset。5. 顺序消息机制#生产者示例代码 for (int i 0; i 10; i) { int orderId i; for(int j 0 ; j 5 ; j ){ Message msg new Message(OrderTopicTest, order_orderId, KEY orderId, (order_orderId step j).getBytes(RemotingHelper.DEFAULT_CHARSET)); SendResult sendResult producer.send(msg, new MessageQueueSelector() { Override public MessageQueue select(ListMessageQueue mqs, Message msg, Object arg) { Integer id (Integer) arg; int index id % mqs.size(); return mqs.get(index); } }, orderId); System.out.printf(%s%n, sendResult); } }通过MessageSelector将orderId相同的消息都转发到同一个MessageQueue中。​ 注入一个MessageListenerOrderly实现。​ #消费者示例代码 consumer.registerMessageListener(new MessageListenerOrderly() { Override public ConsumeOrderlyStatus consumeMessage(ListMessageExt msgs, ConsumeOrderlyContext context) { context.setAutoCommit(true); for(MessageExt msg:msgs){ System.out.println(收到消息内容 new String(msg.getBody())); } return ConsumeOrderlyStatus.SUCCESS; } });实现思路顺序消息用于保证同一业务如订单的多个步骤按顺序处理。核心思路是通过MessageQueueSelector 将同一业务的消息发送到同一个 队列MessageQueue​ 如果不指定MessageSelector对象那么生产者会采用轮询的方式将多条消息依次发送到不同的MessageQueue上消费者端使用顺序消费监听器MessageListenerOrderly 顺序消费会给一个MessageQueue加锁拿到MessageQueue上所有的消息然后再去读取下一个MessageQueue的消息。生产者端通过自定义队列选择器如按订单 ID 取模将同一订单的消息路由到同一队列消费者端使用顺序消费监听器保证消息按序处理。注意点​ 1、理解局部有序与全局有序。大部分业务场景下我们需要的其实是局部有序。如果要保持全局有序那就只保留一个MessageQueue。性能显然非常低。​ 2、生产者端尽可能将有序消息打散到不同的MessageQueue上避免过于集中导致数据热点竞争。​ 3、消费者端只进行有限次数的重试。如果一条消息处理失败RocketMQ会将后续消息阻塞住让消费者进行重试。但是如果消费者一直处理失败超过最大重试次数那么RocketMQ就会跳过这一条消息处理后面的消息这会造成消息乱序。​ 4、消费者端如果确实处理逻辑中出现问题不建议抛出异常可以返回ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT作为替代。6. 延迟消息延迟消息是指消息发送后经过指定延迟时间才能被消费者消费。RocketMQ 通过延迟级别messageDelayLevel实现默认延迟级别为 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h。生产者发送消息时指定延迟级别即可。#示例代码 // 指定固定的延迟级别 Message message new Message(TOPIC, (Hello scheduled message i).getBytes(StandardCharsets.UTF_8)); message.setDelayTimeLevel(3); //10秒之后发送 // 指定消息发送时间 Message message new Message(TOPIC, (Hello scheduled message i).getBytes(StandardCharsets.UTF_8)); message.setDeliverTimeMs(System.currentTimeMillis() 10_000L); //指定10秒之后的时间点7. 批量消息批量消息允许生产者一次发送多条消息减少网络开销提升发送效率。批量发送时需注意单批消息总大小不能超过 Broker 限制默认 4MB。注意点 批量消息的使⽤⾮常简单但是要注意RocketMQ做了限制。同⼀批消息的Topic必须相同另外不⽀持延迟消息。 还有批量消息的⼤⼩不要超过1M如果太⼤就需要⾃⾏分割。#示例代码 ListMessage messages new ArrayList(MESSAGE_COUNT); for (int i 0; i MESSAGE_COUNT; i) { messages.add(new Message(TOPIC, TAG, OrderID i, (Hello world i).getBytes(StandardCharsets.UTF_8))); } //split the large batch into small ones: ListSplitter splitter new ListSplitter(messages); while (splitter.hasNext()) { ListMessage listItem splitter.next(); SendResult sendResult producer.send(listItem); System.out.printf(%s, sendResult); }8. 事务消息事务消息用于解决分布式事务问题保证本地事务与消息发送的最终一致性。核心流程包括发送半消息half message、执行本地事务、提交或回滚事务状态、Broker 根据事务状态决定消息是否可见。RocketMQ 通过事务回查机制保证最终一致性。应用场景以电商为例用户支付订单这一核心操作的同时会涉及到下游物流发货、积分变更、购物车状态清空等多个子系统的变更。这种场景非常适合使用RocketMQ的解耦功能来进行串联。事务的安全性即要保证相关联的这几个业务一定是同时成功或者同时失败的。如果要将四个服务一起作为一个分布式事务来控制可以做到但是会非常麻烦。而使用RocketMQ在中间串联了之后事情可以得到一定程度的简化。由于RocketMQ与消费者端有失败重试机制所以只要消息成功发送到RocketMQ了那么可以认为Branch2.1Branch2.2Branch2.3这几个分支步骤是可以保证最终的数据一致性的。这样一个复杂的分布式事务问题就变成了MinBranch1和Branch2两个步骤的分布式事务问题。​ 然后在此基础上RocketMQ提出了事务消息机制采用两阶段提交的思路保证Main Branch1和Branch2之间的事务一致性。实现思路生产者将消息发送至Apache RocketMQ服务端。Apache RocketMQ服务端将消息持久化成功之后向生产者返回Ack确认消息已经发送成功此时消息被标记为暂不能投递这种状态下的消息即为半事务消息。生产者开始执行本地事务逻辑。生产者根据本地事务执行结果向服务端提交二次确认结果Commit或是Rollback服务端收到确认结果后处理逻辑如下二次确认结果为Commit服务端将半事务消息标记为可投递并投递给消费者。二次确认结果为Rollback服务端将回滚事务不会将半事务消息投递给消费者。在断网或者是生产者应用重启的特殊情况下若服务端未收到发送者提交的二次确认结果或服务端收到的二次确认结果为Unknown未知状态经过固定时间后服务端将对消息生产者即生产者集群中任一生产者实例发起消息回查。生产者收到消息回查后需要检查对应消息的本地事务执行的最终结果。生产者根据检查到的本地事务的最终状态再次提交二次确认服务端仍按照步骤4对半事务消息进行处理。​ 实现时的重点是使用RocketMQ提供的TransactionMQProducer事务生产者在TransactionMQProducer中注入一个TransactionListener事务监听器来执行本地事务以及后续对本地事务的检查。注意点​ 1、半消息是对消费者不可见的一种消息。实际上RocketMQ的做法是将消息转到了一个系统TopicRMQ_SYS_TRANS_HALF_TOPIC。​ 2、事务消息中本地事务回查次数通过参数transactionCheckMax设定默认15次。本地事务回查的间隔通过参数transactionCheckInterval设定默认60秒。超过回查次数后消息将会被丢弃。​ 3、其实了解了事务消息的机制后在具体执行时可以对事务流程进行适当的调整。9. ACL 权限控制机制RocketMQ 支持基于 AccessKey / SecretKey 的 ACL 权限控制可对生产者和消费者的 Topic 访问权限进行精细化管控保障消息安全。三、SpringBoot 整合 RocketMQ1. 快速实战SpringBoot 整合 RocketMQ 时需引入 rocketmq-spring-boot-starter 依赖并注意版本兼容性SpringBoot 3.0.4 要求 JDK 17 以上。核心配置rocketmq.name-server192.168.65.112:9876 rocketmq.producer.groupspringBootGroup rocketmq.consumer.grouptestGroup server.port9000生产者通过注入 RocketMQTemplate 发送消息消费者通过 RocketMQMessageListener 注解声明实现 RocketMQListener 接口处理消息。2. 处理各种消息类型一个 RocketMQTemplate 实例只能包含一个生产者只能往一个 Topic 发送消息需要往其他 Topic 发送时通过 ExtRocketMQTemplateConfiguration() 注解声明子类实例。事务消息的事务监听器通过 RocketMQTransactionListener 注解注入 Spring 容器可通过 rocketMQTemplateBeanName 属性指向具体的 RocketMQTemplate 子类。3. 实现原理RocketMQTemplate注入过程参见 RocketMQAutoConfiguration 自动配置类。Push 模式消费者RocketMQMessageListener 注解的处理入口在 ListenerContainerConfiguration 类中它注入 RocketMQMessageListenerContainerRegistrar 对象启动时调用 startContainer 方法创建并启动 DefaultRocketMQListenerContainer对 DefaultMQPushConsumer 的封装。Pull 模式通过在 RocketMQTemplate 中注入 DefaultLitePullConsumer 实例实现通过 template 的 receive 方法调用 poll 方法主动拉取消息。实际开发中拉模式用得较少。四、RocketMQ 客户端注意事项1. 消息的 ID、Key 和 TagMessageIdRocketMQ 内部为每条消息分配的唯一次索引由 Broker 端生成。复杂业务场景下不建议作为业务唯一索引。KeyMessage 中的补充信息本质是消息属性。建议添加业务唯一性数据如订单 IDRocketMQ 基于 Key 实现消息溯源、消息压缩等功能。Tag用于消息过滤服务端将 Tag 的 hashCode 写入 ConsumeQueue 索引中过滤性能非常高是官方推荐的最佳实践。2. 最佳实践一个应用尽可能用一个 Topic消息子类型用 Tags 标识。Topic 过多会加大元数据维护消耗需合理分配。尽量直接使用 Tag 过滤避免复杂 SQL 过滤减少 Broker 端压力。3. 消费者端幂等控制MQ 系统消息幂等有三种语义at most once最多一次、at least once至少一次、exactly once刚好一次。RocketMQ 只能保证 at least once无法保证 exactly once因此需要业务系统自行保证幂等性。消息重复的三种场景发送时重复消息已持久化但应答失败生产者重发导致重复。投递时重复消息已处理但应答时网络闪断服务端重投导致重复。负载均衡时重复Broker 或客户端重启、扩容缩容触发 Rebalance 导致重复。处理方式使用 MessageId 或业务唯一标识如订单 ID通过 Key 传递进行幂等判断。4. 关注错误消息重试消费者处理失败后Broker 会将消息移入重试队列%RETRY%ConsumeGroup。重试队列出现大量消息意味着消费者运行异常需及时干预。默认最大重试 16 次重试间隔与延迟级别对应取后 16 级可通过 consumer.setMaxReconsumeTimes(20) 定制超过 16 次后重试间隔均为 2 小时。最大重试次数设置对相同 GroupID 下所有 Consumer 实例有效最后启动的 Consumer 会覆盖之前的配置。5. 手动处理死信队列消息超过最大重试次数后进入死信队列%DLQ%ConsumeGroup。死信队列特征一个死信队列对应一个 ConsumeGroup而非某个消费者实例。没有产生死信消息的 ConsumeGroup 不会创建死信队列。死信队列包含该 ConsumeGroup 的所有死信消息不区分 Topic。死信队列中的消息不会再被消费者正常消费。死信队列有效期与正常消息相同默认 3 天对应 broker.conf 的 fileReservedTime 属性。注意默认创建的死信队列权限 perm 为 2禁读需手动改为 6可读可写才能被消费可通过 mqadmin 或 Web 控制台操作。五、面试常见问题与解答Q1RocketMQ 客户端编程的基本流程是什么答生产者流程创建 DefaultMQProducer 并指定生产者组名、指定 NameServer 地址、启动 producer、创建 Message 对象指定 Topic、Tag、消息体、发送消息、关闭 producer。消费者流程创建 DefaultMQPushConsumer 并指定消费者组名、指定 NameServer 地址、订阅 Topic 和 Tag、注册消息监听回调、启动 consumer。核心是 NameServer 路由发现客户端无需指定具体 Broker 地址。Q2RocketMQ 三种发送方式有什么区别如何选择答单向发送sendOneway只管发送不关心结果效率最高适合允许丢失的场景如日志同步发送send阻塞等待 Broker 返回 SendResult安全性最高但效率低异步发送send SendCallback注册回调不阻塞主线程兼顾安全性与高吞吐。选择依据业务场景电商下单等数据安全要求高的场景优先同步或异步发送高并发场景优先异步发送。Q3RocketMQ 如何保证消息不丢失答生产端通过消息确认加多次重试机制保证消息发送到 Broker消费端通过状态确认机制CONSUME_SUCCESS / RECONSUME_LATER保证消息被正常处理失败消息进入重试队列超过最大重试次数进入死信队列。但 RocketMQ 只能保证 at least once无法保证 exactly once业务系统需自行保证幂等。Q4广播模式和集群模式有什么区别答集群模式下一个消息只会被消费者组中的一个实例消费一次Broker 为每个 ConsumerGroup 维护统一 Offset广播模式下消息推送给所有消费者实例处理Offset 转移到 Consumer 端自行保管。广播模式缺点Broker 不维护消费进度处理失败无法重试Offset 丢失后无法获取丢失期间的消息。Q5RocketMQ 如何实现顺序消息答生产者端通过 MessageQueueSelector 将同一业务的消息如按订单 ID 取模路由到同一个 MessageQueue保证消息进入同一队列消费者端使用 MessageListenerOrderly 顺序消费监听器保证同一队列的消息按序处理。这样即可保证同一订单的业务步骤顺序不乱。Q6RocketMQ 如何保证消息幂等答RocketMQ 只能保证 at least once消息可能重复投递发送时重复、投递时重复、负载均衡时重复。业务系统需自行保证幂等使用 MessageId 或业务唯一标识如订单 ID通过 Key 传递进行判断例如在数据库中记录已处理的消息 ID重复消息直接跳过。Q7RocketMQ 消息重试机制是怎样的答消费者返回 RECONSUME_LATER 时Broker 将消息移入重试队列%RETRY%ConsumeGroup避免阻塞原队列。默认最大重试 16 次重试间隔与延迟级别对应1s 到 2h 递增。超过最大重试次数后消息进入死信队列%DLQ%ConsumeGroup需人工介入处理。可通过 setMaxReconsumeTimes 定制重试次数超过 16 次后间隔均为 2 小时。Q8什么是死信队列如何处理答死信队列是消息超过最大重试次数后进入的特殊队列名称格式为 %DLQ%ConsumeGroup。一个死信队列对应一个 ConsumeGroup包含该组所有死信消息。死信消息不会被正常消费需人工排查错误原因可转发到正常 Topic 重新消费或丢弃。注意默认死信队列权限为禁读perm2需手动改为 6 才能读取。Q9RocketMQ 事务消息的原理是什么答事务消息用于解决分布式事务问题。流程生产者发送半消息half message到 BrokerBroker 暂存消息生产者执行本地事务根据本地事务结果向 Broker 提交或回滚事务状态Broker 根据状态决定消息是否可见。若生产者未及时提交状态Broker 会进行事务回查保证最终一致性。