大家好，我是Z哥。

最近在項(xiàng)目中遇到了一個(gè)使用 RabbitMQ 時(shí)的問題，這個(gè)問題我覺得還是有一定普適性的，和大家分享一下，避免大家后續(xù)在同一個(gè)問題上犯錯(cuò)。

消息隊(duì)列(MQ)是在軟件開發(fā)中很常用的中間件，如果一個(gè)程序需要協(xié)調(diào)另一個(gè)程序進(jìn)行數(shù)據(jù)的“write”操作，并且不關(guān)心“write”的結(jié)果，則便會(huì)選擇它。它是一個(gè)保存消息(數(shù)據(jù))的容器，由它來確保消息一定被送達(dá)到目標(biāo)程序。

打個(gè)比喻來說，消息隊(duì)列就是一個(gè)郵差，它負(fù)責(zé)將信件(消息)從源頭送往目的地，并且根據(jù)信件重要性的不同，提供當(dāng)面簽收確認(rèn)或者直接投放兩種服務(wù)。

RabbitMQ 就是一個(gè)典型的消息隊(duì)列，以 AMQP 為標(biāo)準(zhǔn)。歷史也比較悠久，大概是從 2007 年研發(fā)出來的，用的編程語言Erlang也同樣具有年代感。

需要簡單介紹一下 Erlang 的特點(diǎn)，它對(duì)我們理解 RabbitMQ 有很大的幫助。

Erlang 是一種運(yùn)行于“虛擬機(jī)”(類似 JVM)的解釋性語言。是一個(gè)結(jié)構(gòu)化，動(dòng)態(tài)類型編程語言，內(nèi)建并行計(jì)算支持。使用 Erlang 編寫出的應(yīng)用運(yùn)行時(shí)通常由成千上萬個(gè)輕量級(jí)“進(jìn)程”(并非傳統(tǒng)意義上的進(jìn)程)組成，并通過消息傳遞相互通訊。進(jìn)程間上下文切換對(duì)于 Erlang 來說僅僅 只是一兩個(gè)環(huán)節(jié)，比起 C 程序的線程切換要高效得多得多了。

基于百度百科資料進(jìn)行整理

不管是什么 MQ 中間件，作為消息的生產(chǎn)方和消費(fèi)方都需要和 MQ 的服務(wù)端建立連接進(jìn)行通訊。

一般這個(gè)連接都會(huì)使用 TCP 協(xié)議，在 RabbitMQ 里也不例外。大多數(shù) RabbitMQ 的 SDK 都會(huì)將連接封裝為一個(gè)「Connection」對(duì)象。

還沒完，大多數(shù)的 MQ 中間件還會(huì)在「Connection」的基礎(chǔ)上增加一個(gè)「Channel」的概念，以通過復(fù)用的方式提高 TCP 連接的利用率，因?yàn)榻⒑弯N毀 TCP 連接是非常昂貴的開銷。在 RabbitMQ 中的復(fù)用 TCP 連接方式是「Non-blocking I/O」的模式。

關(guān)于NIO，「Non-blocking I/O」的概念，有感興趣的話可以跳轉(zhuǎn)去看之前寫的這篇文章。(分布式系統(tǒng)關(guān)注點(diǎn)——阻塞與非阻塞有什么區(qū)別?)

多說一句，任何方案都不是“銀彈”。當(dāng)每個(gè) Channel 的流量不是很大時(shí)，復(fù)用單一的 Connection 可以在產(chǎn)生性能瓶頸的情況下有效地節(jié)省 TCP 連接資源。但是 Channel 本身的流量很大時(shí)，這時(shí)候多個(gè) Channel 復(fù)用一個(gè) Connection 就會(huì)產(chǎn)生性能瓶頸，進(jìn)而使整體的流量被限制了。此時(shí)就需要開辟多個(gè) Connection，將這些 Channel 均攤到這些 Connection 中，至于哪些 Channel 使用那個(gè) Connection 以及Connection 與 Channel 之間的數(shù)量關(guān)系是多少，需要根據(jù)業(yè)務(wù)自身的實(shí)際情況進(jìn)行調(diào)節(jié)。

