RabbitMQ学习笔记 - 事务机制

2022-01-12 23:07 121阅读 0赞

**参考：<<RabbitMQ实战指南>>**

通过消息持久化可以解决因为服务器的异常崩溃而导致的消息丢失。但是，当消息的生产者将消息发送出去之后，并不能知道消息有没有正确到达服务器。如果不进行特殊配置，默认情况下发送消息的操作是不会返回任何信息给生产者的。

如果消息在到达服务器之前丢失，持久化操作解决不了这个问题，针对这个问题，有两种解决方式：

*  事务机制
 *  发送方确认机制（publisher confirm）

## 事务机制 ##

RabbitMQ客户端中与事务机制相关的三个方法：

*  channel.txSelect：将当前信道设置为事务模式
 *  channel.txCommit：提交事务
 *  channel.txRollback：事务回滚

示例代码：

...
    // 声明交换器
    String exchangeName = "direct.exchange.test.tx";
    AMQP.Exchange.DeclareOk declareOk = channel.exchangeDeclare(exchangeName, "direct", true);
    // 声明队列
    String queueName = "direct.queue.test.tx";
    channel.queueDeclare(queueName, true, false, false, null);
    // 绑定交换器和队列
    String routingKey = "direct.routing-key.test.tx";
    channel.queueBind(queueName, exchangeName, routingKey);
    
    try { 
        channel.txSelect();
        channel.basicPublish(exchangeName, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, "Test Msg".getBytes());
        channel.txCommit();
    } catch (Exception e) { 
        e.printStackTrace();
        channel.txRollback();
    }
    ...

上面代码对应AMQP协议流转过程：

*  客户端发送Tx.Select，将信道置为事务模式
 *  Broker回复Tx.Select-Ok，确认将信道置为事务模式
 *  发送完消息之后，客户端发送Tx.Commit提交事务
 *  Broker回复Tx.Commit-Ok，确认事务提交
 *  若在提交事务之前发生异常，客户端发送Tx.Rollback回滚事务
 *  Broker回复Tx.Rollback-Ok，确认事务回滚

（1）正常情况下事务机制流程：  
![AMQP协议流转过程][AMQP]  
（2）异常情况下回滚流程：  
![AMQP协议流转过程][AMQP 1]

注：

*  事务确实能够解决发送方和RabbitMQ之间消息确认的问题，只有消息成功被接收，事务才能提交成功，否则捕获异常后进行事务回滚，此时可以进行消息重发。
 *  事务机制会使得RabbitMQ的性能急剧下降。

测试：  
发送一万条消息，不考虑创建、销毁连接和信道的时间（因机器、网络等环境不同，测试结果差别比较大）

*  事务：耗时164814ms
 *  无事务：耗时367ms

总的来看，加了事务性能下降比较明显，如果不是特别重要的消息，尽量避免使用事务。如果仍然想要保证消息正确投递，那么还可以使用 [<<发送方确认机制>>][Link 1] 。

[AMQP]: /images/20220112/a076a7fcfb8f474eb18a408602e48518.png
[AMQP 1]: /images/20220112/73511210c27b4bf28e10b8d3e773e519.png
[Link 1]: https://blog.csdn.net/mytt_10566/article/details/92838069