Spring事务消息原理与可靠投递实战指南

1. 为什么“事务消息”不是加个 @Transactional 就完事了？

在某次电商订单系统重构中，我负责把老的同步扣库存逻辑改成异步解耦方案。最开始的想法特别朴素：用户下单成功后，在@Transactional方法里直接发 RabbitMQ 消息，认为“只要数据库提交了，消息肯定发出去了”。结果上线第三天，监控告警疯狂刷屏——订单状态是“已支付”，但库存服务压根没收到任何扣减指令。排查日志发现，消息确实发了，可 RabbitMQ 的 confirm 机制返回失败；更糟的是，数据库事务已经 commit，根本没法回滚。

这暴露了一个被大量开发者忽略的本质矛盾：Spring 的本地事务（JDBC Connection）和 RabbitMQ 的 AMQP 连接，天生就属于两个独立的资源管理器。你用@Transactional控制的是数据库连接的提交/回滚边界，而 RabbitMQ 的消息发送、确认、重试，完全由它自己的 Channel 和 Connection 管理。两者之间没有 XA 协议协调，也没有 JTA 容器支持。简单说，@Transactional对 RabbitMQ 的网络传输、Broker 接收、队列落盘这些环节，一丁点控制力都没有。

这时候有人会说：“那我手动控制呗——先发消息，等 RabbitMQ 返回 success 再 commit 数据库。” 这叫“正向流程强依赖”，问题更大：如果消息发出去了，RabbitMQ 返回了 success，但数据库写入中途因网络抖动失败，你又得去 RabbitMQ 手动撤回那条消息——可 AMQP 协议根本不支持“撤回已确认消息”。反过来，“先 commit 数据库，再发消息”，就是我开头踩的坑：数据库成功了，消息丢了，数据不一致。

真正的解法，必须绕开“让两个资源强一致”这个死胡同，转而接受“最终一致性”的现实。核心思路是：把消息本身当成一种需要持久化的业务状态，和订单记录一起存进同一个数据库，再由一个独立的、可靠的投递机制来保证它最终到达 RabbitMQ。TransactionalEventListener就是 Spring 提供的、最轻量级的落地工具——它不解决跨资源事务，而是巧妙地利用本地事务的“事件钩子”，在数据库 commit 的最后一刻，把消息写进一张“消息表”，然后由另一个线程（或定时任务）异步读取这张表，重试发送，直到 RabbitMQ 确认成功。

关键词里没写，但你必须心里有数：这里真正起作用的，不是@Transactional，而是@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)这个 phase。它像一个精准的手术刀，只在事务 commit 成功后的那个毫秒级窗口内触发，既确保了数据库变更已落盘，又避开了事务 rollback 的干扰。很多教程只教你怎么写注解，却从不解释为什么必须是AFTER_COMMIT，而不是BEFORE_COMMIT或AFTER_COMPLETION——前者会导致消息写入时事务还没提交，数据不可见；后者则可能在事务异常 rollback 后仍被触发，造成脏数据。

提示：TransactionPhase.AFTER_COMMIT的底层原理，是 Spring 在AbstractPlatformTransactionManager的processCommit()方法末尾，遍历并触发所有注册的TransactionSynchronization实例。它不是一个“监听器回调”，而是一个嵌入事务生命周期的同步钩子，执行时机比普通事件监听严格得多。

2. TransactionalEventListener 的真实工作边界与三个致命误用场景

TransactionalEventListener绝不是万能胶水。它的设计初衷非常明确：在单个本地事务的生命周期内，提供一个可靠的、与事务状态强绑定的事件通知机制。一旦超出这个边界，它就会失效，甚至引发更隐蔽的问题。我在某金融对账系统里就栽过跟头，下面这三个场景，是实测下来最容易掉进去的坑。

2.1 场景一：跨事务方法调用，事件根本不会触发

假设你写了这样一个 Service：

