RocketMQ 实战：消息发送、消费与高可用架构

在分布式系统里，RocketMQ 消息队列凭借低延迟、高吞吐与原生事务消息能力，成为电商、金融核心链路的首选中间件。但不少团队在搭建与使用时踩过坑：容器化后客户端连不上、消息重复消费、顺序错乱、Broker 磁盘打满。本文从零实战，用 Docker 快速搭起可用集群，逐一打通发送、消费、顺序与事务消息，并给出生产级高可用架构与避坑清单。

一、RocketMQ 是什么，为什么选它

RocketMQ 是阿里巴巴开源、后捐赠给 Apache 的分布式消息中间件。在众多消息队列里，Kafka 以超高吞吐见长但事务能力偏弱，RabbitMQ 易用却难以支撑海量消息堆积；而 RocketMQ 在事务消息、定时/延迟消息、海量堆积与严格顺序消费之间取得了很好的平衡，因此被广泛用于交易、支付等对可靠性要求极高的核心链路。

它采用 Producer、Broker、Consumer、NameServer 四组件模型：NameServer 作为无状态路由注册中心，Broker 负责实际存储与投递，Topic 之下再细分为多个 MessageQueue 作为并行消费的基本单位。理解这四者关系，是后续所有配置与排错的基础。

二、用 Docker 五分钟搭起单机 broker

本地验证最省事的方式是用 Docker Compose 同时拉起 NameServer 与 Broker。唯一容易踩的坑是 brokerIP1 必须配置为宿主机在容器外可达的 IP——若留空或填 127.0.0.1，容器内的 Broker 向 NameServer 注册的地址客户端根本连不上，表现为「生产者能启动但 send 超时」。

version: '3.8'
services:
  namesrv:
    image: apache/rocketmq:5.3.0
    container_name: rmq-namesrv
    command: sh mqnamesrv
    ports:
      - "9876:9876"
  broker:
    image: apache/rocketmq:5.3.0
    container_name: rmq-broker
    command: sh mqbroker -n namesrv:9876
    environment:
      - JAVA_OPT_EXT=-Xmx512m
    volumes:
      - ./broker.conf:/home/rocketmq/rocketmq-5.3.0/conf/broker.conf
    ports:
      - "10911:10911"
      - "8080:8080"
    depends_on:
      - namesrv

# broker.conf 关键配置
brokerClusterName=DefaultCluster
brokerName=broker-a
brokerId=0
brokerIP1=宿主机可达IP   # 改成你本机局域网/公网可达 IP，否则容器外客户端连不上
listenPort=10911
storePathRootDir=/home/rocketmq/store

启动后执行 docker logs rmq-broker 看到 「The broker[broker-a, 宿主机IP:10911] boot success」即代表就绪。生产环境建议把存储目录挂到 SSD，并在 Docker 编排层面为 Broker 单独分配资源，避免与 NameServer 争抢。

三、三种发送方式：同步、异步与单向

RocketMQ 提供三种发送语义，选型取决于你对可靠性与吞吐的权衡。核心链路（下单、扣款）务必用同步发送拿到明确结果；批量通知、耗时操作可用异步发送提升吞吐；日志、埋点等可丢失数据用单向发送追求极致性能。

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

DefaultMQProducer producer = new DefaultMQProducer("order_producer_group");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();

// 1) 同步发送：等待 Broker 确认，可靠性最高
Message msg = new Message("order_topic", "下单".getBytes());
SendResult sync = producer.send(msg);
System.out.println("sync status=" + sync.getSendStatus());

// 2) 异步发送：回调通知，吞吐更高
producer.send(msg, (sendResult, e) -> {
    if (e == null) System.out.println("async ok=" + sendResult.getMsgId());
    else e.printStackTrace();
});

// 3) 单向发送：不等待结果，吞吐最高但可能丢失（日志/埋点）
producer.sendOneway(msg);

四、消费模式：集群消费、广播与推拉模型

