怎么進(jìn)行RabbitMQ Federation插件的分析

怎么進(jìn)行RabbitMQ Federation插件的分析，很多新手對(duì)此不是很清楚，為了幫助大家解決這個(gè)難題，下面小編將為大家詳細(xì)講解，有這方面需求的人可以來(lái)學(xué)習(xí)下，希望你能有所收獲。

在文章開(kāi)始之前，我們先介紹一下聯(lián)邦機(jī)制的基本概念。聯(lián)邦機(jī)制的實(shí)現(xiàn)，依賴于RabbitMQ的Federation插件，該插件的主要目標(biāo)是為了RabbitMQ可以在多個(gè) Broker節(jié)點(diǎn)或者集群中進(jìn)行消息的無(wú)縫傳遞。

Federation插件可以讓多個(gè)交換器和多個(gè)隊(duì)列進(jìn)行聯(lián)邦。一個(gè)聯(lián)邦交換器或者一個(gè)聯(lián)邦隊(duì)列接受上游(位于其他Broker上的交換器和隊(duì)列)消息。聯(lián)邦交換器能夠?qū)⒃景l(fā)送給上游交換器(upstream exchange)的消息路由到本地的某個(gè)隊(duì)列中;聯(lián)邦隊(duì)列則允許一個(gè)本地消費(fèi)者接收到來(lái)自上游隊(duì)列(upstream queue)的消息。

聯(lián)邦交換器

下面先假設(shè)一種場(chǎng)景，BrokerA服務(wù)部署在上海，BrokerB服務(wù)部署在北京。來(lái)自上海的ClientA向BrokerA的exchangeA發(fā)送消息網(wǎng)絡(luò)延遲很小，但是北京的ClientB向BrokerA的exchangeA發(fā)送消息那么將會(huì)面臨網(wǎng)絡(luò)延遲的問(wèn)題。Federation機(jī)制則可以幫助我們解決這個(gè)問(wèn)題。

首先在BrokerA的exchangeA上與北京的BrokerB建立一條單向的Federation Link。此時(shí)Federation插件會(huì)在BrokerB上建立一個(gè)同名的交換器(可以配置，默認(rèn)同名)，并且還會(huì)建立一個(gè)內(nèi)部交換器federation:exchangeA->Broker B(其中Broker為集群名稱)通過(guò)相同的綁定建進(jìn)行綁定，于此同時(shí)Federation插件會(huì)建立一個(gè)federation:exchangeA->Broker B(BrokerB為集群名稱)，并且將內(nèi)部交換器federation:exchangeA->Broker B綁定到該隊(duì)列。

Federation插件會(huì)在隊(duì)列federation:exchangeA->Broker B與BrokerA中的交換器exchangeA之間建立一條AMQP連接來(lái)實(shí)時(shí)地消費(fèi)隊(duì)列federation:exchangeA->Broker B中的數(shù)據(jù)。這些操作都是內(nèi)部的，對(duì)外部業(yè)務(wù)客戶端來(lái)說(shuō)這條Federation link建立在BrokerA的exchangeA和BrokerB的exchangeA之間。

此時(shí)ClientB可以以較小的網(wǎng)絡(luò)延遲向BrokerB的exchangeA發(fā)送消息，并且該消息會(huì)被正確路由到BrokerA中的exchangeA中，通過(guò)Federation插件我們可以以較小的網(wǎng)絡(luò)延遲向與客戶端屬于不同地域的Broker節(jié)點(diǎn)發(fā)送消息。

"max_hops=1"表示一條消息最多被轉(zhuǎn)發(fā)的次數(shù)為1。

默認(rèn)的交換器(每個(gè)vhost下都會(huì)默認(rèn)創(chuàng)建一個(gè)名為""的交換器)和內(nèi)部交換器，不能對(duì)其使用Federation的功能。

聯(lián)邦隊(duì)列

