1. 程式人生 > >RabbitMQ系列(四)RabbitMQ事務和Confirm傳送方訊息確認——深入解讀(轉載)

RabbitMQ系列(四)RabbitMQ事務和Confirm傳送方訊息確認——深入解讀(轉載)

原文地址:https://yq.aliyun.com/articles/629858

 

RabbitMQ事務和Confirm傳送方訊息確認——深入解讀

RabbitMQ系列文章

  1. RabbitMQ在Ubuntu上的環境搭建
  2. 深入瞭解RabbitMQ工作原理及簡單使用
  3. RabbitMQ交換器Exchange介紹與實踐
  4. RabbitMQ事務和Confirm傳送方訊息確認——深入解讀
  5. 使用Docker部署RabbitMQ叢集
  6. 你不知道的RabbitMQ叢集架構全解

引言

根據前面的知識(深入瞭解RabbitMQ工作原理及簡單使用Rabbit的幾種工作模式介紹與實踐

)我們知道,如果要保證訊息的可靠性,需要對訊息進行持久化處理,然而訊息持久化除了需要程式碼的設定之外,還有一個重要步驟是至關重要的,那就是保證你的訊息順利進入Broker(代理伺服器),如圖所示:

img_eb038c71d910f4b9e8974913765314a0.png

正常情況下,如果訊息經過交換器進入佇列就可以完成訊息的持久化,但如果訊息在沒有到達broker之前出現意外,那就造成訊息丟失,有沒有辦法可以解決這個問題?

RabbitMQ有兩種方式來解決這個問題:

  1. 通過AMQP提供的事務機制實現;
  2. 使用傳送者確認模式實現;

一、事務使用

事務的實現主要是對通道(Channel)的設定,主要的方法有三個:

  1. channel.txSelect()宣告啟動事務模式;

  2. channel.txComment()提交事務;

  3. channel.txRollback()回滾事務;

從上面的可以看出事務都是以tx開頭的,tx應該是transaction extend(事務擴充套件模組)的縮寫,如果有準確的解釋歡迎在部落格下留言。

我們來看具體的程式碼實現:

// 建立連線
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抓個包看看,如圖所示:

img_8005e064b5174fb23501fe5115fb24e3.png

輸入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的情況是不支援事務的,也就是說你即使在收到訊息之後在回滾事務也是於事無補的,佇列已經把訊息移除了;

二、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)); } });

非同步模式的優點,就是執行效率高,不需要等待訊息執行完,只需要監聽訊息即可,以上非同步返回的資訊如下:

img_bbe7b01bad34db80b25cfea54fde9dc5.png

可以看出,程式碼是非同步執行的,訊息確認有可能是批量確認的,是否批量確認在於返回的multiple的引數,此引數為bool值,如果true表示批量執行了deliveryTag這個值以前的所有訊息,如果為false的話表示單條確認。

Confirm效能測試

測試前提:與事務一樣,我們傳送1w條訊息。

方式一:Confirm普通模式

  • 執行花費時間:2253s
  • 執行花費時間:2018s
  • 執行花費時間:2043s

方式二:Confirm批量模式

  • 執行花費時間:1576s
  • 執行花費時間:1400s
  • 執行花費時間:1374s

方式三:Confirm非同步監聽方式

  • 執行花費時間:1498s
  • 執行花費時間:1368s
  • 執行花費時間:1363s

總結

綜合總體測試情況來看:Confirm批量確定和Confirm非同步模式效能相差不大,Confirm模式要比事務快10倍左右。