消息队列（RocketMQ + Kafka）

一、为什么需要消息队列？

单体应用时代，直接调用接口即可完成业务。但到了微服务/分布式时代，系统间调用关系复杂，同步调用会带来诸多问题：

传统同步调用问题：
┌────────┐  ┌────────┐  ┌────────┐
│ 订单服务 │─→│ 库存服务 │─→│ 短信服务 │  如果任一环节失败，整个链路失败
└────────┘  └────────┘  └────────┘

消息队列解决的四大核心问题：

问题	说明	示例
解耦	发布者不依赖消费者，新增消费者无需修改发布者代码	订单创建后，短信、积分、日志各服务独立消费，互不影响
异步	非核心流程异步处理，快速响应客户端	下单后立即返回"下单成功"，短信通知、积分赠送异步完成
削峰	高峰期请求暂存队列，后端按能力消费	秒杀时 10 万 QPS 涌入，MQ 缓冲，后端按 1000 QPS 处理
最终一致性	分布式事务通过消息保证最终一致	订单扣库存，库存服务消费消息扣库存，失败则重试或回滚

二、主流消息队列对比

维度	RocketMQ	Kafka	RabbitMQ
开发语言	Java	Java/Scala	Erlang
吞吐量	10 万级	百万级	万级
延迟	毫秒级	毫秒级	微秒级
可用性	非常高（主从+DLedger）	非常高（分区副本）	高（镜像队列）
消息可靠性	极高（同步刷盘）	高（acks=all）	高
事务消息	支持	支持（0.11+）	不支持
延迟消息	支持（18 个级别）	不支持（需自行实现）	支持（插件）
顺序消息	支持	分区内有序	支持
批量消息	支持	天然支持	不支持
消息回溯	支持（按时间/偏移量）	支持（按偏移量）	不支持
社区活跃	Apache 顶级项目	Apache 顶级项目	活跃
适用场景	金融、电商、事务消息	大数据、日志、流计算	微服务内部通信

选型建议：

RocketMQ： 电商、金融等对事务消息、延迟消息、顺序消息有强需求的场景，阿里系首选
Kafka： 大数据、日志收集、流计算、实时数据分析，海量吞吐场景
RabbitMQ： 中小团队、不需要复杂消息特性的场景，运维简单

三、RocketMQ 架构

3.1 核心组件

                      ┌──────────────────┐
                      │    NameServer     │  （无状态注册中心）
                      │  192.168.1.1:9876 │
                      └────────┬─────────┘
                               │ 注册/发现
          ┌────────────────────┼────────────────────┐
          ▼                    ▼                    ▼
  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐
  │   Broker-A   │    │   Broker-B   │    │   Broker-C   │
  │  Master/Slave│    │  Master/Slave│    │  Master/Slave│
  └──────────────┘    └──────────────┘    └──────────────┘
          ▲                                       ▲
          │ 发送消息                               │ 拉取消息
          │                                       │
  ┌───────┴───────┐                       ┌───────┴───────┐
  │   Producer    │                       │   Consumer    │
  └───────────────┘                       └───────────────┘

组件	角色	说明
NameServer	注册中心	无状态，管理 Broker 路由信息；Broker 启动时注册，Producer/Consumer 定期拉取路由表
Broker	消息存储/转发	核心组件，负责消息存储、投递、查询；Master-Slave 架构，支持同步/异步复制
Producer	消息生产者	发送消息到 Broker，支持同步/异步/单向发送
Consumer	消息消费者	从 Broker 拉取消息（Pull 模式），通过 Consumer Group 协调消费
Topic	消息主题	一类消息的逻辑分类，如 `order_topic`、`payment_topic`
Message Queue	消息队列	Topic 下的物理分区，支持并行消费和顺序消息
Consumer Group	消费者组	同一组内的消费者分摊消费，实现负载均衡

3.2 消息模型

Topic: order_topic
├── Queue 0 ──→ Consumer 1 (Group A)
├── Queue 1 ──→ Consumer 2 (Group A)   ← 同一组内负载均衡
├── Queue 2 ──→ Consumer 3 (Group A)
├── Queue 3 ──→ Consumer 4 (Group A)
│
├── Queue 0 ──→ Consumer 5 (Group B)   ← 不同组独立消费
├── Queue 1 ──→ Consumer 6 (Group B)
└── ...