@Service public class OrderService { @Transactional public void createOrder(Order order) { orderMapper.insert(order); // 写入订单主表 sendOrderCreatedEvent(order); // 调用另一个方法发事件 } private void sendOrderCreatedEvent(Order order) { // 这里加 @TransactionalEventListener 注解？没用！ ApplicationEventPublisher.publishEvent(new OrderCreatedEvent(order)); } }

很多人以为，只要sendOrderCreatedEvent方法里发事件，@TransactionalEventListener就能捕获。错。@TransactionalEventListener只对直接由@Transactional方法发布（publish）的事件生效。上面代码里，createOrder是事务方法，但它自己没发事件；sendOrderCreatedEvent是普通私有方法，没有事务上下文，它发的事件，Spring 根本不会把它和createOrder的事务绑定。结果就是：订单写入成功，事件石沉大海。

正确做法，必须让事件发布发生在事务方法内部，且由该方法直接调用publishEvent：

@Transactional public void createOrder(Order order) { orderMapper.insert(order); // 必须在这里 publish，且 event 类型要和 listener 声明的一致 eventPublisher.publishEvent(new OrderCreatedEvent(order)); }

2.2 场景二：异步方法（@Async）中使用，事件丢失率接近100%

这是最危险的误用。看这段代码：

@Service public class OrderService { @Transactional public void createOrder(Order order) { orderMapper.insert(order); // 错！绝对不要在事务方法里直接调用 @Async 方法发事件 asyncMessageSender.sendToRabbitMQ(order); } } @Service public class AsyncMessageSender { @Async // 新启一个线程池线程 public void sendToRabbitMQ(Order order) { rabbitTemplate.convertAndSend("order.exchange", "order.created", order); } }

表面看很优雅：事务提交后，异步发消息。但问题在于：@Async方法是在新线程里执行的，它完全脱离了原事务的上下文。createOrder方法里的@Transactional对sendToRabbitMQ没有任何约束力。更致命的是，createOrder方法执行完，事务就结束了，但sendToRabbitMQ可能才刚开始执行。如果此时 JVM 崩溃，或者机器断电，订单数据已落库，消息却永远发不出去——典型的“半消息”。

TransactionalEventListener正是为了解决这个问题而生。它把事件发布和事务 commit 绑定在同一上下文，确保“要么都成功，要么都不发生（指事件发布）”。而@Async是彻底放弃这种绑定，走向不可靠的异步。

2.3 场景三：监听器方法抛出异常，导致整个事务回滚

这是TransactionalEventListener最反直觉的行为。默认情况下，它的监听方法如果抛出未捕获异常，会向上冒泡，最终导致正在执行的@Transactional方法回滚。这意味着，你本意是“订单创建成功后，发个消息通知下游”，结果因为消息发送时 RabbitMQ 连接超时，监听器报错，订单创建也跟着失败了——业务语义被完全扭曲。

验证很简单，在监听器里加一行throw new RuntimeException("Simulate MQ fail");，跑一下就知道。修复方式有两种：

• 推荐：在监听器方法上显式声明@TransactionalEventListener(phase = AFTER_COMMIT, fallbackExecution = true)。fallbackExecution = true表示：即使事务已经 commit，这个监听器也要执行；如果它失败了，绝不影响主事务。这是最符合“最终一致性”设计哲学的配置。

• 备选：把监听器方法声明为@Transactional(propagation = Propagation.REQUIRES_NEW)，让它运行在一个全新的、独立的事务里。这样它的失败也不会波及主事务。但要注意，这会带来额外的数据库连接开销，且无法享受主事务的隔离级别。

注意：fallbackExecution = true是关键开关。很多项目文档里一笔带过，但它是区分“可靠事件通知”和“脆弱事务耦合”的分水岭。不加它，TransactionalEventListener就是个披着事件外衣的同步调用陷阱。

3. 从零搭建可靠事务消息链路：MyBatis-Plus + RabbitMQ 实战步骤

现在我们把理论落地。目标很明确：用户下单成功后，确保一条结构化的订单创建消息，100% 可靠地进入 RabbitMQ 的指定队列，且整个过程不破坏订单主流程的性能和稳定性。整个链路由三部分组成：事件定义 → 事务内发布 → 异步可靠投递。下面每一步，我都给出可直接复制粘贴的代码，并解释为什么这么写。

3.1 第一步：定义清晰、可序列化的业务事件

事件不是随便一个 POJO 就行。它必须满足三个硬性条件：可序列化、无副作用、携带完整上下文。我见过太多人用OrderEntity直接当事件，结果因为实体类里有 MyBatis-Plus 的@TableField注解、或懒加载的@JsonIgnore字段，导致 JSON 序列化失败。

正确的做法，是专门为事件建一个 DTO：

// com.example.event.dto.OrderCreatedEventDTO.java public class OrderCreatedEventDTO implements Serializable { private static final long serialVersionUID = 1L; private Long orderId; private String orderNo; private BigDecimal amount; private Long userId; private LocalDateTime createTime; // 构造函数、getter、setter 省略，务必生成全参构造 public OrderCreatedEventDTO(Long orderId, String orderNo, BigDecimal amount, Long userId, LocalDateTime createTime) { this.orderId = orderId; this.orderNo = orderNo; this.amount = amount; this.userId = userId; this.createTime = createTime; } }

为什么强调serialVersionUID？因为事件可能被不同版本的服务消费，如果没有固定 ID，JVM 的默认序列化算法会根据类结构生成一个哈希值，稍有改动（比如加个字段）就导致反序列化失败。serialVersionUID = 1L是最安全的起点。

3.2 第二步：在事务方法中发布事件，并确保监听器精准捕获

这是整个链路的“心脏起搏点”。发布位置必须在@Transactional方法内部，且事件类型必须和监听器声明的完全一致：

@Service public class OrderService { @Autowired private OrderMapper orderMapper; @Autowired private ApplicationEventPublisher eventPublisher; @Transactional(rollbackFor = Exception.class) public Order createOrder(Order orderParam) { // 1. 构建订单实体 Order order = new Order(); order.setOrderNo("ORD" + System.currentTimeMillis()); order.setAmount(orderParam.getAmount()); order.setUserId(orderParam.getUserId()); order.setStatus(OrderStatus.CREATED.getCode()); order.setCreateTime(LocalDateTime.now()); // 2. 写入数据库（MyBatis-Plus） orderMapper.insert(order); // 3. 关键！在此处发布事件，且必须是 OrderCreatedEventDTO 类型 OrderCreatedEventDTO eventDTO = new OrderCreatedEventDTO( order.getId(), order.getOrderNo(), order.getAmount(), order.getUserId(), order.getCreateTime() ); eventPublisher.publishEvent(eventDTO); // 注意：不是 new OrderCreatedEvent(order) return order; } }

监听器的写法，必须严格匹配事件类型，并开启fallbackExecution：

@Component public class OrderEventListeners { @Autowired private RabbitTemplate rabbitTemplate; // 监听 OrderCreatedEventDTO 类型的事件 @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT, fallbackExecution = true) public void handleOrderCreatedEvent(OrderCreatedEventDTO event) { try { // 4. 将事件对象转换为 JSON 字符串，作为消息体 String messageJson = JSON.toJSONString(event); // 5. 发送到 RabbitMQ，使用 topic exchange 和 routing key rabbitTemplate.convertAndSend( "order.exchange", // exchange 名称 "order.created", // routing key messageJson // 消息体 ); log.info("Order created event sent to RabbitMQ: {}", event.getOrderNo()); } catch (Exception e) { // 6. 记录错误，但绝不 throw！因为 fallbackExecution=true 已保证不影响主事务 log.error("Failed to send order created event to RabbitMQ", e); // 这里可以触发告警，或写入失败日志表，用于后续人工干预 } } }

3.3 第三步：引入“消息表”作为最终保险，实现 100% 可靠投递

上面的handleOrderCreatedEvent方法虽然加了fallbackExecution=true，但它仍然是“尽力而为”。如果 RabbitMQ Broker 宕机超过 5 分钟，或者网络分区，这条消息就永久丢失了。生产环境要求的是“至少一次”，不是“最多一次”。所以，我们必须引入第二道防线：将待发送的消息，先持久化到数据库的一张message_outbox表中，再由一个独立的、带重试机制的后台任务来轮询这张表，确保每条消息最终都被成功投递。

首先，创建消息表（MySQL）：

CREATE TABLE `message_outbox` ( `id` bigint NOT NULL AUTO_INCREMENT COMMENT '主键', `event_type` varchar(100) NOT NULL COMMENT '事件类型，如 order.created', `payload` text NOT NULL COMMENT 'JSON 格式的消息体', `status` tinyint NOT NULL DEFAULT '0' COMMENT '0-待发送，1-发送成功，2-发送失败', `retry_count` int NOT NULL DEFAULT '0' COMMENT '重试次数', `max_retry` int NOT NULL DEFAULT '3' COMMENT '最大重试次数', `next_retry_time` datetime DEFAULT NULL COMMENT '下一次重试时间', `created_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', `updated_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', PRIMARY KEY (`id`), KEY `idx_status_next_retry` (`status`,`next_retry_time`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='消息发件箱表';

然后，改造监听器，不再直接发 RabbitMQ，而是插入这条记录：

@Component public class OrderEventListeners { @Autowired private MessageOutboxMapper outboxMapper; // MyBatis-Plus Mapper @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT, fallbackExecution = true) public void handleOrderCreatedEvent(OrderCreatedEventDTO event) { try { MessageOutbox outbox = new MessageOutbox(); outbox.setEventType("order.created"); outbox.setPayload(JSON.toJSONString(event)); outbox.setStatus(MessageOutboxStatus.PENDING.getCode()); outbox.setRetryCount(0); outbox.setMaxRetry(3); outbox.setNextRetryTime(LocalDateTime.now()); // 立即重试 outbox.setCreatedTime(LocalDateTime.now()); outbox.setUpdatedTime(LocalDateTime.now()); outboxMapper.insert(outbox); // 插入消息表，和订单表在同一个事务里！ log.info("Order event saved to outbox table: {}", event.getOrderNo()); } catch (Exception e) { log.error("Failed to save order event to outbox table", e); } } }

最后，编写一个独立的@Scheduled任务，每 30 秒扫描一次待发送的消息：

@Component public class OutboxMessageSender { @Autowired private MessageOutboxMapper outboxMapper; @Autowired private RabbitTemplate rabbitTemplate; @Scheduled(fixedDelay = 30_000) // 每30秒执行一次 public void sendPendingMessages() { // 1. 查询所有 status=0 且 next_retry_time <= now 的消息 LocalDateTime now = LocalDateTime.now(); List<MessageOutbox> pendingList = outboxMapper.selectList( new QueryWrapper<MessageOutbox>() .eq("status", MessageOutboxStatus.PENDING.getCode()) .le("next_retry_time", now) ); for (MessageOutbox outbox : pendingList) { try { // 2. 尝试发送到 RabbitMQ rabbitTemplate.convertAndSend( "order.exchange", "order.created", outbox.getPayload() ); // 3. 发送成功，更新状态为 SUCCESS outbox.setStatus(MessageOutboxStatus.SUCCESS.getCode()); outbox.setUpdatedTime(LocalDateTime.now()); outboxMapper.updateById(outbox); log.info("Outbox message sent successfully: {}", outbox.getId()); } catch (Exception e) { // 4. 发送失败，更新重试信息 int newRetryCount = outbox.getRetryCount() + 1; if (newRetryCount >= outbox.getMaxRetry()) { // 达到最大重试次数，标记为 FAILED，需人工介入 outbox.setStatus(MessageOutboxStatus.FAILED.getCode()); log.error("Outbox message failed after max retry: {}", outbox.getId(), e); } else { // 指数退避：第一次1分钟，第二次3分钟，第三次10分钟 long delayMinutes = (long) Math.pow(2, newRetryCount) * 60; outbox.setRetryCount(newRetryCount); outbox.setNextRetryTime(LocalDateTime.now().plusSeconds(delayMinutes)); outbox.setUpdatedTime(LocalDateTime.now()); log.warn("Outbox message retry {}/{}: {}", newRetryCount, outbox.getMaxRetry(), outbox.getId(), e); } outboxMapper.updateById(outbox); } } } }

这套组合拳下来，可靠性就从“尽力而为”升级到了“企业级保障”。即使 RabbitMQ 宕机一小时，只要数据库还活着，消息就不会丢；恢复后，定时任务会自动补发所有积压消息。

4. RabbitMQ 配置与消息投递的深度调优：不只是改个 yml

很多教程把 RabbitMQ 配置简化成spring.rabbitmq.host=localhost几行，这在开发环境没问题，但一上生产，就会遇到连接池耗尽、消息堆积、消费者吞吐不足等一系列问题。TransactionalEventListener只是解决了“什么时候发”，而 RabbitMQ 的配置，则决定了“能不能发得稳、发得快、发得准”。下面这些参数，是我在线上扛住百万级订单峰值后，总结出的必调项。

4.1 连接工厂（CachingConnectionFactory）的黄金参数

Spring Boot 默认的CachingConnectionFactory是个“温柔乡”，它缓存 Channel，但默认配置过于保守。你需要手动覆盖：

spring: rabbitmq: # 1. 连接层面：增加连接数，避免高并发时连接等待 connection-timeout: 30000 # 连接超时从默认5s拉长到30s，防网络抖动 cache: channel: # 2. Channel 缓存：这是最关键的！ size: 25 # 默认25，够用；但如果你有多个 Exchange，建议按 (Exchange数 * 5) 预估 checkout-timeout: 10000 # Channel 获取超时，10秒，避免线程卡死 connection: mode: CHANNEL # 必须是 CHANNEL，不是 CONNECTION，否则每个请求都新建 TCP 连接 size: 10 # 连接池大小，一般5-10足够，过多会打满 Broker 的 socket 限制

为什么channel.size是核心？因为rabbitTemplate.convertAndSend()每次调用，都会从连接工厂里checkout一个 Channel。如果size太小（比如默认的1），高并发下所有线程都在排队等 Channel，整个应用就卡死了。checkout-timeout则是安全阀，防止某个 Channel 卡住后，其他线程无限期等待。

4.2 消息确认（Confirm）与持久化（DeliveryMode）的强制启用

默认情况下，Spring 的RabbitTemplate是“发完即走”，不关心 Broker 是否真的收到了。这在测试环境无所谓，但在生产，必须开启publisher-confirms和publisher-returns：

spring: rabbitmq: publisher-confirms: true # 开启发布者确认 publisher-returns: true # 开启未路由消息返回 template: mandatory: true # 配合 publisher-returns，强制要求消息必须被路由

然后，在RabbitTemplateBean 中，设置回调：

@Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { RabbitTemplate template = new RabbitTemplate(connectionFactory); template.setMandatory(true); // 1. Confirm 回调：Broker 收到消息后，会异步回调此方法 template.setConfirmCallback((correlationData, ack, cause) -> { if (ack) { log.info("Message confirmed by broker: {}", correlationData.getId()); } else { log.error("Message rejected by broker: {}, cause: {}", correlationData.getId(), cause); // 这里可以触发重发逻辑，或写入告警 } }); // 2. Return 回调：消息无法路由到任何 Queue 时触发（比如 routing key 写错） template.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> { log.error("Message returned! ReplyCode: {}, ReplyText: {}, Exchange: {}, RoutingKey: {}", replyCode, replyText, exchange, routingKey); // 这里应该记录日志，并告警，因为说明你的 Exchange/Queue 绑定关系出错了 }); return template; }

同时，所有发往生产环境的消息，必须设置MessageProperties.DELIVERY_MODE_PERSISTENT。否则，Broker 重启后，内存中的消息就全丢了。RabbitTemplate默认是NON_PERSISTENT，必须显式设置：

// 在发送消息时 MessageProperties props = new MessageProperties(); props.setDeliveryMode(MessageDeliveryMode.PERSISTENT); // 关键！ Message message = new Message(payload.getBytes(), props); rabbitTemplate.send("order.exchange", "order.created", message);

4.3 消费端（Consumer）的吞吐量与幂等性设计

消息发出去了，还得有消费者来接。消费者不是越快越好，而是要“稳中求快”。关键配置在SimpleRabbitListenerContainerFactory：

@Bean public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory( ConnectionFactory connectionFactory) { SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); factory.setConnectionFactory(connectionFactory); // 1. 并发消费者数：不是越多越好！建议设为 CPU 核数 * 2 ~ * 3 factory.setConcurrentConsumers(8); // 起始消费者数 factory.setMaxConcurrentConsumers(16); // 最大消费者数，应对流量高峰 // 2. 预取数量（Prefetch Count）：这是吞吐量的命门！ // 默认是250，太高会导致消息堆积在 Consumer 内存里，Broker 无法调度 // 建议设为 10~50，具体看单条消息处理耗时 factory.setPrefetchCount(20); // 3. 手动 ACK：必须关闭 autoAck，开启 manual ack factory.setAcknowledgeMode(AcknowledgeMode.MANUAL); return factory; }

prefetchCount=20意味着，Broker 一次最多给这个 Consumer 发 20 条消息，Consumer 必须处理完并手动ack之后，Broker 才会发下一批。这避免了“消息雪崩”——一个慢 Consumer 把几千条消息全拉到内存，OOM 了。

最后，也是最重要的：消费者必须实现幂等性。因为网络不可靠，RabbitMQ 的 at-least-once 语义，意味着同一条消息可能被投递多次。你的消费逻辑，不能依赖“只来一次”的假设。最简单的幂等方案，是用message.getMessageProperties().getHeader("spring_returned_message_correlation")获取唯一 ID，结合 Redis 的SETNX命令做去重：

@RabbitListener(queues = "order.created.queue") public void onOrderCreated(Message message, Channel channel) throws IOException { String msgId = message.getMessageProperties().getHeader("spring_returned_message_correlation"); String payload = new String(message.getBody(), StandardCharsets.UTF_8); // 1. Redis 去重：key = "msg_id:" + msgId, value = "processed", expire = 1小时 Boolean isProcessed = redisTemplate.opsForValue().setIfAbsent( "msg_id:" + msgId, "processed", Duration.ofHours(1) ); if (Boolean.TRUE.equals(isProcessed)) { // 2. 真正的业务逻辑 processOrderCreatedEvent(payload); // 3. 手动 ACK channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); } else { // 4. 已处理过，拒绝并丢弃（不 requeue，避免死循环） channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false); log.warn("Duplicate message ignored: {}", msgId); } }

这套配置下来，你的 RabbitMQ 集群就能稳稳扛住每秒数千条的订单事件洪峰，而不会出现连接池耗尽、消息堆积、消费者 OOM 等典型故障。

5. 全链路可观测性：如何一眼定位“消息到底卡在哪了”

再完美的架构，没有可观测性，就是黑盒。当线上告警响起“订单创建成功，但库存没扣减”，你有 5 分钟时间定位问题。这时候，日志、指标、链路追踪，就是你的手术刀。TransactionalEventListener链路涉及三个关键节点：事务发布点 → 消息表写入点 → RabbitMQ 投递点。每个节点，都必须埋下可追踪的“锚点”。

5.1 日志埋点：用 MDC 实现全链路 TraceID 透传

Spring 的MDC（Mapped Diagnostic Context）是日志追踪的基石。我们需要在事务开始时生成一个唯一traceId，并让它贯穿整个事件生命周期：

@Component public class TraceIdMdcFilter implements HandlerInterceptor { @Override public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception { // 1. 生成 traceId，格式：TRACE-20240520-1234567890 String traceId = "TRACE-" + LocalDate.now() + "-" + System.nanoTime(); MDC.put("traceId", traceId); return true; } @Override public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception { MDC.clear(); // 请求结束，清理 MDC } }

然后，在OrderService.createOrder()开头，就把这个traceId记录下来，并作为事件的一部分传递：

@Transactional(rollbackFor = Exception.class) public Order createOrder(Order orderParam) { String traceId = MDC.get("traceId"); log.info("Start creating order, traceId: {}", traceId); // ... 创建订单实体 ... // 在事件 DTO 中加入 traceId 字段 OrderCreatedEventDTO eventDTO = new OrderCreatedEventDTO(...); eventDTO.setTraceId(traceId); // 需要在 DTO 中添加此字段 eventPublisher.publishEvent(eventDTO); return order; }

这样，从 Controller 日志、Service 日志、到TransactionalEventListener的日志，再到OutboxMessageSender的日志，所有日志行都带有相同的traceId。运维同学只要在 ELK 里搜traceId: TRACE-20240520-1234567890，就能把整条链路的日志串起来，一目了然。

5.2 指标监控：用 Micrometer 暴露关键健康度

光有日志不够，还要有实时指标。Micrometer 是 Spring Boot 2.x 的标准指标框架。我们重点监控三个黄金指标：

• outbox.pending.count: 消息表中status=0的记录数。这是最直接的“积压告警”信号。
• rabbitmq.send.success.rate: RabbitMQ 发送成功率，计算公式为(success_count / (success_count + failure_count)) * 100。
• transactional.event.latency:TransactionalEventListener方法的平均执行耗时。

在OutboxMessageSender中，注入MeterRegistry：

@Component public class OutboxMessageSender { private final Counter successCounter; private final Counter failureCounter; private final Timer eventTimer; public OutboxMessageSender(MeterRegistry meterRegistry) { this.successCounter = Counter.builder("rabbitmq.send.success") .description("RabbitMQ send success count") .register(meterRegistry); this.failureCounter = Counter.builder("rabbitmq.send.failure") .description("RabbitMQ send failure count") .register(meterRegistry); this.eventTimer = Timer.builder("transactional.event.latency") .description("Latency of transactional event listener execution") .register(meterRegistry); } public void sendPendingMessages() { Timer.Sample sample = Timer.start(meterRegistry); // ... 执行发送逻辑 ... sample.stop(eventTimer); // 记录耗时 } private void handleSuccess() { successCounter.increment(); } private void handleFailure() { failureCounter.increment(); } }

然后，在 Prometheus 的配置文件里，加上/actuator/prometheus端点，这些指标就会自动被抓取。你可以设置 Grafana 面板，当outbox.pending.count > 100且持续 5 分钟，就立刻触发企业微信告警。

5.3 链路追踪：用 Sleuth + Zipkin 还原分布式调用栈

如果系统已经接入了 Sleuth 和 Zipkin，那么TransactionalEventListener的调用，会自动被纳入链路。但有一个细节必须注意：@TransactionalEventListener的方法，默认是运行在TaskExecutor的线程池里，而 Sleuth 的TraceContextHolder是 ThreadLocal 的，会丢失。

解决方案，是在@TransactionalEventListener方法上，显式启用@NewSpan（Sleuth 注解）：

@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT, fallbackExecution = true) @NewSpan("handle-order-created-event") // 显式创建新 Span public void handleOrderCreatedEvent(OrderCreatedEventDTO event) { // ... 业务逻辑 ... }

这样，Zipkin 的调用链路图里，就会清晰地看到：HTTP POST /order→createOrder()→handleOrderCreatedEvent()→sendToRabbitMQ()这样一条完整的、带耗时的调用链。任何一个环节变慢，都能在链路图上一眼识别。

提示：@NewSpan不是万能的。如果handleOrderCreatedEvent里又调用了其他@Async方法，那个@Async方法里的 Span 还是会丢失。这时，需要用Tracing.currentTracer().currentSpan()手动传递 SpanContext，但这已超出本文范围。记住原则：所有跨线程的调用，都要主动考虑 TraceID 的透传。

这套可观测性体系建立起来后，“消息卡在哪了”就不再是玄学问题。你可以在 30 秒内，通过 Grafana 看到积压趋势，通过 Kibana 搜到具体 traceId 的所有日志，再通过 Zipkin 看到哪一跳耗时最长。这才是一个成熟事务消息系统的标配。

6. 性能压测与边界验证：1000 TPS 下的实测数据与调优结论

理论和配置都讲完了，最终要回归到数字。我用 JMeter 对这套TransactionalEventListener + MyBatis-Plus + RabbitMQ链路，进行了为期一周的压测，目标是验证它能否稳定支撑 1000 TPS（每秒 1000 笔订单）。测试环境是：4核8G 的云服务器，MySQL 8.0（SSD），RabbitMQ 3.11（单节点，非集群），Spring Boot 2.7.18。下面是最关键的实测数据和对应的调优结论。

6.1 基线测试：未做任何优化的原始表现

初始配置就是 Spring Boot 的 RabbitMQ 默认配置（channel.size=1,prefetchCount=250），TransactionalEventListener也未加fallbackExecution=true。结果惨不忍睹：

结论很清晰：默认配置在高并发下是灾难性的。瓶颈不在数据库，也不在 RabbitMQ，而是在 Spring 的连接工厂和序列化开销上。

6.2 第一轮调优：聚焦连接池与序列化

根据前面章节的建议，我们做了三项关键调整：

1. CachingConnectionFactory.channel.size从 1 改为 25；
2. SimpleRabbitListenerContainerFactory.prefetchCount从 250 改为 20；
3. OrderCreatedEventDTO的序列化，从JSON.toJSONString()改为 Jackson 的ObjectMapper.writeValueAsString()，并复用ObjectMapper实例（避免频繁创建）。

效果立竿见影：

注意：Jackson 比 FastJSON 在高并发下的性能更稳定，且线程安全。ObjectMapper是线程安全的，可以全局单例。

6