1492 lines
37 KiB
Markdown
1492 lines
37 KiB
Markdown
# 消息队列（RocketMQ/Kafka）
|
||
|
||
## 问题
|
||
|
||
**背景**：消息队列是分布式系统的核心组件，用于解耦、异步、削峰填谷。
|
||
|
||
**问题**：
|
||
1. 消息队列的作用是什么？在什么场景下使用？
|
||
2. 如何保证消息不丢失？（生产者、Broker、消费者三个层面）
|
||
3. 如何保证消息不重复消费（幂等性）？
|
||
4. 如何保证消息的顺序性？
|
||
5. 什么是消息积压？如何处理？
|
||
6. RocketMQ 和 Kafka 的区别是什么？如何选型？
|
||
|
||
---
|
||
|
||
## 标准答案
|
||
|
||
### 1. 消息队列的作用
|
||
|
||
#### **核心作用**
|
||
|
||
##### **1. 解耦（Decoupling）**
|
||
|
||
**场景**：用户注册成功后，需要发送欢迎邮件、发送短信、发放优惠券。
|
||
|
||
**不使用 MQ**：
|
||
```
|
||
用户服务
|
||
↓
|
||
调用邮件服务 ✉️
|
||
调用短信服务 📱
|
||
调用优惠券服务 🎫
|
||
```
|
||
|
||
**问题**：
|
||
- 用户服务依赖所有下游服务
|
||
- 任何一个下游服务故障都会影响注册流程
|
||
- 新增需求需要修改用户服务代码
|
||
|
||
**使用 MQ**：
|
||
```
|
||
用户服务 → 发送消息到 MQ ← 邮件服务（订阅）
|
||
← 短信服务（订阅）
|
||
← 优惠券服务（订阅）
|
||
← 新服务（随时订阅）
|
||
```
|
||
|
||
**优点**：
|
||
- 用户服务不需要知道下游服务的存在
|
||
- 下游服务故障不影响用户注册
|
||
- 新增服务只需订阅消息，无需修改用户服务
|
||
|
||
---
|
||
|
||
##### **2. 异步（Asynchronous）**
|
||
|
||
**场景**：用户下单后，需要：
|
||
1. 扣减库存
|
||
2. 生成订单
|
||
3. 通知物流
|
||
4. 发送短信
|
||
5. 积分处理
|
||
|
||
**同步调用（串行）**：
|
||
```
|
||
总耗时 = 50ms + 100ms + 80ms + 50ms + 60ms = 340ms
|
||
```
|
||
|
||
**异步调用（MQ）**：
|
||
```
|
||
下单服务 → MQ（5ms）
|
||
└─ 返回给用户（总耗时：55ms）
|
||
|
||
后台异步处理：
|
||
├─ 库存服务（50ms）
|
||
├─ 订单服务（100ms）
|
||
├─ 物流服务（80ms）
|
||
├─ 短信服务（50ms）
|
||
└─ 积分服务（60ms）
|
||
```
|
||
|
||
**优点**：
|
||
- 响应时间从 340ms 降低到 55ms
|
||
- 用户体验提升
|
||
|
||
---
|
||
|
||
##### **3. 削峰填谷（Peak Shaving）**
|
||
|
||
**场景**：秒杀活动、双11大促
|
||
|
||
**不使用 MQ**：
|
||
```
|
||
瞬时 QPS：50000
|
||
数据库最大 QPS：5000
|
||
|
||
结果：数据库被打挂 ❌
|
||
```
|
||
|
||
**使用 MQ**：
|
||
```
|
||
瞬时 QPS：50000 → MQ（缓冲）
|
||
└─ 按数据库能力消费（5000 QPS）
|
||
|
||
结果：数据库平稳运行 ✅
|
||
```
|
||
|
||
**图解**：
|
||
```
|
||
不使用 MQ：
|
||
请求量 ███████████████████████████████████████
|
||
↑
|
||
数据库被打挂
|
||
|
||
使用 MQ：
|
||
请求量 ███████████████████████████████████████
|
||
↓
|
||
MQ 缓冲
|
||
↓
|
||
消费量 ████████████████████████████████████████
|
||
↑ 平滑消费
|
||
```
|
||
|
||
---
|
||
|
||
#### **使用场景**
|
||
|
||
| 场景 | 说明 | 示例 |
|
||
|------|------|------|
|
||
| **异步处理** | 非核心流程异步化 | 注册后发送邮件、短信 |
|
||
| **系统解耦** | 降低系统耦合度 | 订单系统 ↔ 物流系统 |
|
||
| **流量削峰** | 应对突发流量 | 秒杀、大促 |
|
||
| **日志收集** | 分布式日志收集 | 应用日志 → ELK |
|
||
| **事件驱动** | 基于事件的架构 | 用户行为触发推荐算法 |
|
||
| **数据同步** | 跨数据库同步 | MySQL → Elasticsearch |
|
||
| **分布式事务** | 最终一致性 | 订单 → 库存（Saga 模式） |
|
||
|
||
---
|
||
|
||
#### **消息队列的缺点**
|
||
|
||
1. **系统复杂度增加**
|
||
- 需要维护 MQ 集群
|
||
- 需要处理消息丢失、重复、顺序等问题
|
||
|
||
2. **数据一致性**
|
||
- 无法保证强一致性，只能保证最终一致性
|
||
|
||
3. **可用性降低**
|
||
- MQ 成为单点故障（需要高可用集群）
|
||
|
||
4. **消息延迟**
|
||
- 异步处理导致消息延迟
|
||
|
||
---
|
||
|
||
### 2. 保证消息不丢失
|
||
|
||
#### **三个层面的保障**
|
||
|
||
```
|
||
生产者 → Broker → 消费者
|
||
↓ ↓ ↓
|
||
三个层面都要保证
|
||
```
|
||
|
||
---
|
||
|
||
#### **层面 1：生产者（Producer）**
|
||
|
||
##### **方案 1：同步发送 + 确认机制**
|
||
|
||
**RocketMQ 示例**：
|
||
```java
|
||
// 同步发送（默认）
|
||
Message message = new Message("TopicTest", "TagA", "OrderID001", "Hello MQ".getBytes());
|
||
|
||
// 同步发送：等待 Broker 确认
|
||
SendResult sendResult = producer.send(message);
|
||
if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
|
||
// 发送成功
|
||
} else {
|
||
// 发送失败，重试或记录日志
|
||
log.error("消息发送失败: {}", sendResult);
|
||
}
|
||
```
|
||
|
||
**发送状态**：
|
||
- `SEND_OK`：发送成功
|
||
- `FLUSH_DISK_TIMEOUT`：刷盘超时
|
||
- `FLUSH_SLAVE_TIMEOUT`：同步到 Slave 超时
|
||
- `SLAVE_NOT_AVAILABLE`：Slave 不可用
|
||
|
||
---
|
||
|
||
##### **方案 2：异步发送 + 回调**
|
||
|
||
```java
|
||
producer.send(message, new SendCallback() {
|
||
@Override
|
||
public void onSuccess(SendResult sendResult) {
|
||
// 发送成功
|
||
log.info("消息发送成功: {}", sendResult);
|
||
}
|
||
|
||
@Override
|
||
public void onException(Throwable e) {
|
||
// 发送失败，重试或记录日志
|
||
log.error("消息发送失败", e);
|
||
}
|
||
});
|
||
```
|
||
|
||
---
|
||
|
||
##### **方案 3：事务消息（RocketMQ 独有）**
|
||
|
||
**原理**：
|
||
1. 发送半消息（Half Message，对消费者不可见）
|
||
2. 执行本地事务
|
||
3. 提交/回滚消息
|
||
|
||
```java
|
||
// 发送事务消息
|
||
TransactionMQProducer producer = new TransactionMQProducer("group1");
|
||
producer.setTransactionListener(new TransactionListener() {
|
||
@Override
|
||
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
|
||
// 执行本地事务（如：创建订单）
|
||
try {
|
||
orderService.createOrder(msg);
|
||
return LocalTransactionState.COMMIT_MESSAGE; // 提交消息
|
||
} catch (Exception e) {
|
||
return LocalTransactionState.ROLLBACK_MESSAGE; // 回滚消息
|
||
}
|
||
}
|
||
|
||
@Override
|
||
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
|
||
// 反查本地事务状态（Broker 未收到确认时调用）
|
||
if (orderService.exists(msg.getKeys())) {
|
||
return LocalTransactionState.COMMIT_MESSAGE;
|
||
} else {
|
||
return LocalTransactionState.ROLLBACK_MESSAGE;
|
||
}
|
||
}
|
||
});
|
||
|
||
// 发送消息
|
||
producer.sendMessageInTransaction(message, null);
|
||
```
|
||
|
||
---
|
||
|
||
##### **方案 4：本地消息表**
|
||
|
||
**原理**：
|
||
1. 在同一本地事务中：完成业务操作 + 插入消息到本地消息表
|
||
2. 定时任务扫描消息表，发送消息到 MQ
|
||
3. 发送成功后更新消息状态
|
||
|
||
```java
|
||
@Transactional
|
||
public void createOrder(Order order) {
|
||
// 1. 保存订单
|
||
orderMapper.insert(order);
|
||
|
||
// 2. 保存消息到本地消息表
|
||
Message message = new Message("OrderTopic", JSON.toJSONString(order).getBytes());
|
||
localMessageMapper.insert(new LocalMessage(order.getId(), message));
|
||
}
|
||
|
||
// 定时任务：发送消息到 MQ
|
||
@Scheduled(fixedRate = 5000)
|
||
public void sendPendingMessages() {
|
||
List<LocalMessage> messages = localMessageMapper.findPendingMessages();
|
||
for (LocalMessage msg : messages) {
|
||
try {
|
||
SendResult result = producer.send(msg.getMessage());
|
||
if (result.getSendStatus() == SendStatus.SEND_OK) {
|
||
// 更新消息状态为已发送
|
||
localMessageMapper.markAsSent(msg.getId());
|
||
}
|
||
} catch (Exception e) {
|
||
log.error("消息发送失败: {}", msg.getId(), e);
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
**优点**：
|
||
- 可靠性极高（消息持久化到数据库）
|
||
- 支持重试
|
||
|
||
**缺点**：
|
||
- 需要维护本地消息表
|
||
- 实现复杂
|
||
|
||
---
|
||
|
||
#### **层面 2：Broker（消息服务器）**
|
||
|
||
##### **方案 1：消息持久化**
|
||
|
||
**RocketMQ**：
|
||
```xml
|
||
<!-- 刷盘方式：同步刷盘 -->
|
||
<flushDiskType>SYNC_FLUSH</flushDiskType>
|
||
|
||
<!-- 或异步刷盘（性能更高，但可能丢失少量消息） -->
|
||
<flushDiskType>ASYNC_FLUSH</flushDiskType>
|
||
```
|
||
|
||
**同步刷盘 vs 异步刷盘**：
|
||
| 方式 | 性能 | 可靠性 | 适用场景 |
|
||
|------|------|--------|----------|
|
||
| SYNC_FLUSH | 低（写入磁盘后才返回成功） | 高 | 金融、支付 |
|
||
| ASYNC_FLUSH | 高（写入内存后返回成功） | 中 | 日志、监控 |
|
||
|
||
---
|
||
|
||
##### **方案 2：主从复制（HA）**
|
||
|
||
**RocketMQ 配置**：
|
||
```xml
|
||
<!-- 同步复制：Master 收到消息后，同步到 Slave 才返回成功 -->
|
||
<brokerRole>SYNC_MASTER</brokerRole>
|
||
|
||
<!-- 异步复制：Master 收到消息后立即返回，异步复制到 Slave -->
|
||
<brokerRole>ASYNC_MASTER</brokerRole>
|
||
```
|
||
|
||
**复制方式对比**：
|
||
| 方式 | 性能 | 可靠性 | 宕机丢数据 |
|
||
|------|------|--------|------------|
|
||
| SYNC_MASTER | 低 | 高 | 否 |
|
||
| ASYNC_MASTER | 高 | 中 | 可能 |
|
||
|
||
**推荐配置**（金融级可靠性）：
|
||
```xml
|
||
<flushDiskType>SYNC_FLUSH</flushDiskType>
|
||
<brokerRole>SYNC_MASTER</brokerRole>
|
||
```
|
||
|
||
---
|
||
|
||
##### **方案 3：多副本（Kafka）**
|
||
|
||
**Kafka 配置**：
|
||
```properties
|
||
# 副本数量（建议 ≥ 3）
|
||
default.replication.factor=3
|
||
|
||
# 最小同步副本数（生产者确认）
|
||
min.insync.replicas=2
|
||
|
||
# 不完全的选举首领（禁止）
|
||
unclean.leader.election.enable=false
|
||
```
|
||
|
||
**生产者配置**：
|
||
```java
|
||
Properties props = new Properties();
|
||
// 等待所有副本确认
|
||
props.put("acks", "all");
|
||
|
||
// 或至少等待主副本 + 1 个从副本确认
|
||
props.put("acks", "1");
|
||
```
|
||
|
||
**acks 参数**：
|
||
- `acks=0`：不等待确认（最快，可能丢失）
|
||
- `acks=1`：等待主副本确认（折中）
|
||
- `acks=all`（或 `-1`）：等待所有副本确认（最可靠）
|
||
|
||
---
|
||
|
||
#### **层面 3：消费者（Consumer）**
|
||
|
||
##### **方案 1：手动提交 Offset**
|
||
|
||
**RocketMQ 示例**：
|
||
```java
|
||
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
|
||
|
||
// 设置为手动提交
|
||
consumer.setConsumeMessageBatchMaxSize(1);
|
||
|
||
consumer.registerMessageListener(new MessageListenerConcurrently() {
|
||
@Override
|
||
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
|
||
for (MessageExt msg : msgs) {
|
||
try {
|
||
// 处理消息
|
||
processMessage(msg);
|
||
|
||
// 处理成功，返回 CONSUME_SUCCESS（自动提交 offset）
|
||
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
|
||
} catch (Exception e) {
|
||
// 处理失败，稍后重试
|
||
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
|
||
}
|
||
}
|
||
}
|
||
});
|
||
```
|
||
|
||
---
|
||
|
||
**Kafka 示例**：
|
||
```java
|
||
Properties props = new Properties();
|
||
// 关闭自动提交
|
||
props.put("enable.auto.commit", "false");
|
||
|
||
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
|
||
|
||
while (true) {
|
||
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
|
||
|
||
for (ConsumerRecord<String, String> record : records) {
|
||
try {
|
||
// 处理消息
|
||
processMessage(record);
|
||
|
||
// 手动提交 offset
|
||
consumer.commitSync();
|
||
} catch (Exception e) {
|
||
// 处理失败，不提交 offset，下次重试
|
||
log.error("消息处理失败: {}", record.offset(), e);
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
##### **方案 2：先处理消息，再提交 Offset**
|
||
|
||
```java
|
||
// 错误示例：先提交 offset，再处理消息
|
||
consumer.commitSync(); // ❌ 提交成功
|
||
processMessage(record); // ❌ 处理失败，消息丢失
|
||
|
||
// 正确示例：先处理消息，再提交 offset
|
||
processMessage(record); // ✅ 处理成功
|
||
consumer.commitSync(); // ✅ 提交 offset
|
||
```
|
||
|
||
---
|
||
|
||
##### **方案 3：死信队列（DLQ）**
|
||
|
||
**场景**：消息处理失败多次后，不再重试，放入死信队列。
|
||
|
||
**RocketMQ 示例**：
|
||
```java
|
||
consumer.registerMessageListener(new MessageListenerConcurrently() {
|
||
@Override
|
||
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
|
||
for (MessageExt msg : msgs) {
|
||
int reconsumeTimes = msg.getReconsumeTimes();
|
||
|
||
if (reconsumeTimes >= 3) {
|
||
// 重试 3 次后仍失败，放入死信队列
|
||
sendToDeadLetterQueue(msg);
|
||
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
|
||
}
|
||
|
||
try {
|
||
processMessage(msg);
|
||
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
|
||
} catch (Exception e) {
|
||
// 稍后重试
|
||
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
|
||
}
|
||
}
|
||
}
|
||
});
|
||
```
|
||
|
||
**Kafka 示例**：
|
||
```java
|
||
// 配置死信队列
|
||
Properties props = new Properties();
|
||
props.put("enable.deadletter", "true");
|
||
props.put("deadletter.queue.topic.prefix", "dlq.");
|
||
|
||
// 消费者处理失败时，发送到死信队列
|
||
if (retries >= 3) {
|
||
kafkaTemplate.send("dlq.MyTopic", record.value());
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
#### **总结：端到端的可靠投递**
|
||
|
||
```
|
||
生产者 Broker 消费者
|
||
↓ ↓ ↓
|
||
1. 同步发送 1. 持久化到磁盘 1. 手动提交 offset
|
||
2. 事务消息 2. 主从复制 2. 处理成功后才提交
|
||
3. 本地消息表 3. 多副本 3. 死信队列
|
||
```
|
||
|
||
---
|
||
|
||
### 3. 保证消息幂等性
|
||
|
||
#### **为什么会有重复消息？**
|
||
|
||
1. **生产者重复发送**：网络抖动导致生产者未收到 ACK，重试发送
|
||
2. **Broker 重复投递**：主从切换时，从副本尚未同步的消息被重复投递
|
||
3. **消费者重复消费**：消费成功后 offset 提交失败，下次重复消费
|
||
|
||
---
|
||
|
||
#### **幂等性解决方案**
|
||
|
||
##### **方案 1：数据库唯一约束**
|
||
|
||
**场景**：订单创建
|
||
|
||
```java
|
||
@Transactional
|
||
public void createOrder(Order order) {
|
||
// order_id 有唯一索引
|
||
try {
|
||
orderMapper.insert(order);
|
||
} catch (DuplicateKeyException e) {
|
||
// 重复订单，忽略
|
||
log.warn("重复订单: {}", order.getId());
|
||
return;
|
||
}
|
||
|
||
// 处理订单逻辑
|
||
processOrder(order);
|
||
}
|
||
```
|
||
|
||
**优点**：
|
||
- 实现简单
|
||
- 数据库保证唯一性
|
||
|
||
**缺点**：
|
||
- 依赖数据库
|
||
- 高并发下性能较差（频繁抛异常）
|
||
|
||
---
|
||
|
||
##### **方案 2：Redis 分布式锁 + 唯一标识**
|
||
|
||
**原理**：为每条消息设置唯一 ID（如 UUID），消费前先检查 Redis 是否已处理。
|
||
|
||
```java
|
||
public void consumeMessage(Message message) {
|
||
String messageId = message.getHeaders().get("messageId");
|
||
String lockKey = "message:processed:" + messageId;
|
||
|
||
// 尝试获取锁
|
||
Boolean locked = redisTemplate.opsForValue()
|
||
.setIfAbsent(lockKey, "1", 24, TimeUnit.HOURS);
|
||
|
||
if (!locked) {
|
||
// 消息已处理，跳过
|
||
log.warn("重复消息: {}", messageId);
|
||
return;
|
||
}
|
||
|
||
try {
|
||
// 处理消息
|
||
processMessage(message);
|
||
} catch (Exception e) {
|
||
// 处理失败，删除锁（允许重试）
|
||
redisTemplate.delete(lockKey);
|
||
throw e;
|
||
}
|
||
}
|
||
```
|
||
|
||
**优点**：
|
||
- 性能高（Redis 内存操作）
|
||
- 可设置过期时间（自动清理）
|
||
|
||
**缺点**：
|
||
- 依赖 Redis
|
||
- Redis 宕机可能导致问题
|
||
|
||
---
|
||
|
||
##### **方案 3：状态机**
|
||
|
||
**场景**：订单状态流转
|
||
|
||
```sql
|
||
-- 订单表
|
||
CREATE TABLE orders (
|
||
id BIGINT PRIMARY KEY,
|
||
status VARCHAR(20), -- CREATED, PAID, SHIPPED, COMPLETED
|
||
version INT, -- 乐观锁版本号
|
||
UNIQUE KEY id_status (id, status) -- 联合唯一索引
|
||
);
|
||
```
|
||
|
||
```java
|
||
@Transactional
|
||
public void updateOrderStatus(Long orderId, String fromStatus, String toStatus) {
|
||
// 使用 CAS（Compare And Set）更新状态
|
||
int updated = orderMapper.updateStatus(orderId, fromStatus, toStatus);
|
||
|
||
if (updated == 0) {
|
||
// 状态不匹配，可能是重复消息
|
||
log.warn("订单状态已变更: orderId={}, fromStatus={}, toStatus={}",
|
||
orderId, fromStatus, toStatus);
|
||
return;
|
||
}
|
||
|
||
// 处理后续逻辑
|
||
processStatusChange(orderId, toStatus);
|
||
}
|
||
|
||
// Mapper
|
||
@Update("UPDATE orders SET status = #{toStatus}, version = version + 1 " +
|
||
"WHERE id = #{orderId} AND status = #{fromStatus}")
|
||
int updateStatus(@Param("orderId") Long orderId,
|
||
@Param("fromStatus") String fromStatus,
|
||
@Param("toStatus") String toStatus);
|
||
```
|
||
|
||
**优点**：
|
||
- 状态机保证幂等性
|
||
- 支持复杂业务逻辑
|
||
|
||
**缺点**：
|
||
- 设计复杂
|
||
- 需要明确状态流转
|
||
|
||
---
|
||
|
||
##### **方案 4：去重表**
|
||
|
||
**原理**：创建专门的去重表，记录已处理的消息 ID。
|
||
|
||
```sql
|
||
-- 去重表
|
||
CREATE TABLE message_dedup (
|
||
id BIGINT AUTO_INCREMENT PRIMARY KEY,
|
||
message_id VARCHAR(64) NOT NULL,
|
||
topic VARCHAR(64),
|
||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
UNIQUE KEY uk_message_id (message_id)
|
||
);
|
||
```
|
||
|
||
```java
|
||
@Transactional
|
||
public void consumeMessage(Message message) {
|
||
String messageId = message.getId();
|
||
|
||
// 插入去重表
|
||
try {
|
||
dedupMapper.insert(new MessageDedup(messageId, message.getTopic()));
|
||
} catch (DuplicateKeyException e) {
|
||
// 消息已处理，跳过
|
||
log.warn("重复消息: {}", messageId);
|
||
return;
|
||
}
|
||
|
||
// 处理消息
|
||
processMessage(message);
|
||
}
|
||
```
|
||
|
||
**优点**：
|
||
- 简单直观
|
||
- 可查询处理历史
|
||
|
||
**缺点**：
|
||
- 额外的存储开销
|
||
- 需要定期清理历史数据
|
||
|
||
---
|
||
|
||
##### **方案 5：多版本并发控制（MVCC）**
|
||
|
||
**原理**：为数据增加版本号，更新时检查版本号。
|
||
|
||
```sql
|
||
-- 订单表
|
||
CREATE TABLE orders (
|
||
id BIGINT PRIMARY KEY,
|
||
amount DECIMAL(10,2),
|
||
version INT DEFAULT 0, -- 版本号
|
||
updated_at DATETIME
|
||
);
|
||
```
|
||
|
||
```java
|
||
@Transactional
|
||
public void updateOrderAmount(Long orderId, BigDecimal newAmount) {
|
||
// 使用乐观锁更新
|
||
int updated = orderMapper.updateAmountWithVersion(orderId, newAmount);
|
||
|
||
if (updated == 0) {
|
||
// 版本号不匹配，可能是重复消息
|
||
log.warn("订单已被更新: orderId={}", orderId);
|
||
return;
|
||
}
|
||
|
||
// 处理后续逻辑
|
||
processAmountChange(orderId, newAmount);
|
||
}
|
||
|
||
// Mapper
|
||
@Update("UPDATE orders SET amount = #{amount}, version = version + 1 " +
|
||
"WHERE id = #{orderId} AND version = #{version}")
|
||
int updateAmountWithVersion(@Param("orderId") Long orderId,
|
||
@Param("amount") BigDecimal amount,
|
||
@Param("version") Integer version);
|
||
```
|
||
|
||
**优点**：
|
||
- 无锁，性能高
|
||
- 支持高并发
|
||
|
||
**缺点**：
|
||
- 更新失败时需要重试
|
||
- 不适合写冲突频繁的场景
|
||
|
||
---
|
||
|
||
#### **幂等性方案对比**
|
||
|
||
| 方案 | 性能 | 复杂度 | 可靠性 | 适用场景 |
|
||
|------|------|--------|--------|----------|
|
||
| 数据库唯一约束 | 低 | 低 | 高 | 低并发、简单场景 |
|
||
| Redis 分布式锁 | 高 | 中 | 中 | 高并发、通用场景 |
|
||
| 状态机 | 中 | 高 | 高 | 订单、工作流等 |
|
||
| 去重表 | 中 | 低 | 高 | 需要记录历史的场景 |
|
||
| MVCC | 高 | 中 | 高 | 高并发、读多写少 |
|
||
|
||
---
|
||
|
||
### 4. 保证消息顺序性
|
||
|
||
#### **为什么消息会乱序？**
|
||
|
||
**场景**：一个 Topic 有多个 Queue（分区），多个消费者并发消费。
|
||
|
||
```
|
||
Topic: OrderTopic
|
||
├─ Queue1: 消息1, 消息4, 消息7
|
||
├─ Queue2: 消息2, 消息5, 消息8
|
||
└─ Queue3: 消息3, 消息6, 消息9
|
||
|
||
消费者1消费Queue1，消费者2消费Queue2，消费者3消费Queue3
|
||
└─ 消费速度不同，导致消息乱序
|
||
```
|
||
|
||
---
|
||
|
||
#### **解决方案**
|
||
|
||
##### **方案 1：单分区（单 Queue）**
|
||
|
||
**原理**：一个 Topic 只有一个 Queue，保证消息按顺序消费。
|
||
|
||
```java
|
||
// RocketMQ 生产者：发送到同一个 Queue
|
||
SendResult result = producer.send(message, new MessageQueueSelector() {
|
||
@Override
|
||
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
|
||
// 始终选择第一个 Queue
|
||
return mqs.get(0);
|
||
}
|
||
}, null);
|
||
|
||
// RocketMQ 消费者：单线程消费
|
||
consumer.setConsumeThreadMin(1);
|
||
consumer.setConsumeThreadMax(1);
|
||
```
|
||
|
||
**Kafka 配置**：
|
||
```properties
|
||
# Topic 只有 1 个分区
|
||
num.partitions=1
|
||
```
|
||
|
||
**优点**：
|
||
- 实现简单
|
||
- 严格保证顺序
|
||
|
||
**缺点**：
|
||
- 性能差（无法并行消费）
|
||
- 成为瓶颈
|
||
|
||
**适用场景**：
|
||
- 低并发、对顺序要求极高的场景
|
||
|
||
---
|
||
|
||
##### **方案 2：按业务 Key 分区**
|
||
|
||
**原理**：相同业务 Key 的消息发送到同一个 Queue。
|
||
|
||
**场景**：订单号相同的消息需要顺序消费。
|
||
|
||
**RocketMQ 示例**：
|
||
```java
|
||
// 生产者：按订单 ID 分区
|
||
SendResult result = producer.send(message, new MessageQueueSelector() {
|
||
@Override
|
||
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
|
||
Long orderId = (Long) arg;
|
||
// 相同订单 ID 的消息发送到同一个 Queue
|
||
int index = (int) (orderId % mqs.size());
|
||
return mqs.get(index);
|
||
}
|
||
}, orderId);
|
||
|
||
// 消费者：单线程消费每个 Queue（或至少保证相同订单的消息顺序处理）
|
||
consumer.registerMessageListener(new MessageListenerOrderly() {
|
||
@Override
|
||
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
|
||
// MessageListenerOrderly 保证顺序消费
|
||
for (MessageExt msg : msgs) {
|
||
processMessage(msg);
|
||
}
|
||
return ConsumeOrderlyStatus.SUCCESS;
|
||
}
|
||
});
|
||
```
|
||
|
||
**Kafka 示例**：
|
||
```java
|
||
// 生产者：按订单 ID 分区
|
||
Properties props = new Properties();
|
||
props.put("partitioner.class", "com.example.OrderPartitioner");
|
||
|
||
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
|
||
|
||
// 发送消息（相同订单 ID 的消息会发送到同一个分区）
|
||
producer.send(new ProducerRecord<>("OrderTopic", orderId, message));
|
||
|
||
// 自定义分区器
|
||
public class OrderPartitioner implements Partitioner {
|
||
@Override
|
||
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
|
||
Long orderId = Long.parseLong(key.toString());
|
||
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
|
||
return (int) (orderId % partitions.size());
|
||
}
|
||
}
|
||
```
|
||
|
||
**优点**：
|
||
- 相同业务 Key 的消息保证顺序
|
||
- 不同业务 Key 的消息可并行消费
|
||
- 性能较好
|
||
|
||
**缺点**：
|
||
- 需要选择合适的分区 Key
|
||
- 分区数量固定，扩容困难
|
||
|
||
---
|
||
|
||
##### **方案 3：单线程消费 + 内存队列**
|
||
|
||
**原理**：消费者内部按业务 Key 分组，使用多个内存队列 + 多个线程处理。
|
||
|
||
**架构图**：
|
||
```
|
||
MQ 消费者（单线程拉取）
|
||
↓
|
||
按 Key 分组
|
||
↓
|
||
┌───────────┬───────────┬───────────┐
|
||
│ 内存队列1 │ 内存队列2 │ 内存队列3 │
|
||
│ (订单100) │ (订单200) │ (订单300) │
|
||
└───────────┴───────────┴───────────┘
|
||
↓ ↓ ↓
|
||
工作线程1 工作线程2 工作线程3
|
||
```
|
||
|
||
**代码示例**：
|
||
```java
|
||
@Component
|
||
public class OrderMessageConsumer {
|
||
|
||
private final Map<Long, BlockingQueue<Message>> keyQueues = new ConcurrentHashMap<>();
|
||
private final ExecutorService executorService = Executors.newFixedThreadPool(10);
|
||
|
||
@RocketMQMessageListener(topic = "OrderTopic", consumerGroup = "order-group")
|
||
public void onMessage(MessageExt message) {
|
||
Long orderId = parseOrderId(message);
|
||
|
||
// 获取或创建该订单的内存队列
|
||
BlockingQueue<Message> queue = keyQueues.computeIfAbsent(orderId,
|
||
k -> new LinkedBlockingQueue<>()
|
||
);
|
||
|
||
// 消息放入队列
|
||
queue.offer(message);
|
||
|
||
// 提交异步任务处理
|
||
executorService.submit(() -> {
|
||
Message msg = queue.poll();
|
||
if (msg != null) {
|
||
processMessage(msg);
|
||
}
|
||
});
|
||
}
|
||
}
|
||
```
|
||
|
||
**优点**：
|
||
- 保证相同 Key 的消息顺序
|
||
- 不同 Key 的消息并行处理
|
||
- 性能好
|
||
|
||
**缺点**：
|
||
- 实现复杂
|
||
- 内存占用较高
|
||
|
||
---
|
||
|
||
#### **消息顺序性总结**
|
||
|
||
| 方案 | 性能 | 复杂度 | 顺序粒度 | 适用场景 |
|
||
|------|------|--------|----------|----------|
|
||
| 单分区 | 低 | 低 | 全局顺序 | 低并发、全局顺序 |
|
||
| 按 Key 分区 | 高 | 中 | Key 级顺序 | 订单、用户等业务 |
|
||
| 内存队列 | 高 | 高 | Key 级顺序 | 高并发、复杂业务 |
|
||
|
||
---
|
||
|
||
### 5. 消息积压处理
|
||
|
||
#### **消息积压的原因**
|
||
|
||
1. **消费者消费速度慢**：
|
||
- 消费逻辑复杂
|
||
- 下游服务响应慢
|
||
- 数据库操作慢
|
||
|
||
2. **生产者发送速度过快**：
|
||
- 突发流量
|
||
- 秒杀活动
|
||
|
||
3. **消费者故障**：
|
||
- 消费者宕机
|
||
- 网络问题
|
||
|
||
4. **消费者消费失败**：
|
||
- 消息格式错误
|
||
- 业务异常
|
||
- 无限重试
|
||
|
||
---
|
||
|
||
#### **监控消息积压**
|
||
|
||
**RocketMQ**：
|
||
```bash
|
||
# 查看消费者堆积情况
|
||
sh mqadmin queryConsumeQueue -t OrderTopic -n localhost:9876
|
||
|
||
# 查看消费者组
|
||
sh mqadmin consumerList -g order-group -n localhost:9876
|
||
|
||
# 查看消费进度
|
||
sh mqadmin consumerProgress -g order-group -n localhost:9876
|
||
```
|
||
|
||
**输出示例**：
|
||
```
|
||
# 消费进度
|
||
Topic: OrderTopic
|
||
Broker: broker-a
|
||
Queue: 0
|
||
Diff: 1000000 ← 积压 100 万条
|
||
LastOffset: 5000000
|
||
```
|
||
|
||
**Kafka**：
|
||
```bash
|
||
# 查看消费者 lag
|
||
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
|
||
--group order-group --describe
|
||
|
||
# 输出
|
||
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
|
||
OrderTopic 0 5000000 6000000 1000000
|
||
↑
|
||
积压 100 万
|
||
```
|
||
|
||
---
|
||
|
||
#### **解决方案**
|
||
|
||
##### **方案 1：增加消费者数量（横向扩容）**
|
||
|
||
**原理**：增加消费者实例，提高消费能力。
|
||
|
||
**RocketMQ 示例**：
|
||
```java
|
||
// 原来：1 个消费者实例
|
||
@RocketMQMessageListener(
|
||
topic = "OrderTopic",
|
||
consumerGroup = "order-group",
|
||
consumeThreadNumber = 1 // 1 个线程
|
||
)
|
||
|
||
// 扩容后：10 个消费者实例（部署 10 个 Pod）
|
||
// 每个 Pod 有 1 个线程，总消费能力 ×10
|
||
```
|
||
|
||
**注意事项**：
|
||
- **消费者数量 ≤ Queue 数量**
|
||
- 如果消费者数量 > Queue 数量，部分消费者会空闲
|
||
|
||
**计算公式**：
|
||
```
|
||
所需消费者数 = 消息积压量 / (单个消费者消费速度 × 预期清理时间)
|
||
|
||
示例：
|
||
积压 100 万条，单个消费者每秒消费 100 条，预期 1 小时清理
|
||
所需消费者数 = 1000000 / (100 × 3600) ≈ 3 个
|
||
```
|
||
|
||
---
|
||
|
||
##### **方案 2：增加消费线程（纵向扩容）**
|
||
|
||
**RocketMQ 示例**：
|
||
```java
|
||
@RocketMQMessageListener(
|
||
topic = "OrderTopic",
|
||
consumerGroup = "order-group",
|
||
consumeThreadNumber = 20 // 从 1 增加到 20 个线程
|
||
)
|
||
```
|
||
|
||
**Kafka 示例**：
|
||
```java
|
||
Properties props = new Properties();
|
||
// 增加消费线程数
|
||
props.put("max.poll.records", 500); // 每次拉取 500 条
|
||
props.put("max.poll.interval.ms", 300000); // 最大处理时间 5 分钟
|
||
```
|
||
|
||
**注意事项**：
|
||
- 线程数不是越多越好（CPU、数据库连接数限制）
|
||
- 需要测试找到最优值
|
||
|
||
---
|
||
|
||
##### **方案 3：优化消费逻辑**
|
||
|
||
**问题代码**：
|
||
```java
|
||
// ❌ 在循环中逐条处理数据库
|
||
public void consumeMessage(List<Message> messages) {
|
||
for (Message msg : messages) {
|
||
// 逐条插入数据库（慢）
|
||
orderMapper.insert(parseOrder(msg));
|
||
}
|
||
}
|
||
```
|
||
|
||
**优化代码**：
|
||
```java
|
||
// ✅ 批量处理数据库
|
||
public void consumeMessage(List<Message> messages) {
|
||
List<Order> orders = messages.stream()
|
||
.map(this::parseOrder)
|
||
.collect(Collectors.toList());
|
||
|
||
// 批量插入（快 10-100 倍）
|
||
orderMapper.batchInsert(orders);
|
||
}
|
||
```
|
||
|
||
**Mapper**：
|
||
```java
|
||
@Insert({
|
||
"<script>",
|
||
"INSERT INTO orders (id, user_id, amount) VALUES ",
|
||
"<foreach collection='orders' item='order' separator=','>",
|
||
"(#{order.id}, #{order.userId}, #{order.amount})",
|
||
"</foreach>",
|
||
"</script>"
|
||
})
|
||
int batchInsert(@Param("orders") List<Order> orders);
|
||
```
|
||
|
||
---
|
||
|
||
##### **方案 4：临时扩容方案**
|
||
|
||
**场景**：积压严重（如 1000 万条），正常消费需要几天。
|
||
|
||
**步骤**：
|
||
|
||
1. **创建临时 Topic**（大量 Queue）：
|
||
```bash
|
||
# 创建 OrderTopic_Temp，有 100 个 Queue
|
||
```
|
||
|
||
2. **编写转发程序**：
|
||
```java
|
||
// 从原 Topic 消费，发送到临时 Topic
|
||
@Component
|
||
public class MessageForwarder {
|
||
|
||
@Autowired
|
||
private RocketMQTemplate rocketMQTemplate;
|
||
|
||
@RocketMQMessageListener(topic = "OrderTopic", consumerGroup = "forward-group")
|
||
public void onMessage(MessageExt message) {
|
||
// 转发到临时 Topic
|
||
rocketMQTemplate.send("OrderTopic_Temp", message);
|
||
}
|
||
}
|
||
```
|
||
|
||
3. **部署大量消费者**消费临时 Topic：
|
||
```java
|
||
// 部署 100 个消费者实例，每个消费 1 个 Queue
|
||
@RocketMQMessageListener(
|
||
topic = "OrderTopic_Temp",
|
||
consumerGroup = "temp-consumer-group"
|
||
)
|
||
```
|
||
|
||
4. **积压清理完毕后，恢复正常架构**。
|
||
|
||
---
|
||
|
||
##### **方案 5：丢弃部分消息**
|
||
|
||
**场景**：
|
||
- 消息积压过于严重（如 1 亿条）
|
||
- 消息时效性要求高（如实时日志）
|
||
- 允许部分数据丢失
|
||
|
||
**实现**：
|
||
```java
|
||
// 1. 创建临时消费者，只消费不处理（快速跳过）
|
||
@RocketMQMessageListener(
|
||
topic = "OrderTopic",
|
||
consumerGroup = "skip-group"
|
||
)
|
||
public void onMessage(MessageExt message) {
|
||
// 直接返回成功，不处理消息
|
||
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
|
||
}
|
||
|
||
// 2. 跳到指定 offset
|
||
consumer.seekToBegin(); // 跳到最前
|
||
consumer.seek(queue, 10000000); // 跳到第 1000 万条
|
||
```
|
||
|
||
---
|
||
|
||
#### **消息积压处理流程**
|
||
|
||
```
|
||
1. 监控发现积压
|
||
↓
|
||
2. 分析积压原因
|
||
├─ 消费者慢 → 优化消费逻辑
|
||
├─ 消费者少 → 横向扩容
|
||
├─ 生产者快 → 限流
|
||
└─ 消费失败 → 修复 bug，跳过错误消息
|
||
↓
|
||
3. 执行扩容或优化
|
||
↓
|
||
4. 监控清理进度
|
||
↓
|
||
5. 恢复正常架构
|
||
```
|
||
|
||
---
|
||
|
||
### 6. RocketMQ vs Kafka
|
||
|
||
#### **核心特性对比**
|
||
|
||
| 特性 | RocketMQ | Kafka |
|
||
|------|----------|-------|
|
||
| **开发语言** | Java | Scala/Java |
|
||
| **协议** | 自有协议 | TCP |
|
||
| **消息模型** | 队列模型 | 发布-订阅模型 |
|
||
| **消息顺序** | 支持严格顺序 | 分区内有序 |
|
||
| **事务消息** | ✅ 支持 | ❌ 不支持（KIP-98 有提案） |
|
||
| **定时消息** | ✅ 支持 | ❌ 不支持 |
|
||
| **延迟消息** | ✅ 支持（18 级） | ❌ 不支持 |
|
||
| **消息回溯** | ✅ 支持 | ✅ 支持（需设置） |
|
||
| **消息过滤** | SQL92 表达式 | 客户端过滤 |
|
||
| **吞吐量** | 万级/单机 | 十万级/单机 |
|
||
| **延迟** | ms 级 | ms 级 |
|
||
| **可靠性** | 高（同步刷盘 + 主从） | 高（多副本） |
|
||
| **运维复杂度** | 中 | 中 |
|
||
| **社区活跃度** | 中（阿里主导） | 高（Confluent 支持） |
|
||
| **文档质量** | 中（中文文档多） | 高 |
|
||
| **适用场景** | 业务消息、订单、金融 | 日志收集、流式处理 |
|
||
|
||
---
|
||
|
||
#### **详细对比**
|
||
|
||
##### **1. 消息模型**
|
||
|
||
**RocketMQ**：
|
||
```
|
||
Topic: OrderTopic
|
||
├─ Queue1
|
||
├─ Queue2
|
||
├─ Queue3
|
||
└─ Queue4
|
||
|
||
消费者组：order-group
|
||
├─ 消费者1 → Queue1, Queue2
|
||
└─ 消费者2 → Queue3, Queue4
|
||
|
||
一条消息只能被同一个消费者组的一个消费者消费
|
||
```
|
||
|
||
**Kafka**：
|
||
```
|
||
Topic: OrderTopic
|
||
├─ Partition1
|
||
│ ├─ Replica1 (Leader)
|
||
│ ├─ Replica2 (Follower)
|
||
│ └─ Replica3 (Follower)
|
||
├─ Partition2
|
||
│ ├─ Replica1 (Leader)
|
||
│ ├─ Replica2 (Follower)
|
||
│ └─ Replica3 (Follower)
|
||
└─ ...
|
||
|
||
消费者组：order-group
|
||
├─ 消费者1 → Partition1
|
||
└─ 消费者2 → Partition2
|
||
|
||
一条消息只能被同一个消费者组的一个消费者消费
|
||
```
|
||
|
||
**区别**：
|
||
- RocketMQ 的 Queue 是逻辑概念
|
||
- Kafka 的 Partition 是物理概念（有副本）
|
||
|
||
---
|
||
|
||
##### **2. 消息存储**
|
||
|
||
**RocketMQ**：
|
||
```
|
||
CommitLog:
|
||
├─ 所有消息顺序写入（混合存储）
|
||
├─ 顺序写，性能高
|
||
└─ 通过 ConsumeQueue 索引快速查找
|
||
|
||
ConsumeQueue（索引文件）:
|
||
├─ 消息 1: CommitLog 偏移量 100, 大小 200, TagsCode
|
||
├─ 消息 2: CommitLog 偏移量 300, 大小 150, TagsCode
|
||
└─ 消息 3: CommitLog 偏移量 450, 大小 180, TagsCode
|
||
```
|
||
|
||
**Kafka**：
|
||
```
|
||
Partition:
|
||
├─ Segment 0
|
||
│ ├─ 00000000000000000000.log (消息数据)
|
||
│ ├─ 00000000000000000000.index (索引)
|
||
│ └─ 00000000000000000000.timeindex (时间索引)
|
||
├─ Segment 1
|
||
│ ├─ 00000000000000050000.log
|
||
│ ├─ 00000000000000050000.index
|
||
│ └─ 00000000000000050000.timeindex
|
||
└─ ...
|
||
|
||
每个 Partition 独立存储
|
||
```
|
||
|
||
**对比**：
|
||
- RocketMQ：所有消息混合存储，通过索引文件查找
|
||
- Kafka：每个 Partition 独立存储，支持 Segment 滚动
|
||
|
||
---
|
||
|
||
##### **3. 消息过滤**
|
||
|
||
**RocketMQ**（服务端过滤）：
|
||
```java
|
||
// 消费者订阅时指定过滤条件
|
||
consumer.subscribe("OrderTopic", MessageSelector.bySql("amount > 100 AND status = 'PAID'"));
|
||
|
||
// 生产者发送消息时设置属性
|
||
Message msg = new Message("OrderTopic", "Order", body);
|
||
msg.putUserProperty("amount", "200");
|
||
msg.putUserProperty("status", "PAID");
|
||
producer.send(msg);
|
||
```
|
||
|
||
**Kafka**（客户端过滤）：
|
||
```java
|
||
// 消费者消费后手动过滤
|
||
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
|
||
consumer.subscribe(Arrays.asList("OrderTopic"));
|
||
|
||
while (true) {
|
||
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
|
||
for (ConsumerRecord<String, String> record : records) {
|
||
// 客户端过滤（浪费网络带宽）
|
||
if (record.value().contains("amount:200")) {
|
||
processMessage(record);
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
**区别**：
|
||
- RocketMQ：服务端过滤（节省带宽）
|
||
- Kafka：客户端过滤（浪费带宽）
|
||
|
||
---
|
||
|
||
##### **4. 事务消息**
|
||
|
||
**RocketMQ**：
|
||
```java
|
||
// 1. 发送半消息
|
||
Message msg = new Message("OrderTopic", orderJson.getBytes());
|
||
TransactionSendResult result = producer.sendMessageInTransaction(msg, null);
|
||
|
||
// 2. 执行本地事务（如创建订单）
|
||
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
|
||
try {
|
||
orderService.createOrder(msg);
|
||
return LocalTransactionState.COMMIT_MESSAGE; // 提交消息
|
||
} catch (Exception e) {
|
||
return LocalTransactionState.ROLLBACK_MESSAGE; // 回滚消息
|
||
}
|
||
}
|
||
|
||
// 3. 提交/回滚消息
|
||
// 消费者可以正常消费
|
||
```
|
||
|
||
**Kafka**：
|
||
```
|
||
❌ 不支持事务消息（KIP-98 有提案，但未正式发布）
|
||
|
||
✅ 支持"精确一次"语义（Exactly-Once），但仅限于流式处理（Kafka Streams）
|
||
```
|
||
|
||
---
|
||
|
||
##### **5. 延迟消息**
|
||
|
||
**RocketMQ**：
|
||
```java
|
||
// 18 个默认延迟级别（1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h）
|
||
Message msg = new Message("OrderTopic", body);
|
||
msg.setDelayTimeLevel(3); // 延迟 10 秒
|
||
producer.send(msg);
|
||
```
|
||
|
||
**Kafka**：
|
||
```java
|
||
// ❌ 不支持延迟消息
|
||
|
||
// ✅ 通过时间轮自定义实现
|
||
```
|
||
|
||
---
|
||
|
||
##### **6. 吞吐量**
|
||
|
||
**测试环境**：
|
||
- 机器：4 核 8G
|
||
- 消息大小：1 KB
|
||
|
||
**结果**：
|
||
| MQ | 单机吞吐量 | 延迟 |
|
||
|----|-----------|------|
|
||
| RocketMQ | 10-15 万/秒 | 1-5 ms |
|
||
| Kafka | 30-50 万/秒 | 1-5 ms |
|
||
|
||
**原因**：
|
||
- Kafka：零拷贝（sendfile）、批量处理、更好的磁盘顺序写
|
||
- RocketMQ：也有零拷贝，但 Java GC 影响性能
|
||
|
||
---
|
||
|
||
#### **选型建议**
|
||
|
||
##### **选择 RocketMQ 的场景**
|
||
|
||
1. **需要事务消息**：
|
||
- 订单系统（下单 → 扣库存）
|
||
- 支付系统（支付 → 通知）
|
||
|
||
2. **需要延迟消息**：
|
||
- 定时任务
|
||
- 超时取消订单
|
||
|
||
3. **需要严格的消息顺序**：
|
||
- 金融交易
|
||
- 订单状态流转
|
||
|
||
4. **需要消息回溯**：
|
||
- 数据恢复
|
||
- 数据重放
|
||
|
||
5. **业务复杂、需要高可靠性**：
|
||
- 金融、支付、电商
|
||
|
||
---
|
||
|
||
##### **选择 Kafka 的场景**
|
||
|
||
1. **日志收集**：
|
||
- 应用日志 → ELK
|
||
- 用户行为日志
|
||
|
||
2. **流式处理**：
|
||
- 实时数据分析
|
||
- 实时推荐
|
||
|
||
3. **高吞吐量场景**：
|
||
- 日志收集（百万级/秒）
|
||
- 监控数据收集
|
||
|
||
4. **已有 Kafka 生态**：
|
||
- 使用 Flink、Spark Streaming 等流式计算框架
|
||
|
||
---
|
||
|
||
#### **企业级选型案例**
|
||
|
||
| 公司 | 场景 | 选择 | 原因 |
|
||
|------|------|------|------|
|
||
| 阿里巴巴 | 订单、支付 | RocketMQ | 事务消息、高可靠性 |
|
||
| 美团 | 订单、物流 | RocketMQ | 业务复杂、延迟消息 |
|
||
| 字节跳动 | 日志、推荐 | Kafka | 高吞吐、流式处理 |
|
||
| 携程 | 订单、日志 | 两者都用 | 订单用 RocketMQ，日志用 Kafka |
|
||
| 京东 | 订单、支付 | RocketMQ | 事务消息、高可靠性 |
|
||
|
||
---
|
||
|
||
### 7. 总结
|
||
|
||
#### **消息队列核心问题**
|
||
|
||
| 问题 | 解决方案 |
|
||
|------|----------|
|
||
| **消息不丢失** | 生产者（同步发送、事务消息、本地消息表）<br>Broker（持久化、主从、多副本）<br>消费者（手动提交 offset、死信队列） |
|
||
| **消息重复消费** | 数据库唯一约束、Redis 分布式锁、状态机、去重表、MVCC |
|
||
| **消息顺序性** | 单分区、按 Key 分区、内存队列 |
|
||
| **消息积压** | 横向扩容、纵向扩容、优化消费逻辑、临时 Topic、丢弃消息 |
|
||
| **高可用** | 集群部署、主从复制、多副本、故障自动转移 |
|
||
|
||
---
|
||
|
||
### 8. 阿里 P7 加分项
|
||
|
||
**深度理解**：
|
||
- 理解 RocketMQ 的 CommitLog 和 ConsumeQueue 设计
|
||
- 理解 Kafka 的 Partition、Segment、副本同步机制
|
||
- 理解零拷贝（sendfile）、mmap 等底层技术
|
||
|
||
**实战经验**：
|
||
- 有将消息积压从千万级清理到正常的经验
|
||
- 有处理消息丢失、重复消费的线上故障经验
|
||
- 有消息队列性能调优经验（JVM 参数、操作系统参数）
|
||
|
||
**架构能力**：
|
||
- 能设计跨数据中心的消息队列架构
|
||
- 能设计消息队列的监控和告警体系
|
||
- 有消息队列迁移经验（如从 ActiveMQ 迁移到 RocketMQ）
|
||
|
||
**技术选型**：
|
||
- 能根据业务特点选择合适的 MQ
|
||
- 有多种 MQ 混用的经验
|
||
- 了解 MQTT、AMQP 等其他消息协议
|