RabbitMQ系列(四)RabbitMQ事务和Confirm发送方消息确认——深刻解读

RabbitMQ事务和Confirm发送方消息确认——深刻解读html

RabbitMQ系列文章

  1. RabbitMQ在Ubuntu上的环境搭建
  2. 深刻了解RabbitMQ工做原理及简单使用
  3. RabbitMQ交换器Exchange介绍与实践
  4. RabbitMQ事务和Confirm发送方消息确认——深刻解读
  5. 使用Docker部署RabbitMQ集群
  6. 你不知道的RabbitMQ集群架构全解

引言

根据前面的知识(深刻了解RabbitMQ工做原理及简单使用Rabbit的几种工做模式介绍与实践)咱们知道,若是要保证消息的可靠性,须要对消息进行持久化处理,然而消息持久化除了须要代码的设置以外,还有一个重要步骤是相当重要的,那就是保证你的消息顺利进入Broker(代理服务器),如图所示:java

正常状况下,若是消息通过交换器进入队列就能够完成消息的持久化,但若是消息在没有到达broker以前出现意外,那就形成消息丢失,有没有办法能够解决这个问题?服务器

RabbitMQ有两种方式来解决这个问题:架构

  1. 经过AMQP提供的事务机制实现;
  2. 使用发送者确认模式实现;

1、事务使用

事务的实现主要是对信道(Channel)的设置,主要的方法有三个:异步

  1. channel.txSelect()声明启动事务模式;ide

  2. channel.txComment()提交事务;性能

  3. channel.txRollback()回滚事务;测试

从上面的能够看出事务都是以tx开头的,tx应该是transaction extend(事务扩展模块)的缩写,若是有准确的解释欢迎在博客下留言。代理

咱们来看具体的代码实现:code

// 建立链接
ConnectionFactory factory = new ConnectionFactory();
factory.setUsername(config.UserName);
factory.setPassword(config.Password);
factory.setVirtualHost(config.VHost);
factory.setHost(config.Host);
factory.setPort(config.Port);   
Connection conn = factory.newConnection();
// 建立信道
Channel channel = conn.createChannel();
// 声明队列
channel.queueDeclare(_queueName, true, false, false, null);
String message = String.format("时间 => %s", new Date().getTime());
try {
    channel.txSelect(); // 声明事务
    // 发送消息
    channel.basicPublish("", _queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));
    channel.txCommit(); // 提交事务
} catch (Exception e) {
    channel.txRollback();
} finally {
    channel.close();
    conn.close();
}

注意:用户需把config.xx配置成本身Rabbit的信息。

从上面的代码咱们能够看出,在发送消息以前的代码和以前介绍的都是同样的,只是在发送消息以前,须要声明channel为事务模式,提交或者回滚事务便可。

了解了事务的实现以后,那么事务到底是怎么执行的,让咱们来使用wireshark抓个包看看,如图所示:

输入ip.addr==rabbitip && amqp查看客户端和rabbit之间的通信,能够看到交互流程:

  • 客户端发送给服务器Tx.Select(开启事务模式)
  • 服务器端返回Tx.Select-Ok(开启事务模式ok)
  • 推送消息
  • 客户端发送给事务提交Tx.Commit
  • 服务器端返回Tx.Commit-Ok

以上就完成了事务的交互流程,若是其中任意一个环节出现问题,就会抛出IoException移除,这样用户就能够拦截异常进行事务回滚,或决定要不要重复消息。

那么,既然已经有事务了,没什么还要使用发送方确认模式呢,缘由是由于事务的性能是很是差的。事务性能测试

事务模式,结果以下:

  • 事务模式,发送1w条数据,执行花费时间:14197s
  • 事务模式,发送1w条数据,执行花费时间:13597s
  • 事务模式,发送1w条数据,执行花费时间:14216s

非事务模式,结果以下:

  • 非事务模式,发送1w条数据,执行花费时间:101s
  • 非事务模式,发送1w条数据,执行花费时间:77s
  • 非事务模式,发送1w条数据,执行花费时间:106s

从上面能够看出,非事务模式的性能是事务模式的性能高149倍,个人电脑测试是这样的结果,不一样的电脑配置略有差别,但结论是同样的,事务模式的性能要差不少,那有没有既能保证消息的可靠性又能兼顾性能的解决方案呢?那就是接下来要讲的Confirm发送方确认模式。

扩展知识

咱们知道,消费者可使用消息自动或手动发送来确认消费消息,那若是咱们在消费者模式中使用事务(固然若是使用了手动确认消息,彻底用不到事务的),会发生什么呢?

消费者模式使用事务

假设消费者模式中使用了事务,而且在消息确认以后进行了事务回滚,那么RabbitMQ会产生什么样的变化?

