消息队列是分布式系统异步解耦的核心基础设施。RocketMQ 适合金融级可靠消息，Kafka 适合高吞吐日志流处理，Pulsar 则是云原生时代的统一消息平台。
1. 存储模型对比
| 特性 | Kafka | RocketMQ | Pulsar |
|---|---|---|---|
| 存储 | 日志分段文件 | CommitLog + ConsumeQueue | BookKeeper (分布式日志) |
| 消费模型 | Pull | Push + Pull | Push + Pull |
| 顺序保证 | Partition 内有序 | Queue 内有序 | Partition 内有序 |
| 延迟消息 | TimeWheel (有限) | 18 级延迟 | 原生支持 |
| 事务消息 | 事务生产者 | ✅ 原生支持 | ✅ 原生支持 |
| 多租户 | ❌ | ❌ | ✅ |
| Geo 复制 | MirrorMaker | 不完全 | ✅ 内置 |
| 吞吐 | 百万级/秒 | 十万级/秒 | 百万级/秒 |
2. RocketMQ 核心架构
Producer → NameServer (路由发现)
↓
Broker Master/Slave
↓
Consumer ← NameServer
2.1 存储结构
CommitLog # 所有消息顺序写入（1G 一个文件）
├── ConsumeQueue # 消费队列索引（定长 20byte/条目）
│ ├── TopicA/Queue0
│ ├── TopicA/Queue1
│ └── TopicB/Queue0
└── IndexFile # 按 Key/Time 索引
2.2 事务消息
// 发送半消息
TransactionMQProducer producer = new TransactionMQProducer("tx_group");
producer.setTransactionListener(new TransactionListener() {
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
try {
// 执行本地事务
orderService.createOrder((Order) arg);
return LocalTransactionState.COMMIT_MESSAGE;
} catch (Exception e) {
return LocalTransactionState.ROLLBACK_MESSAGE;
}
}
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
// 本地事务回查
Order order = orderService.getById(msg.getKeys());
return order != null ? COMMIT_MESSAGE : ROLLBACK_MESSAGE;
}
});
2.3 顺序消费
// 全局顺序：所有消息发到同一个 Queue
SendResult result = producer.send(msg, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
return mqs.get(0); // 固定发送到 Queue 0
}
}, null);
// 局部顺序：按 Sharding Key 选择 Queue（如订单 ID）
SendResult result = producer.send(msg, (mqs, m, arg) -> {
int index = ((String) arg).hashCode() % mqs.size();
return mqs.get(Math.abs(index));
}, orderId);
// 消费者：单线程消费保证顺序
consumer.registerMessageListener(new MessageListenerOrderly() {
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext ctx) {
for (MessageExt msg : msgs) {
process(msg);
}
return ConsumeOrderlyStatus.SUCCESS;
}
});
3. Kafka 核心原理
3.1 ISR 机制
Partition 0:
Leader: Broker 1 (offset: 0-1000)
ISR: [Broker 1, Broker 2] # 同步副本集合
OSR: [Broker 3] # 滞后副本
副本滞后判定: replica.lag.time.max.ms=10000
3.2 生产者配置
Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 幂等生产者（自动去重）
props.put("enable.idempotence", "true");
props.put("acks", "all");
props.put("retries", Integer.MAX_VALUE);
// 事务生产者
props.put("transactional.id", "prod-1");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
// 事务发送
producer.initTransactions();
try {
producer.beginTransaction();
producer.send(new ProducerRecord<>("topic", "key", "value"));
producer.sendOffsetsToTransaction(consumer.position(consumer.assignment()), consumer.groupMetadata());
producer.commitTransaction();
} catch (Exception e) {
producer.abortTransaction();
}
3.3 消费者分区分配
| 策略 | 算法 | 特点 |
|---|---|---|
| Range | 按 Topic 范围分配 | 默认，可能不均衡 |
| RoundRobin | 轮询分配 | 较均衡 |
| Sticky | 最小化重新分配 | 再均衡影响小 |
| Cooperative | 增量再均衡 | Kafka 2.4+ 推荐 |
4. Pulsar：云原生消息队列
Pulsar 的核心创新是 存储与计算分离：
- Broker：无状态，处理生产和消费
- BookKeeper：分布式日志存储
- ZooKeeper/etcd：元数据管理
Producer → Broker → Bookie (Ledger)
↓
Consumer ← Broker
4.1 多租户与命名空间
# 创建租户
pulsar-admin tenants create my-org
# 创建命名空间
pulsar-admin namespaces create my-org/finance
# 设置策略
pulsar-admin namespaces set-retention my-org/finance \
--size 10G --time 3d
4.2 统一消息模型
Pulsar 统一了队列（Queue）和流（Stream）：
- Exclusive/Failover：队列模式，单消费者
- Shared：队列模式，多消费者竞争
- Key_Shared：按 Key 路由到固定消费者（顺序保证 + 并行）
5. 选型建议
| 场景 | 推荐 | 理由 |
|---|---|---|
| 金融交易 | RocketMQ | 强一致、事务消息完善 |
| 日志采集/大数据 | Kafka | 吞吐最高、生态成熟 |
| 云原生/多租户 | Pulsar | 存储计算分离、弹性扩缩 |
| 实时计算 | Kafka/Pulsar | 流处理集成好 |
| 延迟消息 | RocketMQ | 18 级延迟原生支持 |
6. 消息队列最佳实践
- 生产端：异步发送、批量发送、失败重试、限流保护
- 消费端：幂等设计、消费确认、死信队列、消费速率监控
- 运维端：监控堆积延迟、磁盘水位、消费组 Rebalance
7. 消息队列在云原生时代的演进
Serverless 事件驱动
现代云原生架构中，消息队列与 Serverless 函数深度集成：
# AWS EventBridge + Lambda 示例
EventBus:
Type: AWS::Events::EventBus
Properties:
Name: order-events
OrderCreatedRule:
Type: AWS::Events::Rule
Properties:
EventBusName: !Ref EventBus
EventPattern:
source:
- order.service
detail-type:
- OrderCreated
Targets:
- Arn: !GetAtt ProcessOrderFunction.Arn
Id: ProcessOrder
事件网格（Event Mesh）
┌─────────────────────────────────────────────────┐
│ Event Mesh │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Kafka │◄──►│ Broker │◄──►│ RabbitMQ │ │
│ │ (流处理) │ │ (协议桥接)│ │ (任务队列)│ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ ▲ ▲ │
│ ┌──────┴──────┐ ┌──────┴──────┐ │
│ │ CloudEvents │ │ CloudEvents │ │
│ │ (统一协议) │ │ (统一协议) │ │
│ └─────────────┘ └─────────────┘ │
│ ▲ ▲ │
│ ┌──────┴──────┐ ┌──────┴──────┐ │
│ │ 服务 A │ │ 服务 B │ │
│ └─────────────┘ └─────────────┘ │
└─────────────────────────────────────────────────┘
事件网格通过 CloudEvents 规范统一事件格式，让不同消息系统之间的事件可以自由流转。
8. 消息队列监控指标体系
| 指标 | 告警阈值 | 说明 |
|---|---|---|
| 消息堆积量 | > 10万条 | 消费速度跟不上生产速度 |
| 消费延迟 | > 5分钟 | 从生产到消费的时间差 |
| 消费速率 | 下降 50% | 消费者异常或处理变慢 |
| 磁盘使用率 | > 80% | 消息存储空间不足 |
| 消费组 Rebalance | 频繁触发 | 消费者上下线过于频繁 |
| 死信队列积压 | > 1000条 | 大量消息消费失败 |
延伸阅读
- Spring 集成消息队列 — Spring 与 RabbitMQ/Kafka/RocketMQ 的集成方案
- 事件驱动架构设计 — 消息队列在架构层面的设计模式
- Kafka Connect 数据集成 — 数据管道构建
总结
RocketMQ、Kafka 和 Pulsar 代表了消息队列技术的不同演进方向：RocketMQ 面向金融级可靠消息，Kafka 面向高吞吐日志流处理，Pulsar 则是云原生时代的统一消息平台。选型应基于业务对事务、延迟、吞吐和扩展性的具体需求。无论选择哪款消息队列，都必须重视消费端的幂等设计和死信队列机制，这是保障消息不丢失、业务不重复的关键防线。
消息队列的避坑指南
顺序消息的使用误区：
很多业务场景"需要顺序"实际上是伪需求。全局顺序会严重限制并行度，导致吞吐量骤降。真正的顺序需求通常可以降级为：
- 分区顺序：同一分区（如同一 user_id）内有序，不同分区间无序
- 局部顺序：同一业务实体（如订单）的操作有序
RocketMQ 的 MessageQueueSelector 和 Kafka 的 Partitioner 都支持自定义分区策略。切勿用单分区实现全局顺序，除非吞吐量需求极低。
消息补偿与对账机制：
金融级业务中，消息队列仅作为异步通知，核心状态变更仍需对账兜底：
日终对账流程：
1. 提取今日所有成功支付订单
2. 提取消费端处理成功的订单
3. 找出差异集（已支付未处理 / 已处理未支付）
4. 人工或自动补偿
消息体大小控制：
Kakfa 默认单条消息上限 1MB，RocketMQ 默认 4MB。过大的消息会：
- 降低 Broker 吞吐量
- 增加内存压力
- 延长消费延迟
建议：消息体仅包含业务 ID 和关键摘要，详细数据通过对象存储传递 URL 引用。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。