核心规则：

一个 Queue 只能被同一个 Consumer Group 内的一个 Consumer 消费
一个 Consumer 可以消费多个 Queue
Consumer 数量 > Queue 数量时，多余的 Consumer 空闲
不同 Consumer Group 独立消费同一 Topic（广播效果）

四、RocketMQ 消息类型

4.1 同步发送

生产者发送消息后等待 Broker 确认，可靠性最高，但有一定延迟。

java

// 同步发送
SendResult result = producer.send(
    new Message("order_topic", "TagA", "Order-1001", 
        "订单数据".getBytes(StandardCharsets.UTF_8)));
System.out.println("发送结果: " + result.getSendStatus());  // SEND_OK

4.2 异步发送

生产者发送后立即返回，通过回调处理结果。适合对延迟敏感的场景。

java

// 异步发送
producer.send(message, new SendCallback() {
    @Override
    public void onSuccess(SendResult result) {
        System.out.println("发送成功: " + result.getMsgId());
    }
    
    @Override
    public void onException(Throwable e) {
        System.err.println("发送失败: " + e.getMessage());
        // 重试或记录
    }
});

4.3 单向发送

只发送不等待结果，不关心是否成功，吞吐量最高。适合日志收集等可靠性要求低的场景。

java

// 单向发送
producer.sendOneway(message);

4.4 顺序消息

全局有序： 一个 Topic 只有一个 Queue，所有消息全局有序（但吞吐量低，不推荐）。

分区有序： 通过 MessageQueueSelector 将同一业务标识的消息发送到同一个 Queue，Queue 内保证 FIFO 顺序。

java

// 分区有序：同一订单的消息进入同一个 Queue
SendResult result = producer.send(message, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        Long orderId = (Long) arg;  // 订单 ID
        int index = (int) (orderId % mqs.size());
        return mqs.get(index);
    }
}, orderId);

// 消费者端：顺序消费需要注册 MessageListenerOrderly
consumer.registerMessageListener(new MessageListenerOrderly() {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, 
            ConsumeOrderlyContext context) {
        for (MessageExt msg : msgs) {
            String body = new String(msg.getBody());
            System.out.println("消费: " + body);
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
});

4.5 延迟消息

RocketMQ 支持 18 个延迟级别，消息发送后不会立即投递，而是延迟指定时间后投递。

java

// 延迟消息（18 个级别：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h）
Message message = new Message("order_topic", "订单数据".getBytes());
message.setDelayTimeLevel(3);  // 延迟 10 秒
producer.send(message);

// 典型场景：订单超时未支付自动取消
// 下单时发送延迟 30 分钟的取消检查消息

4.6 事务消息

RocketMQ 的分布式事务方案：半消息 + 事务状态回查，保证本地事务和消息发送的原子性。

流程：
1. Producer 发送半消息（half message）到 Broker
2. Broker 返回半消息发送成功
3. Producer 执行本地事务
4. Producer 根据本地事务结果提交（Commit）或回滚（Rollback）半消息
5. Broker 未收到确认时，定期回查 Producer 的事务状态
6. Commit 的消息投递给 Consumer，Rollback 的消息不投递

java

// 事务消息生产者
TransactionMQProducer producer = new TransactionMQProducer("tx_producer_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) {
        // Broker 回查本地事务状态
        String orderId = msg.getKeys();
        Order order = orderService.getById(orderId);
        return order != null ? 
            LocalTransactionState.COMMIT_MESSAGE : 
            LocalTransactionState.ROLLBACK_MESSAGE;
    }
});

// 发送事务消息
Message message = new Message("order_topic", "TagA", 
    order.getOrderId(), JSON.toJSONBytes(order));
producer.sendMessageInTransaction(message, order);

五、RocketMQ 消费模式

5.1 集群消费 vs 广播消费

模式	行为	适用场景
集群消费（Clustering）	同一 Consumer Group 内，每条消息只被一个 Consumer 消费	负载均衡，默认模式
广播消费（Broadcasting）	同一 Consumer Group 内，每条消息被所有 Consumer 消费	配置刷新、缓存更新

java

// 集群消费（默认）
consumer.setMessageModel(MessageModel.CLUSTERING);

