rocketmq

RocketMQ 分布式消息队列部署与运维

补充说明

rocketmq 是 Apache 的分布式消息中间件，核心由轻量级 NameServer、负责存储与投递的 Broker、以及生产者和消费者客户端组成。Topic 在 Broker 上被切成多个 Queue（分区），同一 Queue 内有序，消费者组按 Queue 分配消费，因此 Queue 数决定消费并行度上限。

它擅长削峰、异步解耦、顺序消息、事务消息和延时/重试投递，Java 生态接入成本低。与 Kafka 相比，RocketMQ 提供原生的顺序、事务和定时消息，不依赖分区再平衡扩消费；与 RabbitMQ 相比，它更适合高吞吐、长时间堆积和分区顺序场景。

4.x 使用 Remoting 协议（10911），5.x 增加 Proxy 与 gRPC 接入，但 NameServer/Broker/Queue 的模型不变。生产上要显式建 Topic、关闭自动创建、按消费者并行度规划 Queue 数，并持续监控堆积、重试和 DLQ。

常用示例

export NAMESRV_ADDR=127.0.0.1:9876
nohup sh bin/mqnamesrv &
nohup sh bin/mqbroker -n 127.0.0.1:9876 -c conf/broker.conf &
sh bin/mqshutdown broker
sh bin/mqshutdown namesrv
sh bin/mqadmin clusterList -n 127.0.0.1:9876
sh bin/mqadmin brokerStatus -n 127.0.0.1:9876 -b 127.0.0.1:10911
sh bin/mqadmin topicStatus -n 127.0.0.1:9876 -t OrderTopic
sh bin/mqadmin topicRoute -n 127.0.0.1:9876 -t OrderTopic
sh bin/mqadmin updateTopic -n 127.0.0.1:9876 -c DefaultCluster -t OrderTopic -r 8 -w 8
sh bin/mqadmin consumerProgress -n 127.0.0.1:9876 -g order-consumer
sh bin/mqadmin resetOffsetByTime -n 127.0.0.1:9876 -g order-consumer -t OrderTopic -s 2026-01-01#00:00:00:000
sh bin/mqadmin sendMessage -n 127.0.0.1:9876 -t OrderTopic -p 'hello'

最小事务消息与顺序消费写法：

TransactionMQProducer txProducer = new TransactionMQProducer("tx-producer");
txProducer.setNamesrvAddr("127.0.0.1:9876");
txProducer.setTransactionListener(new TransactionListener() {
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        return LocalTransactionState.COMMIT_MESSAGE; // 本地事务成功才提交
    }
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        return LocalTransactionState.COMMIT_MESSAGE; // 回查必须幂等
    }
});
txProducer.sendMessageInTransaction(new Message("OrderTopic", "body".getBytes()), null);

consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
    // 同一 Queue 串行消费，返回 SUCCESS 或 SUSPEND_CURRENT_QUEUE_A_MOMENT
    return ConsumeOrderlyStatus.SUCCESS;
});
consumer.subscribe("OrderTopic", "*");
consumer.start();

常见配置

brokerClusterName=DefaultCluster
brokerName=broker-a
brokerId=0
brokerRole=SYNC_MASTER
flushDiskType=ASYNC_FLUSH
namesrvAddr=127.0.0.1:9876
deleteWhen=04
fileReservedTime=72
defaultTopicQueueNums=8
autoCreateTopicEnable=false
transactionTimeOut=6000
transactionCheckMax=15

brokerId=0 为主节点，大于 0 为从节点；brokerRole 取 SYNC_MASTER、ASYNC_MASTER 或 SLAVE。deleteWhen 与 fileReservedTime 默认分别为 04 点和 72 小时；transactionTimeOut 默认 6000ms、transactionCheckMax 默认 15；defaultTopicQueueNums 在 5.x 源码中默认为 8，旧版为 4，建议显式指定。

易错点与排错

• No route info of this topic：NameServer 地址错、Topic 未创建或 Broker 未注册。先查 clusterList、topicRoute，再确认 9876/10911 端口和防火墙。
• 自动建 Topic 会让不同生产者用不同 Queue 数并绕过治理；生产设置 autoCreateTopicEnable=false，用 updateTopic 显式创建。
• brokerId 冲突或从节点也配成 0 会导致主从关系异常；主节点固定 0，从节点使用大于 0 的唯一值。
• SYNC_MASTER 下从节点离线可能让写入超时；用 brokerStatus 确认同步状态，再决定是否临时降级为 ASYNC_MASTER。
• 消费堆积看 consumerProgress 的 Diff Total，不能只看 Broker 磁盘；消费者实例数超过 Queue 数时，多出的实例空闲。
• 重试消息进入 %RETRY%<consumerGroup>，超过最大次数进入 %DLQ%<consumerGroup>；maxReconsumeTimes=-1 时并发消费重试 16 次，顺序消费会持续重试。
• 顺序消息要求同一业务键经 MessageQueueSelector 落到同一 Queue，并使用 MessageListenerOrderly；换成并发监听器会丢顺序。
• 事务消息先写半消息，本地事务结果未知时 Broker 定时回查；回查必须幂等，UNKNOW 超过 transactionCheckMax 后消息会被丢弃或进入 DLQ。
• 默认最大消息 4MB；大消息先压缩或写对象存储，只发引用，避免 Broker 内存、网络和 CommitLog 被撑爆。
• 重复消费是 at-least-once 的常态；用业务唯一键或去重表保证幂等，不要假设 Exactly-Once。
• 磁盘使用率超过阈值会拒绝写入；清理过期 CommitLog 前确认消费进度，并监控数据盘和 diskMaxUsedSpaceRatio。
• 5.x 客户端走 Proxy/gRPC（常见 8081），4.x 客户端走 Remoting（10911）；接入点与端口混用会连接失败。

相关命令

• rabbitmq — 队列模型与确认机制对照
• kafka-broker — 分区日志型 Broker 运维
• rabbitmq-queue-backlog — 消息堆积的通用排查
• kafka-vs-pulsar — 消息中间件选型对照
• middleware-mental-model — 队列、分区、副本与背压

读完自测

1. NameServer、Broker、Topic/Queue 各自负责什么？
2. 顺序消息为什么要求同一业务键进入同一 Queue？
3. 事务消息的本地事务结果未知时，Broker 如何确认？