结果分为两种状况:

  1. autoAck=false手动应对的时候是支持事务的,也就是说即便你已经手动确认了消息已经收到了,但在确认消息会等事务的返回解决以后,在作决定是确认消息仍是从新放回队列,若是你手动确认如今以后,又回滚了事务,那么已事务回滚为主,此条消息会从新放回队列;
  2. autoAck=true若是自定确认为true的状况是不支持事务的,也就是说你即便在收到消息以后在回滚事务也是于事无补的,队列已经把消息移除了;

2、Confirm发送方确认模式

Confirm发送方确认模式使用和事务相似,也是经过设置Channel进行发送方确认的。

Confirm的三种实现方式:

方式一:channel.waitForConfirms()普通发送方确认模式;

方式二:channel.waitForConfirmsOrDie()批量确认模式;

方式三:channel.addConfirmListener()异步监听发送方确认模式;

方式一:普通Confirm模式

// 建立链接
ConnectionFactory factory = new ConnectionFactory();
factory.setUsername(config.UserName);
factory.setPassword(config.Password);
factory.setVirtualHost(config.VHost);
factory.setHost(config.Host);
factory.setPort(config.Port);
Connection conn = factory.newConnection();
// 建立信道
Channel channel = conn.createChannel();
// 声明队列
channel.queueDeclare(config.QueueName, false, false, false, null);
// 开启发送方确认模式
channel.confirmSelect();
String message = String.format("时间 => %s", new Date().getTime());
channel.basicPublish("", config.QueueName, null, message.getBytes("UTF-8"));
if (channel.waitForConfirms()) {
    System.out.println("消息发送成功" );
}

看代码能够知道,咱们只须要在推送消息以前,channel.confirmSelect()声明开启发送方确认模式,再使用channel.waitForConfirms()等待消息被服务器确认便可。

方式二:批量Confirm模式

// 建立链接
ConnectionFactory factory = new ConnectionFactory();
factory.setUsername(config.UserName);
factory.setPassword(config.Password);
factory.setVirtualHost(config.VHost);
factory.setHost(config.Host);
factory.setPort(config.Port);
Connection conn = factory.newConnection();
// 建立信道
Channel channel = conn.createChannel();
// 声明队列
channel.queueDeclare(config.QueueName, false, false, false, null);
// 开启发送方确认模式
channel.confirmSelect();
for (int i = 0; i < 10; i++) {
    String message = String.format("时间 => %s", new Date().getTime());
    channel.basicPublish("", config.QueueName, null, message.getBytes("UTF-8"));
}
channel.waitForConfirmsOrDie(); //直到全部信息都发布,只要有一个未确认就会IOException
System.out.println("所有执行完成");

以上代码能够看出来channel.waitForConfirmsOrDie(),使用同步方式等全部的消息发送以后才会执行后面代码,只要有一个消息未被确认就会抛出IOException异常。

方式三:异步Confirm模式

// 建立链接
ConnectionFactory factory = new ConnectionFactory();
factory.setUsername(config.UserName);
factory.setPassword(config.Password);
factory.setVirtualHost(config.VHost);
factory.setHost(config.Host);
factory.setPort(config.Port);
Connection conn = factory.newConnection();
// 建立信道
Channel channel = conn.createChannel();
// 声明队列
channel.queueDeclare(config.QueueName, false, false, false, null);
// 开启发送方确认模式
channel.confirmSelect();
for (int i = 0; i < 10; i++) {
    String message = String.format("时间 => %s", new Date().getTime());
    channel.basicPublish("", config.QueueName, null, message.getBytes("UTF-8"));
}
//异步监听确认和未确认的消息
channel.addConfirmListener(new ConfirmListener() {
    @Override
    public void handleNack(long deliveryTag, boolean multiple) throws IOException {
        System.out.println("未确认消息,标识:" + deliveryTag);
    }
    @Override
    public void handleAck(long deliveryTag, boolean multiple) throws IOException {
        System.out.println(String.format("已确认消息,标识:%d,多个消息:%b", deliveryTag, multiple));
    }
});

异步模式的优势,就是执行效率高,不须要等待消息执行完,只须要监听消息便可,以上异步返回的信息以下:

能够看出,代码是异步执行的,消息确认有多是批量确认的,是否批量确认在于返回的multiple的参数,此参数为bool值,若是true表示批量执行了deliveryTag这个值之前的全部消息,若是为false的话表示单条确认。

Confirm性能测试

测试前提:与事务同样,咱们发送1w条消息。

方式一:Confirm普通模式

  • 执行花费时间:2253s
  • 执行花费时间:2018s
  • 执行花费时间:2043s

方式二:Confirm批量模式

  • 执行花费时间:1576s
  • 执行花费时间:1400s
  • 执行花费时间:1374s

方式三:Confirm异步监听方式

  • 执行花费时间:1498s
  • 执行花费时间:1368s
  • 执行花费时间:1363s

总结

综合整体测试状况来看:Confirm批量肯定和Confirm异步模式性能相差不大,Confirm模式要比事务快10倍左右。