// 广播消费
consumer.setMessageModel(MessageModel.BROADCASTING);

5.2 Push vs Pull

RocketMQ 底层是 Pull 模式，但提供了 Push 的封装（长轮询）。

模式	原理	优点	缺点
Push（封装）	Broker 有新消息时主动推送	实时性好	消费者处理能力不足时会积压
Pull	Consumer 主动拉取	消费者自主控制速率	实时性差，需处理空轮询

java

// Push 模式（封装的长轮询，最常用）
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
    for (MessageExt msg : msgs) {
        processMessage(msg);
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});

// Pull 模式（自主拉取，灵活控制）
consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
    // 顺序消费
    return ConsumeOrderlyStatus.SUCCESS;
});

六、Kafka 架构

6.1 核心组件

                      ┌──────────────────┐
                      │   Zookeeper /    │
                      │   KRaft (3.3+)   │  （元数据管理）
                      └────────┬─────────┘
                               │
          ┌────────────────────┼────────────────────┐
          ▼                    ▼                    ▼
  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐
  │   Broker-1   │    │   Broker-2   │    │   Broker-3   │
  │  Part-0(主)  │    │  Part-0(从)  │    │  Part-1(从)  │
  │  Part-1(从)  │    │  Part-1(主)  │    │  Part-0(从)  │
  └──────────────┘    └──────────────┘    └──────────────┘
          ▲                                       ▲
          │                                       │
  ┌───────┴───────┐                       ┌───────┴───────┐
  │   Producer    │                       │   Consumer    │
  └───────────────┘                       └───────────────┘

组件	角色	说明
Broker	Kafka 服务节点	负责消息存储和转发
Topic	消息主题	逻辑分类，一个 Topic 可跨多个 Broker
Partition	分区	物理分片，每个 Partition 是有序的、不可变的消息序列
Producer	生产者	发送消息，可指定 Partition
Consumer	消费者	拉取消息（Pull）
Consumer Group	消费者组	组内消费者分摊 Partition，实现负载均衡
Zookeeper/KRaft	元数据管理	管理 Broker 列表、Controller 选举、Topic 配置（Kafka 3.3+ 用 KRaft 替代）

6.2 分区机制

为什么需要分区？

水平扩展： 一个 Topic 的数据分散到多个 Broker，突破单机磁盘 IO 瓶颈
并行处理： 多个 Consumer 同时消费不同 Partition，提高消费速度
顺序保证： Partition 内消息严格有序

分区策略：

java

// 1. 指定 Partition（直接指定）
producer.send(new ProducerRecord<>("my_topic", 0, key, value));

// 2. 指定 Key（Hash 到固定 Partition）
// 相同 Key 的消息进入同一 Partition，保证顺序
producer.send(new ProducerRecord<>("my_topic", "order-1001", orderData));

// 3. 轮询（默认，无 Key 时）
producer.send(new ProducerRecord<>("my_topic", value));

// 4. 自定义分区器
public class CustomPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
            Object value, byte[] valueBytes, Cluster cluster) {
        // 自定义逻辑：VIP 用户消息进高优先级 Partition
        String userType = ((Message) value).getUserType();
        return "VIP".equals(userType) ? 0 : 1;
    }
}

// 配置自定义分区器
props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, 
    CustomPartitioner.class.getName());

6.3 消费者组与 Rebalance

Consumer Group 规则：

一个 Partition 只能被同一个 Group 内的一个 Consumer 消费
Consumer 数量 > Partition 数量时，多余的 Consumer 空闲
增加 Consumer 或 Partition 时触发 Rebalance

Rebalance 触发条件：

Consumer 加入/离开 Group
Topic 新增 Partition
Consumer 超时未发送心跳

Rebalance 过程（Eager Protocol）：

Group Coordinator 通知所有 Consumer 停止消费
所有 Consumer 放弃当前 Partition
重新分配 Partition 给 Consumer
Consumer 从新分配的 Partition 恢复消费

新版 Cooperative Rebalance（Kafka 2.4+）： 增量重分配，不停止所有消费，减少 STW（Stop-The-World）时间。

java

// 配置 Cooperative Rebalance
props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
    CooperativeStickyAssignor.class.getName());

6.4 Offset 管理

Kafka 中每条消息在 Partition 内有一个唯一递增的 Offset。Consumer 通过 Offset 记录消费进度。

