事件驱动架构：从理念到落地
大约 14 分钟
前言
如果你写过 Spring Boot 项目，你大概率用过 @EventListener 或消息队列。但"用消息队列"和"事件驱动架构"是两回事。事件驱动架构(Event-Driven Architecture, EDA)是一种架构风格，不只是技术选型。
本文带你从"为什么需要事件驱动"开始，理解三种事件模式、Command vs Event 的本质区别，再到实战落地。
第一部分：什么是事件驱动架构
1.1 一个简单的对比
传统请求-响应模式（你每天都在写的）：
// Controller → Service → 同步返回
@PostMapping("/orders")
public Result<OrderResponse> createOrder(@RequestBody OrderRequest request) {
// 1. 校验库存
// 2. 计算价格
// 3. 扣减库存
// 4. 保存订单
// 5. 发送通知
// 6. 记录日志
// 7. 返回结果
return Result.success(response);
}
// 所有事情串行完成，一步慢了全卡
事件驱动模式：
@PostMapping("/orders")
public Result<OrderResponse> createOrder(@RequestBody OrderRequest request) {
// 1. 保存订单（核心操作）
Order order = orderService.create(request);
// 2. 发布事件（其他操作异步化）
eventPublisher.publish(new OrderCreatedEvent(order));
// 3. 立即返回
return Result.success(OrderResponse.from(order));
}
// 事件消费者各自处理：
// - InventoryListener: 扣库存
// - NotificationListener: 发短信
// - LogListener: 记录审计日志
// - AnalyticsListener: 更新实时报表
// 异步并行处理，不阻塞主流程
1.2 EDA 的核心思想
事件驱动架构的核心非常简单：一个系统组件发布事件，其他感兴趣的组件接收并响应事件。
┌──────────┐ 事件 ┌──────────────┐ 事件 ┌──────────┐
│ 订单服务 │ ────────→ │ 消息代理 │ ────────→ │ 库存服务 │
│ │ │ (Event Bus) │ │ │
│ 发布者 │ │ │ │ 消费者 │
└──────────┘ └──────┬───────┘ └──────────┘
│
┌─────────┼─────────┐
▼ ▼ ▼
┌────────┐┌────────┐┌────────┐
│通知服务││日志服务││分析服务│
│消费者 ││消费者 ││消费者 │
└────────┘└────────┘└────────┘
三个关键词：解耦、异步、最终一致性。
第二部分：三种事件模式
2.1 模式一：事件通知（Event Notification）
最简单的模式。事件只携带发生了什么的标识信息，不携带状态数据。消费者收到事件后，如果需要数据，自己回调查询。
事件内容：
{
"eventType": "OrderCreated",
"orderId": "ORD-2026-001",
"timestamp": "2026-06-11T10:30:00Z"
}
（只有订单 ID，没有订单详细信息）
// 发布者
public class OrderService {
public void createOrder(CreateOrderCommand cmd) {
Order order = orderRepository.save(cmd.toOrder());
// 只发通知，不携带数据
eventBus.publish(new OrderCreatedEvent(order.getId()));
}
}
// 消费者
@Component
public class InventoryListener {
@Autowired private OrderClient orderClient; // 回调查询
@EventListener
public void onOrderCreated(OrderCreatedEvent event) {
// 需要数据？自己回调
OrderDTO order = orderClient.getOrder(event.getOrderId());
inventoryService.deduct(order.getItems());
}
}
优点：事件体小，传输效率高
缺点：消费者需要回调查询，增加耦合和延迟；源服务压力增大（大量回调查询）
2.2 模式二：事件携带状态转移（Event-Carried State Transfer）
事件携带足够的数据，消费者不需要回调源服务。这是最推荐的生产实践。
事件内容：
{
"eventType": "OrderCreated",
"orderId": "ORD-2026-001",
"customerId": "CUS-1001",
"items": [
{"productId": "SKU-001", "quantity": 2, "price": 99.00},
{"productId": "SKU-002", "quantity": 1, "price": 199.00}
],
"totalAmount": 397.00,
"shippingAddress": {
"province": "北京",
"city": "北京",
"detail": "朝阳区xxx"
},
"timestamp": "2026-06-11T10:30:00Z"
}
（包含消费者需要的所有信息）
// 发布者：携带完整状态
public class OrderService {
public void createOrder(CreateOrderCommand cmd) {
Order order = orderRepository.save(cmd.toOrder());
// 发布携带完整状态的事件
eventBus.publish(OrderCreatedEvent.from(order));
// OrderCreatedEvent.from(order) 将订单所有需要的字段都放入事件
}
}
// 消费者：无需回调查询，完全自包含
@Component
public class InventoryListener {
@EventListener
public void onOrderCreated(OrderCreatedEvent event) {
// 直接从事件中获取所有需要的数据
for (OrderItem item : event.getItems()) {
inventoryService.deduct(item.getProductId(), item.getQuantity());
}
}
}
最佳实践：携带"消费者需要但未来不会频繁改变"的数据。注意：不要在事件中包含消费者不需要的敏感数据。
2.3 模式三：事件溯源（Event Sourcing）
最"重"的模式。不存储当前状态，而是存储所有状态变更的事件序列。当前状态 = 所有事件的"回放"结果。
用户余额的变化：
事件序列：
INIT: 余额 = 0
+ DepositEvent(1000) → 余额 = 1000
+ WithdrawEvent(300) → 余额 = 700
+ DepositEvent(500) → 余额 = 1200
+ WithdrawEvent(200) → 余额 = 1000
当前余额 = SUM(所有事件的金额变化) = 1000
/**
* 事件溯源版本的账户服务
*/
public class Account {
private AccountId id;
private List<AccountEvent> events = new ArrayList<>();
// 不是存储"余额 = 1000"，而是记录每个事件
public void deposit(Money amount) {
apply(new MoneyDepositedEvent(this.id, amount, LocalDateTime.now()));
}
public void withdraw(Money amount) {
if (getCurrentBalance().lessThan(amount)) {
throw new InsufficientBalanceException();
}
apply(new MoneyWithdrawnEvent(this.id, amount, LocalDateTime.now()));
}
private void apply(AccountEvent event) {
events.add(event);
}
// 当前余额 = 回放所有事件
public Money getCurrentBalance() {
return events.stream()
.map(e -> switch (e) {
case MoneyDepositedEvent de -> de.getAmount();
case MoneyWithdrawnEvent we -> we.getAmount().negate();
default -> Money.ZERO;
})
.reduce(Money.ZERO, Money::add);
}
// 从事件序列重建账户
public static Account fromEvents(List<AccountEvent> events) {
Account account = new Account();
events.forEach(e -> account.apply(e));
return account;
}
}
事件溯源的优势：
- 完整的审计日志（谁在什么时候做了什么）
- 可以回溯到任意历史状态
- 方便分析业务行为模式
事件溯源的代价：
- 实现复杂度高
- 数据量大（事件数量远大于状态数量）
- 查询当前状态需要回放（可通过快照优化）
金融、担保、审计等强监管场景特别适合事件溯源。
2.4 三种模式的选择指南
| 模式 | 适用场景 | 不适用场景 |
|---|---|---|
| 事件通知 | 消费者需要最新的实时数据 | 高并发、希望完全解耦 |
| 携带状态转移 | 大多数场景（推荐） | 事件数据过大（超过 MB 级） |
| 事件溯源 | 金融/审计/需要完整历史 | 简单 CRUD、高 QPS 查询 |
第三部分：Command vs Event
3.1 本质区别
这是最容易混淆的概念之一。
| 维度 | Command（命令） | Event（事件） |
|---|---|---|
| 意图 | 请求某事发生 | 告知某事已发生 |
| 方向 | 指向特定接收者 | 广播给关心的人 |
| 命名 | 祈使句：CreateOrder | 过去式：OrderCreated |
| 可否拒绝 | 可以 | 不可拒绝（已经发生了！） |
| 一个 vs 多个 | 一个命令一个接收者 | 一个事件多个消费者 |
| 时间 | 未来时 | 过去时 |
// Command：请求做某事（可能被拒绝）
public class CreateOrderCommand {
private CustomerId customerId;
private List<OrderItem> items;
private Money totalAmount;
}
// "请帮我创建一个订单"
// Event：告知某事已发生（不可否认）
public class OrderCreatedEvent {
private OrderId orderId;
private CustomerId customerId;
private List<OrderItem> items;
private Money totalAmount;
private LocalDateTime createdAt;
}
// "订单已创建，编号 ORD-2026-001"
3.2 实践：Command 转 Event
@Service
public class OrderApplicationService {
// 接收 Command
public OrderCreatedEvent createOrder(CreateOrderCommand command) {
// 1. 处理命令（可能失败）
Order order = Order.create(command);
// 2. 持久化
orderRepository.save(order);
// 3. 返回事件（告知世界发生了什么）
return new OrderCreatedEvent(
order.getId(),
order.getCustomerId(),
order.getItems(),
order.getTotalAmount(),
Instant.now()
);
// 如果步骤 1 失败了，步骤 2-3 都不会执行，事件不会产生
}
}
一句话区分：
- 你命令朋友："帮我买杯咖啡" → Command（可能被拒绝："我没空"）
- 你告诉朋友："我刚买了杯咖啡" → Event（事实已发生，无法拒绝）
第四部分：消息代理的选择
4.1 三大消息队列对比
| 特性 | Kafka | RocketMQ | RabbitMQ |
|---|---|---|---|
| 设计目标 | 高吞吐日志流 | 金融级可靠消息 | 灵活路由 |
| 吞吐量 | 极高（百万/秒） | 高（十万/秒） | 中（万/秒） |
| 延迟 | 毫秒级 | 毫秒级 | 微秒级 |
| 持久化 | 磁盘顺序写 | 同步/异步刷盘 | 内存+磁盘 |
| 事务消息 | 支持（0.11+） | 原生支持 | 支持（但复杂） |
| 顺序消息 | 分区内有序 | 支持 | 支持 |
| 延迟消息 | 不支持原生 | 支持（18级） | 插件支持 |
| 消息回溯 | 支持（按时间） | 支持（按时间） | 不支持 |
| 协议 | 自有协议 | 自有协议 | AMQP |
| 运维复杂度 | 中 | 中 | 低 |
| Java 生态 | Spring Kafka | Spring Cloud Stream | Spring AMQP |
4.2 Kafka：事件流的最佳选择
// Kafka 生产者和消费者
@Service
public class KafkaOrderEventPublisher {
@Autowired private KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate;
public void publish(OrderCreatedEvent event) {
// 使用 orderId 作为 key，保证同一个订单的事件有序
kafkaTemplate.send("order-events",
event.getOrderId().toString(), // key
event // value
);
}
}
@Component
public class KafkaInventoryConsumer {
@KafkaListener(topics = "order-events", groupId = "inventory-group")
public void onOrderCreated(
@Payload OrderCreatedEvent event,
@Header(KafkaHeaders.RECEIVED_KEY) String key) {
log.info("收到订单事件: orderId={}, key={}", event.getOrderId(), key);
for (OrderItem item : event.getItems()) {
inventoryService.deduct(item.getProductId(), item.getQuantity());
}
}
}
# Kafka 生产者配置
spring:
kafka:
producer:
bootstrap-servers: kafka-broker:9092
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
# 可靠性配置
acks: all # 所有 ISR 确认
retries: 3
enable-idempotence: true # 幂等生产者
compression-type: snappy # 压缩
4.3 RocketMQ：金融场景的首选
// RocketMQ 事务消息 - 实现分布式事务
@Service
public class RocketMQOrderService {
@Autowired private RocketMQTemplate rocketMQTemplate;
@Transactional
public void createOrder(CreateOrderCommand command) {
Order order = orderRepository.save(command.toOrder());
// 事务消息：先发半消息，本地事务提交后消息才投递
rocketMQTemplate.sendMessageInTransaction(
"order-topic",
MessageBuilder.withPayload(OrderCreatedEvent.from(order)).build(),
order.getId() // 传递给事务检查器的参数
);
}
}
// 事务检查器：处理半消息的状态回查
@RocketMQTransactionListener
public class OrderTransactionListener
implements RocketMQLocalTransactionListener {
@Override
public RocketMQLocalTransactionState executeLocalTransaction(
Message msg, Object arg) {
try {
Long orderId = (Long) arg;
// 检查订单是否已成功创建
Order order = orderRepository.findById(new OrderId(orderId))
.orElse(null);
if (order != null) {
return RocketMQLocalTransactionState.COMMIT;
}
return RocketMQLocalTransactionState.ROLLBACK;
} catch (Exception e) {
return RocketMQLocalTransactionState.UNKNOWN;
}
}
@Override
public RocketMQLocalTransactionState checkLocalTransaction(
Message msg) {
// 回查逻辑：消息服务端定期回调此方法确认事务状态
Long orderId = extractOrderId(msg);
return orderRepository.existsById(new OrderId(orderId))
? RocketMQLocalTransactionState.COMMIT
: RocketMQLocalTransactionState.ROLLBACK;
}
}
// 消费者
@Service
@RocketMQMessageListener(
topic = "order-topic",
consumerGroup = "inventory-group",
consumeMode = ConsumeMode.ORDERLY // 顺序消费
)
public class InventoryConsumer
implements RocketMQListener<OrderCreatedEvent> {
@Override
public void onMessage(OrderCreatedEvent event) {
// RocketMQ 保证重试，需要实现幂等
if (inventoryDeductLogRepository.exists(event.getOrderId())) {
return; // 幂等：已处理过
}
inventoryService.deduct(event.getItems());
// 记录处理日志（用于幂等判断）
inventoryDeductLogRepository.save(
new DeductLog(event.getOrderId()));
}
}
第五部分：实战——订单系统的事件驱动改造
5.1 改造前（同步耦合）
@Service
public class OrderService {
@Autowired private InventoryService inventoryService;
@Autowired private PaymentService paymentService;
@Autowired private NotificationService notificationService;
@Autowired private CouponService couponService;
@Autowired private LogService logService;
@Transactional
public Order createOrder(OrderRequest request) {
// 同步调用所有服务 → 耦合严重
inventoryService.deduct(request.getItems()); // 2s
couponService.use(request.getCouponId()); // 0.5s
Order order = saveOrder(request); // 0.1s
paymentService.createPayment(order); // 1s
notificationService.sendSms(order); // 0.5s
logService.recordAuditLog(order); // 0.2s
return order;
// 总耗时: 4.3 秒！用户体验极差
}
}
5.2 改造后（事件驱动）
@Service
public class OrderService {
@Autowired private OrderRepository orderRepository;
@Autowired private DomainEventPublisher eventPublisher;
@Transactional
public Order createOrder(CreateOrderCommand command) {
// 1. 创建订单（核心操作）
Order order = Order.create(command);
orderRepository.save(order);
// 2. 发布领域事件
eventPublisher.publish(OrderCreatedEvent.from(order));
// 3. 立即返回
return order;
// 总耗时: ~0.1 秒
}
}
// === 各消费者独立处理 ===
// 库存服务 - 扣减库存
@Component
public class OrderInventoryHandler {
@EventListener
@Transactional
public void handleOrderCreated(OrderCreatedEvent event) {
for (OrderItem item : event.getItems()) {
inventoryRepository.deduct(
item.getProductId(), item.getQuantity());
}
}
}
// 支付服务 - 创建支付单
@Component
public class OrderPaymentHandler {
@EventListener
public void handleOrderCreated(OrderCreatedEvent event) {
paymentService.createPayment(
event.getOrderId(), event.getTotalAmount());
}
}
// 通知服务 - 发送短信
@Component
public class OrderNotificationHandler {
@EventListener
@Async // 异步处理
public void handleOrderCreated(OrderCreatedEvent event) {
notificationService.sendOrderConfirmation(
event.getCustomerId(), event.getOrderId());
}
}
// 审计日志服务
@Component
public class OrderAuditLogHandler {
@EventListener
@Async
public void handleOrderCreated(OrderCreatedEvent event) {
auditLogRepository.save(new AuditLog(
"ORDER_CREATED",
event.getOrderId(),
event.getCustomerId(),
Instant.now()
));
}
}
5.3 事件版本管理
事件结构会随着业务发展而变化，需要做好版本管理：
/**
* 带版本号的订单创建事件
* 向后兼容：新字段增加默认值，旧字段不删除
*/
public class OrderCreatedEvent implements DomainEvent {
// 事件元数据
private String eventId;
private String eventType = "OrderCreated";
private int version; // 事件版本号
private Instant occurredAt;
// V1 字段
private OrderId orderId;
private CustomerId customerId;
private Money totalAmount;
// V2 新增字段（2025-03）
private String channel; // 下单渠道 (APP/WEB/MINI_PROGRAM)
private String promotionCode; // 推广码
// V3 新增字段（2025-06）
private boolean isPresale; // 是否预售
private Instant expectedShipDate; // 预计发货日期
// 构造方法
private OrderCreatedEvent() {} // 序列化用
public static OrderCreatedEvent from(Order order) {
OrderCreatedEvent event = new OrderCreatedEvent();
event.eventId = UUID.randomUUID().toString();
event.version = 3; // 当前版本
event.occurredAt = Instant.now();
event.orderId = order.getId();
event.customerId = order.getCustomerId();
event.totalAmount = order.getTotalAmount();
event.channel = order.getChannel();
event.promotionCode = order.getPromotionCode();
event.isPresale = order.isPresale();
event.expectedShipDate = order.getExpectedShipDate();
return event;
}
}
// 消费者做版本兼容处理
@Component
public class InventoryHandler {
@EventListener
public void handleOrderCreated(OrderCreatedEvent event) {
// 按版本处理不同逻辑
switch (event.getVersion()) {
case 1:
handleV1(event);
break;
case 2:
handleV2(event);
break;
case 3:
handleV3(event);
break;
default:
log.warn("未知事件版本: {}", event.getVersion());
// 尝试降级处理
handleV1(event);
}
}
private void handleV1(OrderCreatedEvent event) {
// 处理 V1 逻辑（没有渠道、预售等信息）
}
}
事件版本管理最佳实践：
- 只增不减：新字段只添加，旧字段标记
@Deprecated但不删除 - 消费者容错：未知字段忽略（不报错）
- 版本号递增：每次结构变更，版本号 +1
- 降级兼容：新消费者应能处理旧版本事件
第六部分：最终一致性处理
6.1 拥抱最终一致性
事件驱动架构的关键转变：从强一致性到最终一致性。
强一致性（同步）：
订单创建 → 等待库存扣减完成 → 返回给用户
用户等待时间 = 所有步骤耗时之和
最终一致性（异步）：
订单创建 → 立即返回给用户
库存扣减（异步）→ 可能晚几秒完成
用户等待时间 = 订单创建耗时
代价：用户可能看到"订单已创建，库存处理中..."的状态
收益：系统吞吐量提升 10 倍以上
6.2 处理失败重试
@Component
public class RobustInventoryHandler {
private static final int MAX_RETRIES = 3;
@EventListener
public void handleOrderCreated(OrderCreatedEvent event) {
for (int attempt = 1; attempt <= MAX_RETRIES; attempt++) {
try {
inventoryService.deduct(event.getItems());
return; // 成功
} catch (InsufficientInventoryException e) {
// 库存不足 - 不重试，直接处理
handleInsufficientInventory(event);
return;
} catch (TemporaryException e) {
// 临时错误 - 重试
log.warn("库存扣减临时失败，第 {} 次重试", attempt);
sleep(1000L * attempt); // 指数退避
}
}
// 重试耗尽 - 发送到死信队列
deadLetterQueue.send(event);
alertService.alert("库存扣减失败", event);
}
}
6.3 幂等性保障
事件可能被重复投递（at-least-once 语义），消费者必须实现幂等：
@Component
public class IdempotentInventoryHandler {
@Autowired private EventProcessLogRepository logRepository;
@EventListener
@Transactional
public void handleOrderCreated(OrderCreatedEvent event) {
// 幂等检查：用 eventId 作为唯一键
String eventId = event.getEventId();
if (logRepository.existsByEventId(eventId)) {
log.info("事件已处理，跳过: eventId={}", eventId);
return; // 幂等：已处理过
}
// 处理业务
inventoryService.deduct(event.getItems());
// 记录处理日志（插入成功 = 唯一索引保证不会重复处理）
logRepository.save(new EventProcessLog(eventId, "PROCESSED", Instant.now()));
}
}
// 数据库表设计
CREATE TABLE t_event_process_log (
event_id VARCHAR(64) PRIMARY KEY, -- 唯一约束 = 幂等保证
status VARCHAR(20),
processed_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
第七部分：EDA 的陷阱
7.1 事件风暴
症状：一个事件触发另一个事件，形成事件链，最终无法追踪数据流向：
OrderCreated → InventoryDeducted → StockLowAlert
→ ReplenishmentRequested
→ ReplenishmentApproved
→ PurchaseOrderCreated
→ ...
解决方案：
- 事件溯源记录所有事件（虽然不能阻止风暴，但能追溯）
- 明确的事件所有权：谁能发布什么事件
- 事件流图：用文档/图表显式化事件依赖关系
- 断路保护：检测事件循环并熔断
7.2 调试困难
症状：一个请求分散在多个服务的多个事件中，像"大海捞针"：
排查"为什么用户没收到库存扣减通知"：
1. 查订单服务的 order-created 事件是否发布
2. 查库存服务是否消费了该事件
3. 查 stock-low-alert 事件是否发布
4. 查通知服务是否消费了 stock-low-alert 事件
5. 查短信网关是否成功发送
解决方案：
- 全链路追踪：Jaeger/Zipkin 关联所有事件
- 事件 ID 传递：在事件中携带 traceId
- 事件追踪平台：可视化事件流
- 每个事件携带 correlationId：串联同一个业务流程的所有事件
public class OrderCreatedEvent {
private String eventId;
private String traceId; // 链路追踪 ID（从上游提取）
private String correlationId; // 业务流程 ID（关联同一个流程）
// ...
}
// 发布事件时传递 traceId
public void publish(Order order, String traceId) {
OrderCreatedEvent event = OrderCreatedEvent.from(order);
event.setTraceId(traceId);
event.setCorrelationId(order.getOrderNo().toString());
eventBus.publish(event);
}
7.3 数据不一致的排查
事件驱动下的数据不一致比同步调用更难排查，因为问题可能在你看到的时候已经"过去了"。
最佳实践：
- 监控事件延迟：如果事件处理延迟超过阈值，告警
- 对账机制：定期对比源系统和目标系统的数据
- 补偿任务：定期扫描"未完成"的业务流程并补偿
/**
* 每天凌晨 2 点执行的对账任务
*/
@Component
public class DailyReconciliationJob {
@Scheduled(cron = "0 0 2 * * ?")
public void reconcile() {
// 查询"订单已创建但库存未扣减"的异常数据
List<Order> unreconciledOrders = orderRepository
.findOrdersWithoutInventoryDeduction(
LocalDate.now().minusDays(1));
for (Order order : unreconciledOrders) {
log.warn("发现未对账订单: {}", order.getId());
// 补偿：重新发布库存扣减事件
eventPublisher.publish(
new InventoryDeductCompensationEvent(order));
}
}
}
第八部分：架构演进路线
从单体到事件驱动三步走
Phase 1: 内部事件
在单体应用内使用 Spring Events (@EventListener)
订单模块事件 → 通知模块、日志模块
目标是理解事件思维，不引入外部依赖
Phase 2: 外部事件总线
引入 Kafka/RocketMQ
核心跨模块通信使用消息队列
非核心仍保留同步调用
Phase 3: 完整 EDA
所有模块间通信通过事件
全链路追踪
事件版本管理
自动化对账和补偿
对于大多数项目，Phase 2 就是最优状态。Phase 3 适合事件驱动的产品型公司（如电商平台、金融交易系统）。
总结
| 概念 | 要点 |
|---|---|
| 事件通知 | 轻量，只发 ID，消费者回调查询 |
| 携带状态转移 | 推荐：事件自包含，消费者无需回调 |
| 事件溯源 | 不存状态存事件历史，适合审计 |
| Command vs Event | Command=请求，Event=事实 |
| 消息选型 | 大数据→Kafka，金融→RocketMQ，路由灵活→RabbitMQ |
| 幂等 | 靠 eventId + 唯一索引保证 |
| 版本管理 | 只增不减，消费者容错 |
| 对账 | 定期检查数据一致性，补偿异常 |
最终一致性不是 bug，是 feature。 拥抱它，用对账和监控来管理它，而不是试图消灭它。
参考资料
- Martin Fowler, "What do you mean by 'Event-Driven'?"
- Greg Young, "CQRS and Event Sourcing"
- Confluent, "Event-Carried State Transfer"
- Chris Richardson, "Pattern: Event-driven architecture"
- RocketMQ 官方文档 - 事务消息