快捷導(dǎo)航

深入理解Spring中RabbitMQ的Channel

更新時(shí)間：2023年08月28日 09:17:12 作者：Sam_Deep_Thinking

這篇文章主要介紹了深入理解Spring中RabbitMQ的Channel,在RabbitMq中,channel表示邏輯連接或者叫虛擬連接,是棣屬于TCP連接的,一個(gè)TCP連接里可以創(chuàng)建多個(gè)channel,在Rabbit MQ里,消息的發(fā)送和接收都是基于channel的,需要的朋友可以參考下

概述

在 AMQP 協(xié)議中，有 channel 的概念，在 RabbitMq 中, channel 表示邏輯連接或者叫虛擬連接,是棣屬于 TCP 連接的。

一個(gè) TCP 連接里可以創(chuàng)建多個(gè) channel ,在 Rabbit MQ 里，消息的發(fā)送和接收都是基于 channel 的。

connection和channel的關(guān)系

有了 TCP 連接后,還需要 channel 的原因如下:

創(chuàng)建和銷毀TCP連接很耗時(shí);
打開太多TCP連接,耗操作系統(tǒng)資源，并發(fā)量大到一定程度,系統(tǒng)的吞吐量會(huì)降低；
使用一個(gè)connection多channel的方式,可以提升連接的利用率。

因此采用多個(gè) channel 多路復(fù)用一個(gè) TCP 連接的方式才比較合理。

channel線程不安全

channel 不是線程安全的，線程并發(fā)的去訪問(wèn)同一個(gè) channel 會(huì)出問(wèn)題。

這里有幾種處理方式：

全局公用一個(gè)channel且使用全局鎖,讓操作channel排隊(duì).這種明顯性能是不行的；
一個(gè)線程對(duì)應(yīng)創(chuàng)建一個(gè)新的channel，但是要處理好一個(gè)連接能支撐的最大channel數(shù)量；
一個(gè)線程對(duì)應(yīng)一個(gè)channel,但是是從channel池子拿的,不是每次都創(chuàng)建新的.一旦一個(gè)線程完成了一個(gè)channel的使用，它將返回到池中，從而使該channel可用于另一個(gè)線程。

量不大的話,使用第二種方式就可以了。量大的話，建議使用第三種方式，畢竟創(chuàng)建和銷毀 channel 也是耗時(shí)耗資源的.在 spring amqp 中，提供了一個(gè)緩存 channel 的方案。

可以在創(chuàng)建 CachingConnectionFactory 時(shí)指定緩存的模式。

connectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);
connectionFactory.setChannelCacheSize(25);

上面的兩行代碼,表示 channel 共用唯一的一個(gè)連接,且緩存了25個(gè) channel ,注意這里的25個(gè)并不是說(shuō)，這個(gè)連接里只能最多創(chuàng)建25個(gè) channel ,而是說(shuō)最多緩存25個(gè) channel 。舉個(gè)例子，假設(shè)并發(fā)發(fā)送100條消息，在 CachingConnectionFactory.CacheMode.CHANNEL 模式下，瞬間會(huì)創(chuàng)建100個(gè) channel 的，然后往緩存里放25個(gè) channel ，當(dāng)流量下去了，剛剛創(chuàng)建的多余的 channel 會(huì)自動(dòng)關(guān)閉掉的，緩存里只保留25個(gè)。

使用這種方式的話，要注意緩存的 channel 數(shù)量，不能太小，不然流量一大,仍然會(huì)造成頻繁關(guān)閉 channel 的情況。當(dāng)然我們也不能說(shuō)有多少并發(fā)，就創(chuàng)建多少個(gè) channel ，還是要限制一下，這個(gè)時(shí)候可以使用：

connectionFactory.setChannelCheckoutTimeout(1000);

當(dāng) ChannelCheckoutTimeout 的值大于0的時(shí)候， ChannelCacheSize 的值就是最大的 channel 數(shù)量了，一旦從緩存中獲取不到 channel ，等待 ChannelCheckoutTimeout 毫秒后,如果還是獲取不到的，就會(huì)拋 AmqpTimeoutException 了。

我們也可以自己實(shí)現(xiàn) channel pool ，但是不太建議怎么做，畢竟 spring amqp 還是相當(dāng)成熟的，直接使用就可以了。