RabbitMQ 事务消息的核心机制与 Java 实现示例

// 生产者示例（原生 Java 客户端）
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
try (Connection connection = factory.newConnection();
     Channel channel = connection.createChannel()) {

    channel.txSelect(); // 开启事务
    try {
        // 发送消息（事务中）
        channel.basicPublish("", "myQueue", null, "事务消息内容".getBytes());
        
        // 模拟本地业务逻辑（如数据库操作）
        boolean success = processBusinessLogic();
        if (success) {
            channel.txCommit(); // 提交事务
        } else {
            channel.txRollback(); // 回滚事务
        }
    } catch (Exception e) {
        channel.txRollback(); // 异常回滚
    }
}

// Spring Boot 配置（application.properties）
spring.rabbitmq.host=localhost
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

// 发送事务消息示例
@Autowired
private RabbitTemplate rabbitTemplate;

public void sendTransactionalMessage() {
    rabbitTemplate.execute(channel -> {
        channel.txSelect();
        try {
            // 发送消息
            rabbitTemplate.convertAndSend("myExchange", "routingKey", "事务消息");
            
            // 本地业务逻辑
            if (businessOperation()) {
                channel.txCommit();
            } else {
                channel.txRollback();
            }
            return null;
        } catch (Exception e) {
            channel.txRollback();
            throw e;
        }
    });
}