java

// 自动提交（默认，简单但有重复消费风险）
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);

// 手动提交（精确控制，推荐）
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

// 同步提交（阻塞直到提交成功）
consumer.commitSync();

// 异步提交（不阻塞，吞吐量高）
consumer.commitAsync((offsets, exception) -> {
    if (exception != null) {
        System.err.println("提交失败: " + exception.getMessage());
    }
});

// 精确消费：处理完消息再提交（At-Least-Once 语义）
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);  // 处理消息
    }
    consumer.commitSync();  // 处理完一批再提交 Offset
}

Offset 存储： Kafka 0.9+ 后，Offset 存储在内部 Topic __consumer_offsets 中，不再依赖 Zookeeper。

七、Kafka 高吞吐原因

Kafka 能实现百万级 TPS，核心在于以下设计：

技术	原理	效果
顺序读写磁盘	消息以追加方式写入日志文件，磁盘顺序读写速度接近内存随机读写	磁盘 IO 不再是瓶颈
零拷贝（Zero-Copy）	`sendfile()` 系统调用，数据从磁盘直接传输到网卡，不经过用户态	减少 CPU 拷贝和上下文切换
批量发送	Producer 将多条消息打包成一批发送，`batch.size` 和 `linger.ms` 控制	减少网络请求次数
数据压缩	支持 gzip、snappy、lz4、zstd 压缩，压缩比高	减少网络传输和磁盘存储
Page Cache	利用操作系统的页缓存，读写都走 Page Cache	减少实际磁盘 IO
分区并行	多个 Partition 分布在多台 Broker，生产和消费并行	横向扩展，线性提升吞吐

java

// 生产者配置优化
Properties props = new Properties();
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);       // 16KB 批量大小
props.put(ProducerConfig.LINGER_MS_CONFIG, 10);            // 等待 10ms 凑批
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");  // 压缩算法
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);  // 32MB 缓冲区
props.put(ProducerConfig.ACKS_CONFIG, "1");                // Leader 写入成功即返回

八、Kafka Exactly-Once 语义

8.1 三种消息语义

语义	含义	实现
At-Most-Once	最多一次，可能丢消息	`acks=0`，发送即忘
At-Least-Once	至少一次，可能重复	`acks=1/all` + 重试，默认
Exactly-Once	精确一次，不丢不重	幂等 Producer + 事务

8.2 幂等 Producer

Kafka 0.11+ 引入，通过 Producer ID (PID) + Sequence Number 实现单 Partition 内 exactly-once。

java

// 开启幂等性（自动设置 acks=all, retries=Integer.MAX_VALUE, max.in.flight=1）
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

原理： Broker 为每个 Producer 分配 PID，每条消息带 Sequence Number。Broker 检查 Sequence Number 是否连续，跳过重复消息。

8.3 事务

Kafka 0.11+ 引入事务，支持跨 Partition 的原子写入。

java

// 事务 Producer
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tx-order-producer");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("order_topic", orderData));
    producer.send(new ProducerRecord<>("log_topic", logData));
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

// 事务消费者：只读取已提交事务的消息
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");

九、消息可靠性保障

9.1 可靠性全景图

Producer 发送 ──→ Broker 存储 ──→ Consumer 消费
      │                │                │
  重试机制          同步刷盘         手动ACK
  同步发送          多副本复制       幂等处理
  事务消息          集群部署         死信队列

9.2 RocketMQ 可靠性配置

java

// Producer 端
producer.setRetryTimesWhenSendFailed(3);          // 同步发送失败重试 3 次
producer.setRetryTimesWhenSendAsyncFailed(3);     // 异步发送失败重试 3 次

// Broker 端（broker.conf）
flushDiskType = SYNC_FLUSH        // 同步刷盘（每条消息都 fsync）
brokerRole = SYNC_MASTER          // 同步复制（Master 等待 Slave 确认）

java

// Consumer 端
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
    try {
        for (MessageExt msg : msgs) {
            processMessage(msg);
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;  // 消费成功才确认
    } catch (Exception e) {
        // 消费失败，返回 RECONSUME_LATER 触发重试
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    }
});

// 设置最大重试次数
consumer.setMaxReconsumeTimes(16);  // 默认 16 次后进入死信队列