消费侧首先要分清集群消费与广播消费：同一 ConsumerGroup 下，集群消费会把消息平均分给组内实例，一条消息只被消费一次；广播消费则每个实例都收到全量消息，常用于本地缓存刷新这类「人人都要」的场景。所谓 PushConsumer 本质也是 Broker 长轮询，并非真的服务端推送。

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order_consumer_group");
consumer.setNamesrvAddr("127.0.0.1:9876");
consumer.subscribe("order_topic", "*");
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
    for (MessageExt m : msgs) {
        System.out.println("recv=" + new String(m.getBody()));
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
consumer.start();

五、顺序消息与事务消息

严格顺序场景（如订单状态流转：创建→支付→发货）必须保证同一业务主键的消息落到同一个 MessageQueue，RocketMQ 仅保证单队列内有序。通过自定义 MessageQueueSelector 按订单 id 取模即可实现。

// 顺序消息：同一业务主键（如订单 id）路由到同一 MessageQueue
producer.send(msg, (mqs, m, arg) -> {
    int orderId = (int) arg;
    return mqs.get(orderId % mqs.size());
}, orderId);

事务消息解决的是「本地数据库事务」与「消息发送」的一致性：先发半消息（对消费者不可见），执行本地事务后提交或回滚，Broker 还会对悬而未决的半消息发起回查。这是实现「下单后扣库存、发积分」这类跨系统一致性的利器。

TransactionMQProducer txProducer = new TransactionMQProducer("tx_group");
txProducer.setNamesrvAddr("127.0.0.1:9876");
txProducer.setTransactionListener(new TransactionListener() {
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            orderService.create((Order) arg);            // 本地事务：落库
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        return orderService.exists(msg.getKeys())
               ? LocalTransactionState.COMMIT_MESSAGE
               : LocalTransactionState.UNKNOW;
    }
});
txProducer.start();

六、重试、死信与消费幂等

消费失败在集群模式下会自动重试（默认最多 16 次，间隔随次数递增），达到上限后消息进入 %DLQ% 死信队列，需要人工或定时任务补偿。广播模式则不重试、不进死信，每条实例自行负责。

网络重试与 Broker 重复投递无法完全避免，因此消费侧必须做幂等。最朴素可靠的做法是用消息唯一 key（如 msgId 或业务主键）配合 Redis 的 SETNX 去重，已处理过直接返回成功。

// 用消息唯一 key 做去重，保证重试/重复投递只处理一次（见 Redis 实战：https://fsdata.site/?p=266）
String key = "mq:idemp:" + msg.getMsgId();
Boolean first = redisTemplate.opsForValue().setIfAbsent(key, "1", Duration.ofHours(24));
if (Boolean.FALSE.equals(first)) {
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; // 已处理，直接 ack
}
businessService.handle(msg);

七、高可用架构：主从、Dledger 与多副本

单机 Master 只能用于本地测试。生产环境要在可用性与数据可靠性之间取舍：主从异步双写实现简单、性能高，但主宕机可能丢少量已接收未同步的消息；Dledger 多副本基于 Raft 共识自动选主，提供强一致与故障自动切换，是金融、核心交易的首选。

无论哪种架构，NameServer 都应部署多个节点组成集群，客户端配置多个地址以提升路由可用性。Dledger 模式下建议 3 个 Broker 节点组成一组，容忍单点故障。

八、生产对接与避坑清单

可观测方面，推荐 RocketMQ Exporter 接入 Prometheus + Grafana，重点盯「消费滞后（lag）」与「堆积量」；控制台可用官方 console 容器。接入流程最好放进 GitHub Actions 做生产者/消费者集成测试，避免每次发布手动验证。

常见坑位速查

① brokerIP1 未配宿主机 IP，容器外连不上；② 磁盘 IO 成为瓶颈，建议 SSD 并监控 store 目录使用率；③ 单条消息默认上限 4MB，大消息请改用外部存储+引用；④ 消费逻辑必须幂等，否则重试会重复处理；⑤ 外网暴露 console 务必走 Docker+反向代理并加鉴权，别裸奔在公网。

掌握上述发送/消费/顺序/事务与高可用要点后，RocketMQ 就能稳稳支撑你的核心异步链路。下一篇可以延伸到消息轨迹追踪与跨集群复制，欢迎在评论区交流你踩过的坑。