Apache RocketMQ

Apache RocketMQ 简介

Apache RocketMQ 是一款开源的分布式消息中间件，由阿里巴巴开发并贡献给 Apache 基金会。它专为金融级高可靠、低延迟场景设计，支持万亿级消息堆积和毫秒级投递延迟，广泛应用于电商交易、实时计算、日志收集等核心业务场景。RocketMQ 采用发布-订阅模型，具备强大的事务消息、顺序消息和消息回溯能力，是阿里巴巴双11大促的核心消息引擎。

核心特性

1. 金融级高可靠
   • 消息持久化存储（支持同步/异步刷盘）
   • 多副本机制（基于 Raft 协议）保障数据不丢失
2. 低延迟与高吞吐
   • 单机支持 10万级 TPS，集群可扩展至百万级
   • 平均投递延迟 **<1ms**（SSD 环境下）
3. 消息类型丰富
   • 事务消息：确保分布式事务一致性（如订单支付）
   • 顺序消息：保证同一业务ID的消息顺序性（如订单状态变更）
   • 延迟消息：支持定时/延时投递（如优惠券到期提醒）
4. 分布式架构
   • NameServer：轻量级服务发现组件（无 ZooKeeper 依赖）
   • Broker：消息存储与转发节点（支持主从切换）
   • Producer/Consumer：支持集群化部署
5. 生态集成
   • 无缝对接 RocketMQ Connect（数据同步工具）
   • 支持 OpenMessaging 标准协议
   • 与 Flink、Spark 等流计算引擎集成

核心概念

典型应用场景

1. 电商交易系统
   • 订单创建、支付、库存扣减的异步解耦
   • 双11大促的流量削峰填谷
2. 金融支付
   • 跨行转账的事务消息保障
   • 实时对账与风控处理
3. 物联网（IoT）
   • 海量设备数据的实时采集与分发
4. 日志收集
   • 替代 Kafka 实现日志聚合（如 ELK 架构）
5. 微服务通信
   • 服务间通过消息总线（Event Bus）解耦

快速入门示例

1. 启动 RocketMQ 单机版

# 下载并解压（需提前安装 JDK 8+）
wget https://archive.apache.org/dist/rocketmq/5.0.0/rocketmq-all-5.0.0-bin-release.zip
unzip rocketmq-all-5.0.0-bin-release.zip
cd rocketmq-5.0.0

# 启动 NameServer
nohup bin/mqnamesrv &

# 启动 Broker
nohup bin/mqbroker -n localhost:9876 &

2. 发送与消费消息（Java API）

// 生产者示例
DefaultMQProducer producer = new DefaultMQProducer("producer_group");
producer.setNamesrvAddr("localhost:9876");
producer.start();
Message msg = new Message("test_topic", "TagA", "Hello RocketMQ".getBytes());
SendResult result = producer.send(msg);

// 消费者示例
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");
consumer.subscribe("test_topic", "TagA");
consumer.registerMessageListener((List<MessageExt> msgs, ConsumeConcurrentlyContext ctx) -> {
    System.out.println("Received: " + new String(msgs.get(0).getBody()));
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
consumer.start();

性能优化建议

1. 部署架构
   • 生产环境至少部署 2主2从 Broker 集群
   • 使用 SSD 存储 CommitLog（storePathCommitLog 参数）
2. 消息设计
   • 合理设置 Topic 分片数（Queue 数量 = 消费者线程数 × 2）
   • 避免单条消息过大（建议 **<1MB**）
3. 参数调优
   • 调整 sendLatencyFaultEnable=true 规避故障 Broker
   • 增大 pullBatchSize 提升消费吞吐量
4. 监控告警
   • 通过 RocketMQ Dashboard 监控堆积量（Consumer Lag）
   • 设置 Broker CPU/磁盘使用率阈值告警