Spring AMQP事务发布消息时，队列不存在未抛出异常问题

问题根源 #

你对Spring AMQP事务机制的理解存在偏差：channelTransacted(true)和@Transactional仅保证本地事务（如数据库操作）与RabbitMQ事务的一致性——即要么两者都提交，要么都回滚，但它们不会主动检测消息是否成功路由到目标队列。

当你发送消息到不存在的队列时，RabbitMQ默认会直接丢弃消息，不会主动通知生产者，因此事务依然会正常提交，代码不会抛出异常。

正确实现方案 #

要实现“消息未入队则快速失败并抛异常”，需要结合Mandatory模式或Publisher Confirm机制，配合事务一起使用：

1. 开启Mandatory模式捕获路由失败 #

Mandatory模式开启后，当消息无法路由到任何队列时，RabbitMQ会将消息返回给生产者，我们可以通过回调捕获该事件并抛出异常。

配置RabbitTemplate

@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
    RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    // 开启RabbitMQ事务，与Spring事务绑定
    rabbitTemplate.setChannelTransacted(true);
    // 开启Mandatory模式，强制RabbitMQ返回路由失败的消息
    rabbitTemplate.setMandatory(true);
    // 设置路由失败回调，抛出异常触发事务回滚
    rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
        String errorMsg = String.format("消息路由失败: 交换机=%s, 路由键=%s, 错误码=%d, 错误信息=%s",
                exchange, routingKey, replyCode, replyText);
        throw new AmqpException(errorMsg);
    });
    return rabbitTemplate;
}

配置事务管理器

确保Spring能管理RabbitMQ事务：

@Bean
public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory) {
    return new RabbitTransactionManager(connectionFactory);
}

2. 配合@Transactional实现事务一致性 #

在业务方法上添加@Transactional，确保数据库操作与消息发送要么都成功，要么都回滚：

@Service
public class TaskService {
    @Autowired
    private TaskRepository taskRepository;
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Transactional(rollbackFor = Exception.class)
    public void submitTask(Task task) {
        // 保存任务到数据库
        taskRepository.save(task);
        // 发送消息到目标队列
        rabbitTemplate.convertAndSend("target-queue", task);
    }
}

此时如果队列不存在，ReturnCallback会触发并抛出AmqpException，Spring事务会自动回滚数据库的保存操作，同时异常会向上传递，即时告知用户任务未入队。

3. 可选：开启Publisher Confirm机制（更全面的消息确认） #

如果需要确认消息是否被RabbitMQ成功接收（不仅仅是路由），可以开启Publisher Confirm机制：

@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
    RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    rabbitTemplate.setChannelTransacted(true);
    rabbitTemplate.setMandatory(true);
    // 设置Publisher Confirm类型为CORRELATED，开启消息确认
    rabbitTemplate.setPublisherConfirmType(CorrelationData.ConfirmType.CORRELATED);
    // 消息未被RabbitMQ接收时抛出异常
    rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
        if (!ack) {
            throw new AmqpException("消息未被RabbitMQ确认接收: " + cause);
        }
    });
    rabbitTemplate.setReturnCallback(...); // 保留路由失败回调
    return rabbitTemplate;
}

关键注意事项 #

• 确保业务方法被Spring代理：不要用private方法，避免内部调用（内部调用不会触发Spring事务代理）。
• 如果使用自定义交换机，需确保交换机存在；若交换机也不存在，RabbitMQ会直接拒绝消息，此时会触发ConfirmCallback的ack=false。

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