RocketMQ 是 Java 业务系统里最常见的一档 MQ：吞吐够用、事务消息和延迟消息原生支持、控制台好用。这篇按「装起来 → 用起来 → 出问题怎么查」整理。
一、架构与存储
角色职责：
- NameServer：无状态的路由注册中心，Broker 定时上报，Producer/Consumer 拉取路由。节点间不通信，所以部署 2–3 台互不影响。
- Broker：存消息，分 Master/Slave。Master 负责读写，Slave 只做备份（可开从读）。
- Producer：从 NameServer 拿路由，按策略选队列发送。
- Consumer：从 NameServer 拿路由，拉取消息。消费进度（offset）保存在 Broker。
二、部署（Docker 快速起一套）
docker run -d --name rmqnamesrv -p 9876:9876 \
apache/rocketmq:5.1.4 sh mqnamesrv
docker run -d --name rmqbroker -p 10911:10911 -p 10909:10909 \
-e "NAMESRV_ADDR=rmqnamesrv:9876" \
-v /data/rocketmq/store:/home/rocketmq/store \
apache/rocketmq:5.1.4 sh mqbroker -c /home/rocketmq/rocketmq-5.1.4/conf/broker.conf
broker.conf 生产建议：
brokerClusterName=DefaultCluster
brokerName=broker-a
brokerId=0 # 0=Master，>0=Slave
# 刷盘：SYNC_FLUSH 更稳，ASYNC_FLUSH 更快
flushDiskType=ASYNC_FLUSH
# 主从同步：SYNC_MASTER 不丢消息，ASYNC_MASTER 性能更好
brokerRole=ASYNC_MASTER
# 消息保留时间（默认 72 小时，堆积场景要调大）
fileReservedTime=72
# 存储根路径要挂到数据盘
storePathRootDir=/data/rocketmq/store
选型建议：金融/交易类用 SYNC_FLUSH + SYNC_MASTER，一般业务用 ASYNC_FLUSH + SYNC_MASTER，日志类可以都异步。
三、四类消息怎么用
1. 普通消息（同步 / 异步 / 单向）
// 同步：等 Broker 返回，可靠性最高，业务消息默认用它
SendResult r = producer.send(msg);
// 异步：高吞吐场景，回调里处理失败
producer.send(msg, new SendCallback() {
public void onSuccess(SendResult r) {}
public void onException(Throwable e) { /* 补偿：落库重试 */ }
});
// 单向：发了不管，只适合日志这类可丢的
producer.sendOneway(msg);
2. 顺序消息
同一业务键（如订单号）发到同一队列 + 消费端单线程消费该队列：
producer.send(msg, new MessageQueueSelector() {
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
long orderId = (Long) arg;
return mqs.get((int) (orderId % mqs.size()));
}
}, orderId);
// 消费端必须用 MessageListenerOrderly
consumer.registerMessageListener(new MessageListenerOrderly() {
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext ctx) {
return ConsumeOrderlyStatus.SUCCESS; // 失败用 SUSPEND_CURRENT_QUEUE_A_MOMENT
}
});
注意：顺序消息不能跳过失败的消息，一条卡住整条队列就停了。所以顺序消息一定要有最大重试次数 + 失败落库的兜底。
3. 事务消息
TransactionMQProducer producer = new TransactionMQProducer("group");
producer.setTransactionListener(new TransactionListener() {
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
try {
orderService.create((OrderDTO) arg); // 本地事务
return LocalTransactionState.COMMIT_MESSAGE;
} catch (Exception e) {
return LocalTransactionState.ROLLBACK_MESSAGE;
}
}
// ★ 网络抖动没收到 commit 时，Broker 会回查，必须实现
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
return orderService.exists(msg.getKeys())
? LocalTransactionState.COMMIT_MESSAGE
: LocalTransactionState.UNKNOW; // 还不清楚就返回 UNKNOW，稍后再查
}
});
4. 延迟消息
RocketMQ 的延迟是固定 18 个等级，不支持任意时间：
msg.setDelayTimeLevel(3); // 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
要任意时间延迟，只能用时间轮 + 定时扫描，或用 RocketMQ 5.x 的「定时消息」（支持精确到秒）。
四、消费端要点
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order_consumer_group");
consumer.setNamesrvAddr("10.0.0.10:9876");
consumer.subscribe("order_topic", "*");
consumer.setConsumeThreadMin(20);
consumer.setConsumeThreadMax(64);
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, ctx) -> {
for (MessageExt m : msgs) {
try {
bizService.handle(m);
} catch (BizException e) {
log.warn("业务失败，不再重试 msgId={}", m.getMsgId(), e);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; // 不可重试直接确认
} catch (Exception e) {
log.error("消费异常，将重试 msgId={}", m.getMsgId(), e);
return ConsumeConcurrentlyStatus.RECONSUME_LATER; // 可重试
}
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
consumer.start();
关键区分：CONSUME_SUCCESS 是「确认成功，不再投递」；RECONSUME_LATER 是「稍后重试」。参数错误这类重试多少次都不会成功的异常，必须返回 SUCCESS 并落死信表，否则会占满重试队列。
重试机制：消费失败的消息会进 %RETRY%<group> 重试 Topic，延迟等级递增（最多 16 次），超过后进 %DLQ%<group> 死信队列。死信队列必须有人看，否则消息就真的没了。
五、运维命令清单（mqadmin）
# 查看 Topic 列表与队列分布
sh mqadmin topicList -n 127.0.0.1:9876
sh mqadmin topicRoute -t order_topic -n 127.0.0.1:9876
# 查看消费进度（★ 排查堆积第一命令）
sh mqadmin consumerProgress -g order_consumer_group -n 127.0.0.1:9876
# 关键列：Diff Total = 堆积量
# 按消息 ID / Key 查消息
sh mqadmin queryMsgById -i 0A010001... -n 127.0.0.1:9876
sh mqadmin queryMsgByKey -t order_topic -k ORDER123 -n 127.0.0.1:9876
# 重置消费位点（重放消息，出故障后常用）
sh mqadmin resetOffsetByTime -g group -t topic -s now -n 127.0.0.1:9876
# 查看集群状态
sh mqadmin clusterList -n 127.0.0.1:9876
# 更新 broker 配置（无需重启）
sh mqadmin updateBrokerConfig -b 10.0.0.10:10911 -k fileReservedTime -v 168 -n 127.0.0.1:9876
六、堆积排查的标准动作
consumerProgress看 Diff Total —— 确认是真的堆积还是只是延迟。- 看消费 TPS 曲线 —— 是消费变慢还是生产暴涨。
- 看消费者实例的线程栈 ——
jstack里有没有大量线程卡在同一个地方（慢 SQL、HTTP 超时）。 - 看 Broker 侧 —— 磁盘 IO 是不是满了（
iostat -x 1，%util接近 100 就是瓶颈）。 - 应急：增加消费者实例（注意不能超过队列数）；或者临时把消息转存，先恢复主链路。
扩容前先看队列数：topicRoute 里 readQueueNums。消费者实例数 > 队列数时多出来的实例完全空闲。所以 Topic 建的时候队列数就要留余量（我一般配 32）。
七、生产参数与规范
| 项 | 建议 |
|---|---|
| Topic 队列数 | 按峰值 TPS / 单队列能力估算，留 2–4 倍余量 |
| 消息保留 | fileReservedTime ≥ 72h，堆积时能回放 |
| 消息体 | 控制在 1KB 以内，太大影响吞吐；大 payload 走对象存储 + 传引用 |
| Tag 使用 | 同一业务的不同事件用 Tag 区分，别建一堆 Topic |
| 消费幂等 | 用 getKeys() 里的业务唯一键做去重，不要用 msgId（重试时 msgId 不变，但重投会产生新的） |
| 监控 | 堆积量、消费 TPS、失败率、死信队列长度 —— 四项必告警 |
八、一个真实教训
某次批量生成任务（清册场景，脱敏），消费端在处理每条消息时都要查一次配置中心的地区规则，配置中心当时网络抖动，RT 从 5ms 涨到 800ms，消费能力掉了两个数量级，堆积在半小时内涨到百万。
改法是把地区规则改成本地缓存 + 变更推送，消费端不再依赖实时远程查询；同时给消费端加了「下游 RT 超过阈值就快速失败并告警」的保护 —— 宁可堆积也不要把消费者拖死。
这印证了我在 MQ 通识那篇里写的：消费端的远程调用是吞吐杀手。MQ 本身很少是瓶颈，瓶颈在消费逻辑。
一起聊聊