9.3 Kafka 可靠性配置

java

// Producer 端
props.put(ProducerConfig.ACKS_CONFIG, "all");           // 所有 ISR 副本确认
props.put(ProducerConfig.RETRIES_CONFIG, 3);            // 重试次数
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);  // 顺序保证

// Broker 端（server.properties）
min.insync.replicas=2              // 最少同步副本数（ack=all 时等待 ISR 中至少 2 个副本确认）
default.replication.factor=3       // 默认副本数
unclean.leader.election.enable=false  // 不允许 ISR 外的副本成为 Leader

// Consumer 端
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);  // 手动提交
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");  // 无 Offset 时从头消费

十、消息幂等性

为什么消息会重复？

Producer 发送消息，Broker 已存储但 ACK 丢失 → Producer 重试 → 消息重复
Consumer 消费完消息，Commit Offset 前宕机 → 重启后重新消费 → 重复消费

解决方案：

10.1 数据库唯一索引

java

// 消费消息时，利用数据库唯一约束防重
@Transactional
public void consumeOrderMessage(OrderMessage msg) {
    try {
        // 订单 ID 作为唯一键，重复插入抛异常
        orderMapper.insert(new Order(msg.getOrderId(), msg.getAmount()));
        // 后续业务逻辑
    } catch (DuplicateKeyException e) {
        // 重复消息，直接忽略
        System.out.println("重复消息，已忽略: " + msg.getOrderId());
    }
}

10.2 Redis 去重

java

// 使用 Redis SETNX 实现去重
public boolean consumeMessage(String messageId, String messageBody) {
    String dedupKey = "msg:dedup:" + messageId;
    // 如果 SETNX 成功，说明是第一次处理
    Boolean isNew = redisTemplate.opsForValue()
        .setIfAbsent(dedupKey, "1", 24, TimeUnit.HOURS);
    
    if (Boolean.TRUE.equals(isNew)) {
        // 处理消息
        processMessage(messageBody);
        return true;
    } else {
        System.out.println("重复消息，已跳过: " + messageId);
        return false;
    }
}

10.3 乐观锁版本号

java

// 数据库更新时使用版本号
UPDATE order SET status = 'PAID', version = version + 1 
WHERE order_id = #{orderId} AND version = #{version}

// 如果 affected_rows == 0，说明已被其他消息处理过，跳过

十一、消息顺序性

11.1 RocketMQ 顺序消息

java

// 生产者：同一订单 ID 的消息发到同一个 Queue
producer.send(message, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        Long orderId = (Long) arg;
        return mqs.get((int) (orderId % mqs.size()));
    }
}, orderId);

// 消费者：使用 MessageListenerOrderly，同一 Queue 的消息串行消费
consumer.registerMessageListener(new MessageListenerOrderly() {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
            ConsumeOrderlyContext context) {
        // 顺序消费，前一条成功才消费下一条
        for (MessageExt msg : msgs) {
            processMessage(msg);
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
});

11.2 Kafka 分区内有序

java

// 生产者：相同 Key 的消息进入同一 Partition
producer.send(new ProducerRecord<>("order_topic", "order-1001", messageData));

// 消费者：单 Partition 内消息有序，多 Partition 之间无序
// 如需全局有序，Topic 只设 1 个 Partition（但严重限制吞吐量）

RocketMQ vs Kafka 顺序消息对比：

维度	RocketMQ	Kafka
实现难度	简单，原生支持	分区内有序，全局有序需要 1 个 Partition
全局有序	1 个 Queue	1 个 Partition
分区有序	MessageQueueSelector	指定 Key 或自定义 Partition
性能影响	顺序消费时 Queue 锁竞争	1 个 Partition 时无并行

十二、死信队列

死信队列（Dead Letter Queue）： 消息消费失败且重试次数耗尽后，会被投递到死信队列，避免阻塞正常消息的消费。

12.1 RocketMQ 死信队列

java

// 消费失败返回 RECONSUME_LATER，默认最多重试 16 次
consumer.setMaxReconsumeTimes(16);  // 16 次后进入死信队列

// 死信队列命名：%DLQ% + ConsumerGroup
// 例如：%DLQ%order_consumer_group
// 可通过 Admin 工具查看和重放死信消息

java

// 死信消息处理
// 方案一：定时任务扫描死信队列，人工处理
// 方案二：监听死信队列，自动告警
consumer.subscribe("%DLQ%order_consumer_group", "*");
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
    for (MessageExt msg : msgs) {
        // 记录到数据库，触发告警
        deadLetterService.save(msg);
        alertService.send("消息消费失败进入死信: " + msg.getMsgId());
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});