隊(duì)列queue1和queue2原本在broker2中，由于某種需求將其配置為federated queue并將broker1作為upstream。Federation插件會(huì)在broker1上創(chuàng)建同名的隊(duì)列queue1和queue2，與broker2中的隊(duì)列queue1和queue2分別建立兩條單向獨(dú)立的Federation link。當(dāng)有消費(fèi)者ClientA連接broker2并通過(guò)Basic.Consume消費(fèi)隊(duì)列queue1(或queue2)中的消息時(shí)，如果隊(duì)列queue1(或queue2)中本身有若干消息堆積，那么ClientA直接消費(fèi)這些消息，此時(shí)broker2中的queue1(或queue2)并不會(huì)拉取broker1中的queue1(或queue2)的消息；如果隊(duì)列queue1(或queue2)中沒(méi)有消息堆積或者消息被消費(fèi)完了，那么它會(huì)通過(guò)Federation link拉取在broker1中的上游隊(duì)列queue1(或queue2)中的消息(如果有消息)，然后存儲(chǔ)到本地，之后再被消費(fèi)者ClientA進(jìn)行消費(fèi)。

和federated exchange不同，一條消息可以在聯(lián)邦隊(duì)列間轉(zhuǎn)發(fā)無(wú)限次。兩個(gè)隊(duì)列可以互為聯(lián)邦隊(duì)列。

如果兩個(gè)隊(duì)列互為聯(lián)邦隊(duì)列，隊(duì)列中的消息除了被消費(fèi)，還會(huì)轉(zhuǎn)向有多余消費(fèi)能力的一方，如果這種"多余的消費(fèi)能力"在broker1和broker2中來(lái)回切換，那么消費(fèi)也會(huì)在broker1和broker2中的隊(duì)列queue中來(lái)回轉(zhuǎn)發(fā)

federation queue只能使用Basic.Consume進(jìn)行消費(fèi)，并且不具備傳遞性。

Federation使用

Federation插件默認(rèn)在RabbitMQ發(fā)布包中，執(zhí)行rabbitmq-plugins enable rabbitmq_federation命令可以開(kāi)啟Federation功能。rabbitmq-plugins enable rabbitmq_federation命令會(huì)同時(shí)開(kāi)啟amqp_client插件。如果要開(kāi)啟Federation的管理插件，需要執(zhí)行rabbitmq-plugins enable rabbitmqfederation_management命令。

當(dāng)需要在集群中使用Federation功能的時(shí)候，集群中所有的節(jié)點(diǎn)都應(yīng)該開(kāi)啟Federation插件。

在前面的章節(jié)中我們搭建了一個(gè)3個(gè)節(jié)點(diǎn)的rabbitmq集群(假設(shè)在上海)，我們?cè)诖罱ㄒ粋€(gè)兩個(gè)節(jié)點(diǎn)的集群(假設(shè)在北京)，這個(gè)讀者自行去搭建吧。在我們第一個(gè)集群中默認(rèn)的vhost下有一個(gè)exchangeA和queueA，我們北京的一個(gè)客戶端需要向exchangeA中發(fā)消息，為了減輕網(wǎng)絡(luò)延遲，我們使用聯(lián)邦機(jī)制，將北京的集群作為上行資源，下面我們講一下配置。

首先在上海集群中配置Federation Upstreams，配置如下圖：

• name：定義這個(gè)upstream的名稱，必填項(xiàng)

• uri：定義upstream的AMQP連接，我們這里設(shè)置的是amqp://upstreamer:123456@rabbit111:5672(這里是一個(gè)北京集群一個(gè)節(jié)點(diǎn)地址)，格式amqp://username:password@host:port

• Prefetch count：定義Federation內(nèi)部緩存的消息條數(shù)，即在收到上游消息之后且在發(fā)送到下游之前緩存的消息條數(shù)。

• Reconnect delay：Federation link由于某種原因斷開(kāi)之后，需要等待多少秒開(kāi)始重新建立連接。

