消息发送失败时Repository未回滚的问题及方案咨询

你想到的**发布确认（Publisher Confirms）**方案是完全正确的，这是解决“确保MQ消息发送成功后再执行DB持久化”这类一致性问题的标准方案，以下是具体分析和优化建议：

一、为什么这个方案能解决你的问题 #

• 核心逻辑是通过CorrelationData的Future阻塞等待RabbitMQ的服务器端确认，只有收到明确的肯定确认（ack）后，再执行persistService.save()，从根源上保证了DB写入仅发生在MQ消息确实投递成功之后。
• 如果MQ发送失败（网络断连、Exchange/RoutingKey不存在、超时等），Future会抛出异常，此时事务会自动回滚，同时你可以手动拒绝原消息，让它进入死信队列（DLQ），完全符合你的需求。

二、性能顾虑的缓解方法 #

担心阻塞导致性能下降是合理的，但可以通过以下方式优化：

• 设置合理的超时时间：根据你的业务网络延迟情况设置超时（比如300-500毫秒），避免无意义的长时间阻塞。如果超时，直接判定为发送失败即可。
• 异步确认（非阻塞）：如果不想阻塞主线程，可以使用CorrelationData的回调方法替代Future阻塞，在回调中执行DB持久化。但这种方式需要注意DB操作的事务管理（原事务已结束，需单独开启事务），以及线程安全问题，复杂度稍高，适合对性能要求极高的场景：

  correlationData.getFuture().addCallback(
      ack -> {
          if (Boolean.TRUE.equals(ack)) {
              // 异步执行DB操作，需单独管理事务
              transactionTemplate.execute(status -> {
                  persistService.save(new MyEntity());
                  doSomethingElse();
                  return null;
              });
              // 手动确认原消息
              try {
                  channel.basicAck(deliveryTag, false);
              } catch (IOException e) {
                  // 处理确认失败的情况
              }
          } else {
              // 发送失败，拒绝原消息
              try {
                  channel.basicReject(deliveryTag, false);
              } catch (IOException e) {
                  // 处理拒绝失败的情况
              }
          }
      },
      ex -> {
          // 发送异常，拒绝原消息
          try {
              channel.basicReject(deliveryTag, false);
          } catch (IOException e) {
              // 处理拒绝失败的情况
          }
      }
  );
  

• 批量处理优化：如果业务允许批量接收消息，可以攒一批消息发送后批量等待确认，减少单次阻塞的开销，但仅适用于支持批量处理的场景。

三、具体实现步骤 #

1. 开启RabbitTemplate的发布确认 #

@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
    RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    // 新版Spring AMQP用这个配置开启发布确认
    rabbitTemplate.setPublisherConfirmType(ConfirmType.CORRELATED);
    return rabbitTemplate;
}

2. 修改监听方法与业务逻辑 #

需要关闭自动确认，改为手动确认/拒绝原消息：

@RabbitListener(queues = "your-queue-name", ackMode = "MANUAL")
public void process(Message myMessage, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws Exception {
    Event event = ...从myMessage获取事件;
    handleMessage(event, channel, deliveryTag);
}

@Transactional
public void handleMessage(Event event, Channel channel, long deliveryTag) throws IOException, ExecutionException, InterruptedException, TimeoutException {
    ObjectToSend objectToSend = ...从event获取objectToSend;
    CorrelationData correlationData = new CorrelationData();
    
    rabbitTemplate.convertAndSend(exchange1, routingKey1, objectToSend, correlationData);
    
    // 等待MQ确认，设置超时时间
    Boolean ack = correlationData.getFuture().get(500, TimeUnit.MILLISECONDS);
    
    if (Boolean.TRUE.equals(ack)) {
        // 确认发送成功，执行DB持久化
        persistService.save(new MyEntity());
        doSomethingElse();
        // 手动确认原消息，避免重复消费
        channel.basicAck(deliveryTag, false);
    } else {
        // 发送失败，拒绝原消息到DLQ（第二个参数false表示不重新入队）
        channel.basicReject(deliveryTag, false);
        throw new RuntimeException("MQ消息发送失败，原消息已转入DLQ");
    }
}

四、额外注意事项 #

• 确保RabbitMQ服务器开启了发布确认功能（默认开启，集群环境无需额外配置）。
• 超时时间的设置要结合业务实际，既不能太短导致正常网络波动下的误判，也不能太长影响系统吞吐量。
• 若遇到超时情况，可以考虑增加1-2次重试发送（注意幂等性，避免重复消息），再拒绝消息到DLQ。

内容的提问来源于stack exchange，提问作者obe6