12.2 Kafka 死信队列

Kafka 没有内置死信队列，需要自行实现：

java

// 消费失败时将消息发送到死信 Topic
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        try {
            processRecord(record);
        } catch (Exception e) {
            // 发送到死信 Topic
            ProducerRecord<String, String> dlqRecord = new ProducerRecord<>(
                "dlq_" + record.topic(),
                record.key(),
                record.value());
            dlqRecord.headers().add("original_topic", record.topic().getBytes());
            dlqRecord.headers().add("exception", e.getMessage().getBytes());
            dlqProducer.send(dlqRecord);
        }
    }
    consumer.commitSync();
}

十三、Spring Boot 集成

13.1 RocketMQ Spring Boot Starter

yaml

# application.yml
rocketmq:
  name-server: 192.168.1.1:9876;192.168.1.2:9876
  producer:
    group: order_producer_group
    send-message-timeout: 3000
    retry-times-when-send-failed: 3
    max-message-size: 4194304
  consumer:
    group: order_consumer_group
    topic: order_topic
    max-reconsume-times: 16

java

// 生产者
@RestController
@RequestMapping("/order")
public class OrderController {
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    @PostMapping("/create")
    public String createOrder(@RequestBody Order order) {
        // 同步发送
        SendResult result = rocketMQTemplate.syncSend(
            "order_topic:create", order);
        return result.getSendStatus().name();
    }
    
    @PostMapping("/createAsync")
    public String createOrderAsync(@RequestBody Order order) {
        // 异步发送
        rocketMQTemplate.asyncSend("order_topic:create", order, 
            new SendCallback() {
                @Override
                public void onSuccess(SendResult result) {
                    System.out.println("发送成功");
                }
                @Override
                public void onException(Throwable e) {
                    System.err.println("发送失败");
                }
            });
        return "OK";
    }
    
    @PostMapping("/createDelay")
    public String createOrderDelay(@RequestBody Order order) {
        // 延迟消息：30 分钟后检查订单是否支付
        rocketMQTemplate.syncSend("order_topic:check", 
            MessageBuilder.withPayload(order).build(),
            3000, 16);  // timeout=3000ms, delayLevel=16 (30分钟)
        return "OK";
    }
    
    @PostMapping("/createTransaction")
    public String createOrderTx(@RequestBody Order order) {
        // 事务消息
        rocketMQTemplate.sendMessageInTransaction(
            "order_tx_group", "order_topic:create",
            MessageBuilder.withPayload(order).build(), order);
        return "OK";
    }
}

java

// 消费者
@Component
@RocketMQMessageListener(
    topic = "order_topic",
    consumerGroup = "order_consumer_group",
    selectorExpression = "create || update",  // 过滤 Tag
    consumeMode = ConsumeMode.CONCURRENTLY,    // 并发消费
    maxReconsumeTimes = 16
)
public class OrderConsumer implements RocketMQListener<Order> {
    
    @Override
    public void onMessage(Order order) {
        System.out.println("消费订单: " + order.getOrderId());
        // 业务处理
    }
}

// 顺序消费
@Component
@RocketMQMessageListener(
    topic = "order_topic",
    consumerGroup = "order_consumer_group",
    consumeMode = ConsumeMode.ORDERLY,  // 顺序消费
    selectorExpression = "status_change"
)
public class OrderStatusConsumer implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt message) {
        // 顺序处理订单状态变更
    }
}

13.2 Kafka Spring Boot Starter

yaml

# application.yml
spring:
  kafka:
    bootstrap-servers: 192.168.1.1:9092,192.168.1.2:9092,192.168.1.3:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
      retries: 3
      batch-size: 16384
      linger-ms: 10
      compression-type: lz4
      properties:
        enable.idempotence: true
    consumer:
      group-id: order_consumer_group
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      enable-auto-commit: false
      auto-offset-reset: earliest
      max-poll-records: 500
      properties:
        spring.json.trusted.packages: com.example.dto
        partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignor
    listener:
      ack-mode: manual  # 手动确认
      concurrency: 3    # 并发消费者线程数

