Consumer 的启动和 Producer 非常相似但多了几个消费特有的步骤创建 Consumer 实例设置 ConsumerGroup 名称同一业务/应用的消费者属于同一个 Group设置 NameServer 地址列表配置消费模式集群消费/广播消费和消费位点从哪开始消费订阅 Topic指定 Topic 和 Tag 过滤表达式注册消息监听器MessageListener处理收到的消息调用 start 方法启动 Consumer初始化 MQClientInstance和 Producer 共用同一个实例启动 Netty 客户端连接 NameServer 和 Broker从 NameServer 拉取订阅 Topic 的路由信息触发首次 Rebalance分配 Queue 给当前 Consumer启动定时任务每 30 秒更新路由 定时 RebalanceConsumer 就绪开始消费消息和 Producer 启动的关键区别订阅 TopicConsumer 启动时需要明确告诉 Broker“我要消费哪个 Topic”注册监听器Consumer 需要注册 MessageListener处理消息的业务逻辑就写在这里面触发 Rebalance这是消费特有的环节——Consumer 启动后会立即进行一次 Rebalance确定自己负责消费哪些 Queue消费位点Consumer 需要知道从哪个位置开始消费从最新的开始还是从头开始还是从上次消费的进度继续Pull 模式与 Push 模式的本质区别这是初学者最容易混淆的概念。表面上 RocketMQ 支持两种消费模式——Pull 和 Push但在底层实现上它们都是 Pull 模式。本质区别谁来控制拉取频率模式 控制权 实现方式 用户感知Pull 用户自己控制 用户主动调用 pull() 方法 用户完全掌控拉取时机和批量大小Push Broker“推” 底层依然是不断拉取但被封装成“推送”的感觉 用户只需实现监听器消息来了就被调用Push 模式底层也是 Pull注册监听器后台线程循环拉取返回消息自动回调用户代码ConsumerBroker用户无感知消息来了自动回调监听器底层是长轮询或定时拉取Pull 模式主动调用 pull拉取请求返回消息返回消息用户代码ConsumerBroker用户自己控制什么时候拉、拉多少条、间隔多久Push 模式的底层实现RocketMQ 的 Push 模式DefaultMQPushConsumer在底层其实是一个不断循环的 Pull 线程Consumer 启动后内部有一个 PullMessageService 线程这个线程不断从 Broker 拉取消息每次拉取一批拉到的消息提交到线程池处理处理完成后继续拉取下一批所以本质上Push 模式 自动化的 Pull 多线程消费。长轮询机制的原理与实现既然底层是 Pull那 Broker 怎么做到“消息一来就推送给消费者”呢答案就是长轮询Long Polling。什么是长轮询传统的 Pull 是消费者定时去 Broker 问“有新消息吗”——如果一直没有消息每次请求都是空跑浪费资源。长轮询的优化是如果 Broker 没有新消息请求不立即返回而是“挂起”等待一段时间默认 15 秒。如果在这段时间内有新消息到达立即返回如果超时了还没消息返回空响应。BrokerConsumerBrokerConsumer有消息时的长轮询✅ 快速响应没有消息时的长轮询等待新消息最长 15 秒alt[15 秒内有新消息到达][15 秒超时]拉取请求offset 100检查 ConsumeQueue发现 offset100 处有消息立即返回消息拉取请求offset 200检查 ConsumeQueueoffset200 处暂无消息将请求挂起放入 pending 队列新消息写入触发唤醒返回新消息返回空响应继续发起下一次长轮询长轮询的好处实时性消息一到达挂起的请求立即被唤醒Consumer 几乎实时收到消息节省资源相比短轮询每秒一次长轮询大大减少了空请求的数量保护 Broker通过超时机制避免请求无限挂起导致资源泄漏 小贴士长轮询的超时时间由 pullRequestHoldService 控制默认 15 秒。这个值不是越大越好——太大会占用 Broker 的线程资源太小又会增加空请求的频率。消费者组Consumer Group的概念与作用什么是消费者组消费者组是一组具有相同消费逻辑的 Consumer 实例的集合。它们共同消费同一个 Topic 的消息。Topic: order_topic消费者组order_consumer_groupConsumer 实例 1部署在服务器 AConsumer 实例 2部署在服务器 BConsumer 实例 3部署在服务器 CQueue 0Queue 1Queue 2Queue 3同一个 Group 内的 Consumer 实例共同消费 Topic 的所有消息集群消费不同实例之间负载均衡每个 Queue 只分配给一个实例消费者组的核心作用水平扩展增加 Consumer 实例就能线性提升消费吞吐量消费进度管理同一个 Group 共享一份消费进度Offset任意实例宕机后其他实例可以接替继续消费负载均衡Group 内的实例自动分配 Queue协同工作 核心约束同一个消费者组内的 Consumer必须订阅相同的 Topic 和 Tag。如果订阅不同会导致 Rebalance 混乱和消息丢失。集群消费与广播消费RocketMQ 提供了两种消费模式它们的区别非常核心集群消费Clustering一个 Topic 下的消息只被同一个 Group 中的一个 Consumer 消费一次多个 Consumer 实例之间负载均衡分配 Queue这是 RocketMQ 的默认模式消费者组Topic: order_topic消费消费消费消费消息1Queue0消息2Queue1消息3Queue2消息4Queue3Consumer AConsumer B每条消息只被消费一次多个 Consumer 分摊负载广播消费Broadcasting一个 Topic 下的消息被同一个 Group 中的每个 Consumer 都消费一次每个 Consumer 独立消费所有 Queue 的消息每个 Consumer 独立管理自己的 Offset消费者组Topic: order_topic消费消费消费消费消息1Queue0消息2Queue1Consumer AConsumer B每条消息被每个 Consumer 都消费一次每条消息被消费 N 次N 消费者数量对比维度 集群消费 广播消费消息投递 每条消息只投递给一个 Consumer 每条消息投递给所有 ConsumerOffset 管理 Group 共享一份 每个实例各自独立适用场景 负载均衡、水平扩展 每个节点都需要全量数据如缓存刷新是否默认 ✅ 默认 ❌ 需显式配置Rebalance 机制的触发时机与算法实现Rebalance 是什么Rebalance重平衡是指消费者组内的 Queue 分配发生变化时重新分配 Queue 给各个 Consumer 实例的过程。触发时机是否是否是否是否触发 Rebalance消费者实例启动或关闭执行 RebalanceTopic 的 Queue 数量发生变化消费者组订阅信息变化Broker 变化新增/宕机不触发重新分配所有 Queue具体的触发场景包括Consumer 启动或关闭Group 内成员变化需要重新分配