# rocketmq-demo

**Repository Path**: johnstonguo/rocketmq-demo

## Basic Information

- **Project Name**: rocketmq-demo
- **Description**: 小龙虾生成的demo，有修改
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-03-21
- **Last Updated**: 2026-08-01

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# RocketMQ 完整示例项目

一个结构清晰、功能完整的 RocketMQ Spring Boot 示例项目，涵盖所有常用业务场景。

## 📚 目录导航

- [快速开始](#快速开始)
- [基础场景](#基础场景)
- [高级场景](#高级场景)
- [容错处理](#容错处理)
- [业务实战](#业务实战)
- [配置说明](#配置说明)
- [常见问题](#常见问题)

---

## 快速开始

### 1. 环境要求

- **JDK**: 17+
- **Maven**: 3.6+
- **RocketMQ**: 5.1.0+

### 2. 启动 RocketMQ

#### Windows 环境

```bash
# 下载 RocketMQ (从官网下载后解压到合适位置)
# 假设解压到：D:\rocketmq-all-5.1.0-bin-release

# 进入目录
cd D:\rocketmq-all-5.1.0-bin-release

# 启动 Name Server (双击运行或命令行)
bin\mqnamesrv.cmd

# 启动 Broker (新开一个命令行窗口)
bin\mqbroker.cmd -n 127.0.0.1:9876
```

#### Linux/Mac 环境

```bash
# 下载并解压
wget https://archive.apache.org/dist/rocketmq/5.1.0/rocketmq-all-5.1.0-bin-release.zip
unzip rocketmq-all-5.1.0-bin-release.zip
cd rocketmq-all-5.1.0-bin-release

# 启动 Name Server
nohup sh bin/mqnamesrv &

# 启动 Broker
nohup sh bin/mqbroker -n 127.0.0.1:9876 &
```

### 3. 创建 Topic

```bash
# 方式一：使用脚本批量创建（推荐）
sh create-topics.sh

# 方式二：手动创建单个 Topic
./bin/mqadmin updateTopic -n 127.0.0.1:9876 -t sync-topic -c DefaultCluster
./bin/mqadmin updateTopic -n 127.0.0.1:9876 -t async-topic -c DefaultCluster
./bin/mqadmin updateTopic -n 127.0.0.1:9876 -t ordered-topic -c DefaultCluster
./bin/mqadmin updateTopic -n 127.0.0.1:9876 -t delay-topic -c DefaultCluster
./bin/mqadmin updateTopic -n 127.0.0.1:9876 -t batch-topic -c DefaultCluster
./bin/mqadmin updateTopic -n 127.0.0.1:9876 -t filter-topic -c DefaultCluster
./bin/mqadmin updateTopic -n 127.0.0.1:9876 -t transaction-topic -c DefaultCluster
./bin/mqadmin updateTopic -n 127.0.0.1:9876 -t order-topic -c DefaultCluster
./bin/mqadmin updateTopic -n 127.0.0.1:9876 -t inventory-topic -c DefaultCluster
./bin/mqadmin updateTopic -n 127.0.0.1:9876 -t payment-topic -c DefaultCluster
```

### 4. 构建运行

```bash
# 编译项目
mvn clean package -DskipTests

# 运行应用
java -jar target/rocketmq-demo-1.0.0.jar

# 或者使用 Maven 插件运行
mvn spring-boot:run
```

### 5. 验证

访问测试接口验证是否正常运行：

```bash
# 发送同步消息
curl -X POST "http://localhost:18090/api/messages/send?topic=sync-topic&message=Hello"

# 查看控制台输出，确认消息被消费
```

---

## 基础场景

### 1. 同步发送 (SyncProducer)

**适用场景**：需要确认消息是否发送成功，如重要通知、订单状态变更等

**代码位置**：`src/main/java/com/demo/rocketmq/scenario/basic/SyncProducer.java`

**使用示例**：

```java
@Autowired
private SyncProducer syncProducer;

// 发送简单消息
SendResult result = syncProducer.sendSync("sync-topic", "Hello RocketMQ");
System.out.println("发送成功：MsgId=" + result.msgId());

// 发送带 Tag 的消息（用于分类过滤）
result = syncProducer.sendSync("order-topic", "{\"orderId\":\"123\"}", "ORDER_CREATED");

// 发送带 Key 的消息（用于查询）
result = syncProducer.sendSyncWithKey("order-topic", "{\"orderId\":\"123\"}", "ORDER_123");
```

**控制台输出示例**：
```
[SyncProducer] Sending sync message to: sync-topic
[SyncProducer] Message: Hello RocketMQ
[SyncProducer] Send result: SEND_OK
Message sent to topic sync-topic: Hello RocketMQ
```

---

### 2. 异步发送 (AsyncProducer)

**适用场景**：高并发、对响应时间敏感的场景，如日志收集、用户行为追踪

**代码位置**：`src/main/java/com/demo/rocketmq/scenario/basic/AsyncProducer.java`

**使用示例**：

```java
@Autowired
private AsyncProducer asyncProducer;

// 发送异步消息
asyncProducer.sendAsync("async-topic", "User login event", new AsyncSendCallback() {
    @Override
    public void onSuccess(String msgId, String status) {
        System.out.println("发送成功：MsgId=" + msgId);
    }
    
    @Override
    public void onError(String error) {
        System.err.println("发送失败：" + error);
    }
});
```

**特点**：
- 发送后立即返回，不阻塞主线程
- 通过回调函数处理发送结果
- 适合大批量、高并发场景

---

### 3. 单向发送 (OneWayProducer)

**适用场景**：对可靠性要求不高，追求极致性能，如日志收集、监控数据上报

**代码位置**：`src/main/java/com/demo/rocketmq/scenario/basic/OneWayProducer.java`

**使用示例**：

```java
@Autowired
private OneWayProducer oneWayProducer;

// 发送单向消息（fire and forget）
oneWayProducer.sendOneWay("log-topic", "User clicked button A");
```

**特点**：
- 只负责发送，不关心是否成功
- 无重试机制
- 性能最高，但可靠性最低

---

### 4. 基础消费 (BasicConsumer)

**代码位置**：`src/main/java/com/demo/rocketmq/scenario/basic/BasicConsumer.java`

**配置方式**：

```java
@Component
@RocketMQMessageListener(
    topic = "sync-topic",
    consumerGroup = "basic-consumer-group"
)
public class BasicConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("收到消息：" + message);
    }
}
```

**核心概念**：
- **Topic**: 消息的主题分类
- **Consumer Group**: 消费者组，同一组的消费者负载均衡
- **@RocketMQMessageListener**: 声明一个消费者

---

## 高级场景

### 5. 顺序消息 (OrderedProducer/OrderedConsumer)

**适用场景**：需要保证消息顺序的业务，如订单流程（创建→支付→发货）

**代码位置**：
- Producer: `src/main/java/com/demo/rocketmq/scenario/advanced/OrderedProducer.java`
- Consumer: `src/main/java/com/demo/rocketmq/scenario/advanced/OrderedConsumer.java`

**使用示例**：

```java
@Autowired
private OrderedProducer orderedProducer;

// 按订单维度保证顺序
orderedProducer.sendOrderMessage("order123", "创建订单");
orderedProducer.sendOrderMessage("order123", "支付成功");
orderedProducer.sendOrderMessage("order123", "发货完成");
```

**配置要点**：
- Producer: 使用 `syncSendOrderly` 方法，传入相同的 `orderId` 保证发送到同一队列
- Consumer: `consumeThreadMin=1, consumeThreadMax=1` 保持单线程顺序消费

**消费者配置示例**：
```java
@Component
@RocketMQMessageListener(
    topic = "ordered-topic",
    consumerGroup = "ordered-consumer-group",
    consumeThreadMax = 1  // 关键：单线程保证顺序
)
public class OrderedConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("收到顺序消息：" + message);
    }
}
```

---

### 6. 延迟消息 (DelayProducer/DelayConsumer)

**适用场景**：
- 订单超时未支付，自动取消
- 售后处理提醒
- 缓存预热
- 定时任务调度

**代码位置**：
- Producer: `src/main/java/com/demo/rocketmq/scenario/advanced/DelayProducer.java`
- Consumer: `src/main/java/com/demo/rocketmq/scenario/advanced/DelayConsumer.java`

**使用示例**：

```java
@Autowired
private DelayProducer delayProducer;

// 延迟 5 秒后送达（delayLevel: 1-18）
delayProducer.sendDelayMessage("delay-topic", "超时提醒", 5);

// 延迟级别对应时间：
// 1s, 5s, 10s, 30s, 1m, 2m, 3m, 4m, 5m, 6m, 7m, 8m, 9m, 10m, 20m, 30m, 1h, 2h
```

**实际应用示例**：
```java
// 订单 30 分钟后超时未支付，自动取消
delayProducer.sendDelayMessage("order-timeout", "{\"orderId\":\"123\"}", 5); // 5 级 = 1 分钟
```

---

### 7. 批量消息 (BatchProducer/BatchConsumer)

**适用场景**：一次性发送多条消息，减少网络开销，提高吞吐量

**代码位置**：
- Producer: `src/main/java/com/demo/rocketmq/scenario/advanced/BatchProducer.java`
- Consumer: `src/main/java/com/demo/rocketmq/scenario/advanced/BatchConsumer.java`

**使用示例**：

```java
@Autowired
private BatchProducer batchProducer;

List<String> messages = Arrays.asList(
    "Batch message 1",
    "Batch message 2",
    "Batch message 3"
);

batchProducer.sendBatch("batch-topic", messages);
```

**注意事项**：
- 批量消息总大小不能超过 4MB
- 不适合对实时性要求高的场景

---

### 8. 消息过滤 (FilterProducer/FilterConsumer)

**适用场景**：根据条件过滤消息，如只消费某个用户、某个地区的消息

**代码位置**：
- Producer: `src/main/java/com/demo/rocketmq/scenario/advanced/FilterProducer.java`
- Consumer: `src/main/java/com/demo/rocketmq/scenario/advanced/FilterConsumer.java`

**使用示例**：

```java
// 生产者：设置过滤属性
filterProducer.sendWithFilter("filter-topic", "订单消息", 
    Map.of("userId", "1001", "city", "Beijing"));

// 消费者：SQL92 过滤表达式
@Component
@RocketMQMessageListener(
    topic = "filter-topic",
    consumerGroup = "filter-consumer-group",
    selectorExpression = "userId=1001 AND city='Beijing'"  // SQL92 表达式
)
public class FilterConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("收到过滤后的消息：" + message);
    }
}
```

**SQL92 支持的语法**：
- 比较运算符：`>`, `<`, `>=`, `<=`, `=`, `<>`
- 逻辑运算符：`AND`, `OR`, `NOT`
- 范围：`BETWEEN`, `IN`

**示例**：`age > 20 AND city IN ('Beijing', 'Shanghai')`

---

### 9. 事务消息 (TransactionProducer/TransactionConsumer)

**适用场景**：本地事务与消息发送的原子性，如转账（扣款 + 发消息）

**代码位置**：
- Producer: `src/main/java/com/demo/rocketmq/scenario/advanced/TransactionProducer.java`
- Listener: `src/main/java/com/demo/rocketmq/scenario/advanced/TransactionListenerImpl.java`
- Consumer: `src/main/java/com/demo/rocketmq/scenario/advanced/TransactionConsumer.java`

**使用示例**：

```java
@Autowired
private TransactionProducer transactionProducer;

// 执行本地事务，同时发送事务消息
boolean success = transactionProducer.sendOrderTransaction(
    "order123", 
    "{\"amount\":100}"
);
```

**事务流程**：
1. 发送半消息（Half Message）
2. 执行本地事务
3. 提交/回滚事务
4. 如果步骤 2 失败，RocketMQ 会主动回查事务状态

**注意事项**：
- 事务消息不支持延迟消息和批量消息
- 需要实现 `TransactionListener` 接口
- 本地事务执行时间不能太长（默认 60 秒超时）

---

## 容错处理

### 10. 消费重试 (RetryConsumer)

**适用场景**：消费失败时自动重试，如临时网络问题、数据库锁等

**代码位置**：`src/main/java/com/demo/rocketmq/scenario/faulttolerance/RetryConsumer.java`

**配置方式**：

```java
@Component
@RocketMQMessageListener(
    topic = "retry-topic",
    consumerGroup = "retry-consumer-group",
    maxReconsumeTimes = 3  // 最大重试次数
)
public class RetryConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        // 抛出异常会触发重试
        if (message.contains("error")) {
            throw new RuntimeException("处理失败");
        }
    }
}
```

**重试间隔**（RocketMQ 默认）：
- 第 1 次重试：10 秒后
- 第 2 次重试：30 秒后
- 第 3 次重试：1 分钟后
- 以此类推，最多 18 个级别

---

### 11. 死信队列 (DLQProducer/DLQConsumer)

**适用场景**：消息重试次数超过上限，进入死信队列人工处理

**代码位置**：
- Producer: `src/main/java/com/demo/rocketmq/scenario/faulttolerance/DLQProducer.java`
- Consumer: `src/main/java/com/demo/rocketmq/scenario/faulttolerance/DLQConsumer.java`

**配置方式**：

```java
@Component
@RocketMQMessageListener(
    topic = "dlq-topic",
    consumerGroup = "dlq-consumer-group",
    maxReconsumeTimes = 3  // 超过 3 次进入死信队列
)
public class DLQConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        // 始终失败，触发死信
        throw new RuntimeException("Always fail");
    }
}
```

**死信队列规则**：
- 消息重试次数超过 `maxReconsumeTimes` 后进入死信队列
- 死信队列名称：`%DLQ%{consumerGroup}`
- 死信消息会被保留 3 天（默认）

**处理死信消息**：
1. 查看死信队列中的消息
2. 分析失败原因
3. 修复问题后重新消费

---

## 业务实战

### 12. 订单场景 (OrderProducer/OrderConsumer)

**业务场景**：订单创建、支付、取消等事件通知

**代码位置**：
- Producer: `src/main/java/com/demo/rocketmq/scenario/business/OrderProducer.java`
- Consumer: `src/main/java/com/demo/rocketmq/scenario/business/OrderConsumer.java`

**使用示例**：

```java
@Autowired
private OrderProducer orderProducer;

// 订单创建成功后
orderProducer.sendOrderCreated("ORDER123", "USER001", new BigDecimal("99.00"));

// 订单支付成功后
orderProducer.sendOrderPaid("ORDER123", System.currentTimeMillis());

// 订单取消时
orderProducer.sendOrderCancelled("ORDER123", "用户主动取消");
```

**消费者处理逻辑**（修改 TODO 部分即可）：

```java
private void handleOrderCreated(OrderMessage message) {
    // TODO: 编写你的业务逻辑
    // 例如：通知仓库备货、发送确认邮件等
}

private void handleOrderPaid(OrderMessage message) {
    // TODO: 编写你的业务逻辑
    // 例如：通知仓库发货、发送发货通知等
}

private void handleOrderCancelled(OrderMessage message) {
    // TODO: 编写你的业务逻辑
    // 例如：恢复库存、退款处理等
}
```

---

### 13. 库存场景 (InventoryProducer/InventoryConsumer)

**业务场景**：库存扣减、恢复等操作

**代码位置**：
- Producer: `src/main/java/com/demo/rocketmq/scenario/business/InventoryProducer.java`
- Consumer: `src/main/java/com/demo/rocketmq/scenario/business/InventoryConsumer.java`

**使用示例**：

```java
@Autowired
private InventoryProducer inventoryProducer;

// 扣减库存
inventoryProducer.sendDeduct("ORDER123", "SKU001", 1);

// 恢复库存（订单取消）
inventoryProducer.sendRestore("ORDER123", "SKU001", 1);
```

---

### 14. 支付场景 (PaymentProducer/PaymentConsumer)

**业务场景**：支付成功/失败回调通知

**代码位置**：
- Producer: `src/main/java/com/demo/rocketmq/scenario/business/PaymentProducer.java`
- Consumer: `src/main/java/com/demo/rocketmq/scenario/business/PaymentConsumer.java`

**使用示例**：

```java
@Autowired
private PaymentProducer paymentProducer;

// 支付成功
paymentProducer.sendPaymentSuccess("ORDER123", "PAY999", 
    new BigDecimal("99.00"), "ALIPAY");

// 支付失败
paymentProducer.sendPaymentFailed("ORDER123", "PAY999", 
    new BigDecimal("99.00"), "余额不足");
```

---

## 配置说明

### application.yml 配置

```yaml
server:
  port: 18090

rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: demo-producer-group
    send-message-timeout: 3000
    max-message-size: 4194304
    retry-times-when-send-async-failed: 2
  consumer:
    group: demo-consumer-group
    max-re-consume-times: 3

# Topic 配置
demo:
  topics:
    basic-sync: sync-topic
    basic-async: async-topic
    basic-oneway: oneway-topic
    ordered: ordered-topic
    delay: delay-topic
    batch: batch-topic
    filter: filter-topic
    transaction: transaction-topic
    order: order-topic
    inventory: inventory-topic
    payment: payment-topic
```

### 核心参数说明

**Producer 配置**：
- `group`: 生产者组名
- `send-message-timeout`: 发送超时（毫秒）
- `max-message-size`: 最大消息大小（字节）
- `retry-times-when-send-failed`: 发送失败重试次数

**Consumer 配置**：
- `group`: 消费者组名
- `max-re-consume-times`: 最大重试次数
- `consumeThreadMax`: 最大消费线程数
- `selectorExpression`: 消息过滤表达式

---

## 测试 API

项目提供了 REST API 用于测试各种场景：

```bash
# 1. 发送同步消息
curl -X POST "http://localhost:18090/api/messages/send?topic=sync-topic&message=Hello"

# 2. 发送带 Tag 的消息
curl -X POST "http://localhost:18090/api/messages/send-with-tag?topic=order-topic&tag=ORDER_CREATED&message={\"orderId\":\"123\"}"

# 3. 批量发送消息
curl -X POST "http://localhost:18090/api/messages/send-batch"

# 4. 发送延迟消息
curl -X POST "http://localhost:18090/api/messages/send-delay?delayLevel=3&message=Timeout"

# 5. 发送顺序消息
curl -X POST "http://localhost:18090/api/messages/send-ordered?orderId=order123"

# 6. 发送带过滤条件的消息
curl -X POST "http://localhost:18090/api/messages/send-filter?userId=1001&city=Beijing"

# 7. 发送事务消息
curl -X POST "http://localhost:18090/api/messages/send-transaction?orderId=order123"
```

---

## 常见问题

### 1. 消息堆积怎么办？

**原因**：消费者处理速度跟不上生产者

**解决方案**：
1. 增加 Consumer 实例数量（同一消费者组）
2. 增加消费线程数：`consumeThreadMax=20`
3. 优化消费逻辑，提高处理速度
4. 检查是否有慢 SQL 或外部依赖瓶颈

### 2. 消息重复消费怎么办？

**原因**：网络问题、Broker 重启等导致消息重复投递

**解决方案**：
1. **业务层面实现幂等性**（最重要！）
   - 基于 MessageId 去重
   - 基于业务 ID（如 orderId）去重
2. 使用数据库唯一键约束
3. 使用 Redis 等分布式锁

**示例**：
```java
private final Set<String> processedMessageIds = ConcurrentHashMap.newKeySet();

public void onMessage(String message) {
    String messageId = extractMessageId(message);
    if (!processedMessageIds.add(messageId)) {
        return; // 已处理过，跳过
    }
    // 处理业务逻辑
}
```

### 3. 如何保证消息顺序？

**方案**：
1. 使用顺序消息类型（`syncSendOrderly`）
2. 同一业务 ID 发送到同一队列（如 orderId）
3. 消费端单线程顺序消费（`consumeThreadMax=1`）

**注意**：只能保证局部有序，不能保证全局有序

### 4. 事务消息超时？

**原因**：本地事务执行时间超过 RocketMQ 等待时间（默认 60 秒）

**解决方案**：
1. 优化本地事务执行逻辑
2. 将耗时操作异步化
3. 调整事务超时时间：`transactionTimeout=120000`

### 5. 如何选择合适的 Topic 和 Tag？

**Topic 选择原则**：
- 按业务领域划分：order-topic, payment-topic, inventory-topic
- 不同业务场景使用不同 Topic

**Tag 选择原则**：
- 同一 Topic 下的子分类：ORDER_CREATED, ORDER_PAID, ORDER_CANCELLED
- 用于消息过滤

### 6. 消费者如何负载均衡？

**原理**：
- 同一消费者组的多个消费者会自动负载均衡
- RocketMQ 会将队列均匀分配给各个消费者

**示例**：
```java
// 两个消费者实例，4 个队列
// 每个消费者处理 2 个队列的消息
@Component
@RocketMQMessageListener(
    topic = "order-topic",
    consumerGroup = "order-consumer-group"  // 相同的组名
)
```

---

## 最佳实践

### 1. 消息格式规范

**推荐使用 JSON 格式**：
```json
{
  "messageId": "uuid",
  "eventType": "ORDER_CREATED",
  "timestamp": 1234567890,
  "data": {
    "orderId": "123",
    "userId": "456"
  }
}
```

### 2. 日志记录

**必须记录的关键信息**：
- MessageId（用于追踪和去重）
- 业务 ID（如 orderId）
- 处理时间和结果

```java
log.info("收到订单消息：messageId={}, orderId={}, eventType={}", 
    message.getMessageId(), message.getOrderId(), message.getEventType());
```

### 3. 错误处理

**建议做法**：
1. 捕获异常并记录日志
2. 根据异常类型决定是否重试
3. 对于无法恢复的错误，记录告警

### 4. 监控告警

**需要监控的指标**：
- 消息发送成功率
- 消息消费成功率
- 消息堆积数量
- 死信队列消息数

---

## 更多资料

- [RocketMQ 官方文档](https://rocketmq.apache.org/docs/)
- [RocketMQ Spring Boot Starter](https://github.com/apache/rocketmq-spring)
- [RocketMQ GitHub 仓库](https://github.com/apache/rocketmq)