java

// 生产者
@RestController
@RequestMapping("/kafka")
public class KafkaController {
    
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    
    @PostMapping("/send")
    public String send(@RequestBody Order order) {
        // 异步发送（默认）
        kafkaTemplate.send("order_topic", order.getOrderId(), order);
        return "OK";
    }
    
    @PostMapping("/sendSync")
    public String sendSync(@RequestBody Order order) throws Exception {
        // 同步发送
        SendResult<String, Object> result = kafkaTemplate.send(
            "order_topic", order.getOrderId(), order).get(3, TimeUnit.SECONDS);
        return result.getRecordMetadata().toString();
    }
    
    @PostMapping("/sendBatch")
    public String sendBatch(@RequestBody List<Order> orders) {
        // 批量发送
        List<ListenableFuture<SendResult<String, Object>>> futures = new ArrayList<>();
        for (Order order : orders) {
            futures.add(kafkaTemplate.send("order_topic", 
                order.getOrderId(), order));
        }
        return "OK";
    }
}

java

// 消费者（手动确认）
@Component
public class OrderKafkaConsumer {
    
    @KafkaListener(topics = "order_topic", groupId = "order_consumer_group")
    public void onMessage(ConsumerRecord<String, Order> record,
            Acknowledgment ack) {
        try {
            Order order = record.value();
            System.out.println("消费订单: " + order.getOrderId() + 
                ", partition: " + record.partition() + 
                ", offset: " + record.offset());
            // 业务处理
            processOrder(order);
            // 手动确认
            ack.acknowledge();
        } catch (Exception e) {
            System.err.println("消费失败: " + e.getMessage());
            // 不确认，触发重试
        }
    }
    
    @KafkaListener(topics = "order_topic", groupId = "order_consumer_group",
            concurrency = "3")  // 3 个并发线程
    public void onMessageConcurrent(ConsumerRecord<String, Order> record) {
        // 并发消费
    }
}

十四、面试要点

Q1：消息队列的核心作用是什么？

解耦、异步、削峰、最终一致性。解耦让发布者和消费者独立演进；异步让非核心流程后台处理，提升响应速度；削峰将瞬时高峰流量暂存队列，保护后端服务；最终一致性通过消息实现分布式事务的最终一致性。

Q2：RocketMQ 和 Kafka 的核心区别？怎么选型？

RocketMQ 是 Java 体系，功能更丰富（事务消息、延迟消息、Tag 过滤、消息回溯），适合电商/金融等复杂业务场景。Kafka 吞吐量更高（百万级），适合大数据/日志/流计算场景。选型：事务消息/延迟消息需求 → RocketMQ；海量日志/大数据/流计算 → Kafka；中小团队简单场景 → RabbitMQ。

Q3：如何保证消息不丢失？

三端保障：Producer 端使用同步发送 + 重试机制（RocketMQ 同步发送，Kafka acks=all）；Broker 端使用同步刷盘 + 多副本（RocketMQ SYNC_FLUSH + SYNC_MASTER，Kafka min.insync.replicas>=2）；Consumer 端使用手动 ACK，处理完业务再提交 Offset，消费失败不确认。

Q4：如何保证消息不重复消费？

消息重复不可避免（网络抖动导致 ACK 丢失），需要在消费端做幂等。常用方案：数据库唯一索引（订单 ID 唯一约束）；Redis SETNX 去重（消息 ID 作为 key）；乐观锁版本号（UPDATE 时带版本号，影响行数为 0 则跳过）；业务状态机（已处理状态的消息跳过）。

Q5：RocketMQ 事务消息的实现原理？

半消息机制：Producer 先发送半消息到 Broker（此时 Consumer 不可见），然后执行本地事务；本地事务成功则提交半消息（Consumer 可见），失败则回滚；Broker 未收到确认时定期回查 Producer 的本地事务状态。关键是 TransactionListener 中的 executeLocalTransaction 和 checkLocalTransaction 两个方法。

Q6：Kafka 为什么吞吐量这么高？