• Acknowledgement Mode：定義Federation link 的消息確認(rèn)方式。其有3種：on-confirm、on-publish、no-ack。默認(rèn)為on-confirm，表示在接收到下游的確認(rèn)消息(等待下游的Basic.Ack)之后再向上游發(fā)送消息確認(rèn)，這個(gè)選項(xiàng)可以確保網(wǎng)絡(luò)失敗或者Broker宕機(jī)時(shí)不會(huì)丟失消息，但也是處理速度最慢的選項(xiàng)。如果設(shè)置為on-publish，則表示消息發(fā)送到下游后(并需要等待下游的Basic.Ack)再向上游發(fā)送消息確認(rèn)，這個(gè)選項(xiàng)可以確保在網(wǎng)絡(luò)失敗的情況下不會(huì)丟失消息，但不能確保Broker宕機(jī)時(shí)不會(huì)丟失消息。no-ack表示無(wú)須進(jìn)行消息確認(rèn)，這個(gè)選項(xiàng)處理速度最快，但也最容易丟失消息。

• Trust User-ID：設(shè)定Federation是否使用"Validated User-ID" 這個(gè)功能。如果設(shè)置為false或者沒(méi)有設(shè)置，那么Federation會(huì)忽略消息的user_id 這個(gè)屬性；如果設(shè)置為true，則Federation只會(huì)轉(zhuǎn)發(fā)user_id為上游任意有效的用戶的消息。

• Exchange：指定upstream exchange的名稱，默認(rèn)情況下和federated exchange同名，這里我們指定的是beijingExchange

• Max hops：指定消息被丟棄前在Federation link中最大的跳轉(zhuǎn)次數(shù)。默認(rèn)為1。注意即使設(shè)置max-hops參數(shù)為大于1的值，同一條消息也不會(huì)在同一個(gè)Broker中出現(xiàn)2次，但是有可能會(huì)在多個(gè)節(jié)點(diǎn)中被復(fù)制。

• Expires：指定Federation link斷開(kāi)之后，federated queue所對(duì)應(yīng)的upstream queue的超時(shí)時(shí)間，默認(rèn)為"none"，表示為不刪除，單位為ms。這個(gè)參數(shù)相當(dāng)于設(shè)置普通隊(duì)列的x-expires參數(shù)。設(shè)置這個(gè)值可以避免Federation link斷開(kāi)之后，生產(chǎn)者一直在向北京集群中的upstream exchange發(fā)送消息，這些消息又不能被轉(zhuǎn)發(fā)到上海的集群中而被消費(fèi)掉，進(jìn)而造成upstream exchange中有大量的消息堆積。

• Message TTL：為federated queue所對(duì)應(yīng)的upstream queue設(shè)置，相當(dāng)于普通隊(duì)列的x-message-ttl參數(shù)。默認(rèn)為"none"，表示消息沒(méi)有超時(shí)時(shí)間。

• HA policy：為federated queue所對(duì)應(yīng)的upstream queue設(shè)置，相當(dāng)于普通隊(duì)列的x-ha-policy參數(shù)，默認(rèn)為 "none"，表示隊(duì)列沒(méi)有任何HA。

• Queue：執(zhí)行upstream queue的名稱，默認(rèn)情況下和federated queue同名

在配置完了Federation Upstreams之后，再定義一個(gè)Policy用于匹配交換器exchangeA，并且使用剛才的Federation Upstreams配置，如下圖：當(dāng)策略配置完畢之后，F(xiàn)ederation將會(huì)在北京的集群上建立一個(gè)beijingExchange，并建立Federation link，如下圖：

看完上述內(nèi)容是否對(duì)您有幫助呢？如果還想對(duì)相關(guān)知識(shí)有進(jìn)一步的了解或閱讀更多相關(guān)文章，請(qǐng)關(guān)注億速云行業(yè)資訊頻道，感謝您對(duì)億速云的支持。