Channel 在 AMQP 中是一個(gè)很重要的概念，大多數(shù)操作都是在信道這個(gè)層面展開的。比如, channel.exchangeDeclare、channel.queueDeclare、channel.basicPublish、channel.basicConsume 等方法。RabbitMQ 相關(guān)的 API 與 AMQP 緊密相連，比如 channel.basicPublish 對(duì)應(yīng) AMQP 的 Basic.Publish 命令。

可能你要問了，Channel 是不是也能像 Connection 一樣被復(fù)用?這是個(gè)好問題，也是我們這次遇到問題的關(guān)鍵點(diǎn)。

結(jié)論是：可以，但是需要自己保證客戶端對(duì) Channel 訪問的線程安全問題，因?yàn)樵?Channel 的另一端，在 RabbitMQ 的服務(wù)端，每個(gè) Channel 由一個(gè)單獨(dú)的“進(jìn)程”所管理，如果由于多線程復(fù)用Channel 導(dǎo)致數(shù)據(jù)幀亂序了，RabbitMQ 的服務(wù)端會(huì)主動(dòng)關(guān)閉整個(gè) Connection 。

因此，我們這次犯的錯(cuò)誤就是多線程復(fù)用了同一個(gè) Channel 導(dǎo)致的問題。所以，如果你也用到 streadway/amqp 這個(gè)庫的話，需要特別注意這點(diǎn)。

不過，不同語言的SDK內(nèi)部實(shí)現(xiàn)不同，我們分別使用 Golang 的 AMQP 庫 streadway/amqp，和 RabbitMQ 官方提供的 C# 版本的庫分別模擬過同樣的場景，前者出現(xiàn)問題，后者卻沒有問題。

受限于時(shí)間原因，沒有具體去核實(shí) C# 庫的源碼，主觀猜測是 C# 庫內(nèi)部多做了一些對(duì)于單個(gè) Channel 的線程安全處理。

最后，我整理了三點(diǎn)使用 streadway/amqp 庫的最佳實(shí)踐，你可以看看：

01

golang 中使用 streadway/amqp 時(shí)，需要保證每一個(gè)線程單獨(dú)一個(gè) Channel。

streadway/amqp 庫中的獲取一個(gè) Channel 的方法「Connection.channel()」是線程安全的。但是內(nèi)部有一個(gè) defaultChannelMax 的參數(shù)對(duì) Channel 的數(shù)量進(jìn)行了限制，默認(rèn)是 (2 << 10) - 1，2047。這個(gè)需要注意：

02

我們可以通過調(diào)用 amqp.DialConfig(url string, config Config) 來調(diào)整個(gè)限制。

但是，并不是你調(diào)整了多少就是多少，還需要和 RabbitMQ 服務(wù)端的配置進(jìn)行 min() 函數(shù)的處理，最終為兩者的最小值。

Tips：特別是用云廠商的 MQ 產(chǎn)品，因?yàn)殡A梯收費(fèi)的原因會(huì)對(duì)很多性能參數(shù)做限制，需要格外關(guān)注這點(diǎn)，比如某版本的阿里云 RabbitMQ 實(shí)例限制是單個(gè) Connection 最多 64 個(gè) Channel)

03

正如前面對(duì) Erlang 的簡單介紹，Erlang 是一個(gè)天然支持多“進(jìn)程”設(shè)計(jì)的語言，所以在 RabbitMQ 的服務(wù)端設(shè)計(jì)中，每一個(gè) Queue，每一個(gè) Connection 都是單獨(dú)的一個(gè)“進(jìn)程”。因此如果你想盡可能地壓榨 RabbitMQ 性能，可以通過建立更多的 Connection 或者創(chuàng)建更多的 Queue 來實(shí)現(xiàn)，當(dāng)然需要注意到 Connection 的創(chuàng)建和銷毀的性能開銷問題。