五大原因：顺序写磁盘（追加写入，接近内存速度）；零拷贝（sendfile 系统调用，数据从磁盘直接到网卡）；批量发送（Producer 攒批，减少网络请求）；数据压缩（gzip/snappy/lz4，减少传输量）；Page Cache（利用操作系统页缓存，读写命中缓存时无需实际磁盘 IO）。

Q7：消息积压了怎么办？

排查原因：Consumer 是否有异常（死循环、慢查询、死锁）；Consumer 数量是否足够（增加 Consumer 实例，但不超过 Partition/Queue 数）；消费逻辑是否过慢（优化消费逻辑，减少 DB 操作，增加缓存）。临时方案：新建临时 Consumer Group，将积压消息转移到新 Topic，用更多 Consumer 并行消费，处理完后再切回原 Group。

Q8：Kafka 的 Rebalance 是什么？有什么问题？

Rebalance 是 Consumer Group 内 Partition 重新分配的过程。触发条件：Consumer 加入/离开、Topic 新增 Partition、Consumer 心跳超时。问题：Eager Rebalance 会导致所有 Consumer 停止消费（STW），影响实时性。解决：使用 Cooperative Rebalance（Kafka 2.4+），增量重分配，减少停止时间；合理设置 session.timeout.ms 和 max.poll.interval.ms 避免误触发。

消息队列（RocketMQ + Kafka） ​

一、为什么需要消息队列？ ​

二、主流消息队列对比 ​

三、RocketMQ 架构 ​

3.1 核心组件 ​

3.2 消息模型 ​

四、RocketMQ 消息类型 ​

4.1 同步发送 ​

4.2 异步发送 ​

4.3 单向发送 ​

4.4 顺序消息 ​

4.5 延迟消息 ​

4.6 事务消息 ​

五、RocketMQ 消费模式 ​

5.1 集群消费 vs 广播消费 ​

5.2 Push vs Pull ​

六、Kafka 架构 ​

6.1 核心组件 ​

6.2 分区机制 ​

6.3 消费者组与 Rebalance ​

6.4 Offset 管理 ​

七、Kafka 高吞吐原因 ​

八、Kafka Exactly-Once 语义 ​

8.1 三种消息语义 ​

8.2 幂等 Producer ​

8.3 事务 ​

九、消息可靠性保障 ​

9.1 可靠性全景图 ​

9.2 RocketMQ 可靠性配置 ​

9.3 Kafka 可靠性配置 ​

十、消息幂等性 ​

10.1 数据库唯一索引 ​

10.2 Redis 去重 ​

10.3 乐观锁版本号 ​

十一、消息顺序性 ​

11.1 RocketMQ 顺序消息 ​

11.2 Kafka 分区内有序 ​

十二、死信队列 ​

12.1 RocketMQ 死信队列 ​

12.2 Kafka 死信队列 ​

十三、Spring Boot 集成 ​

13.1 RocketMQ Spring Boot Starter ​

13.2 Kafka Spring Boot Starter ​

十四、面试要点 ​

消息队列（RocketMQ + Kafka）

一、为什么需要消息队列？

二、主流消息队列对比

三、RocketMQ 架构

3.1 核心组件

3.2 消息模型

四、RocketMQ 消息类型

4.1 同步发送

4.2 异步发送

4.3 单向发送

4.4 顺序消息

4.5 延迟消息

4.6 事务消息

五、RocketMQ 消费模式

5.1 集群消费 vs 广播消费

5.2 Push vs Pull

六、Kafka 架构

6.1 核心组件

6.2 分区机制

6.3 消费者组与 Rebalance

6.4 Offset 管理

七、Kafka 高吞吐原因

八、Kafka Exactly-Once 语义

8.1 三种消息语义

8.2 幂等 Producer

8.3 事务

九、消息可靠性保障

9.1 可靠性全景图

9.2 RocketMQ 可靠性配置

9.3 Kafka 可靠性配置

十、消息幂等性

10.1 数据库唯一索引

10.2 Redis 去重

10.3 乐观锁版本号

十一、消息顺序性

11.1 RocketMQ 顺序消息

11.2 Kafka 分区内有序

十二、死信队列

12.1 RocketMQ 死信队列

12.2 Kafka 死信队列

十三、Spring Boot 集成

13.1 RocketMQ Spring Boot Starter

13.2 Kafka Spring Boot Starter

十四、面试要点