RabbitMQ在Openstack中的使用

1. AMQP協(xié)議

??RabbitMQ是Advanced Message Queuing Protocol （AMQP，高級消息隊(duì)列協(xié)議）開放標(biāo)準(zhǔn)的實(shí)現(xiàn)，它支持符合標(biāo)準(zhǔn)的客戶端請求程序與符合標(biāo)準(zhǔn)的消息中間件代理進(jìn)行通信。
??AMQP中的核心概念：
??(1) Broker ：消息中間件的服務(wù)節(jié)點(diǎn)，對于RabbitMQ來說，一個RabbitMQ Broker可以簡單地看作一個RabbitMQ服務(wù)節(jié)點(diǎn)，或者RabbitMQ服務(wù)實(shí)例;
??(2) Virtual Host: 虛擬主機(jī)，表示一批交換器、消息隊(duì)列和相關(guān)對象。虛擬主機(jī)是共享相同的身份認(rèn)證和加密環(huán)境的獨(dú)立服務(wù)器域。每個 vhost 本質(zhì)上就是一個 mini 版的 RabbitMQ 服務(wù)器，擁有自己的隊(duì)列、交換器、綁定和權(quán)限機(jī)制；
??(3) Producer：生產(chǎn)者，消息投遞方，生產(chǎn)者創(chuàng)建消息，然后發(fā)布到RabbitMQ中;
??(4) Consumer ：消費(fèi)者，就是接收消息的一方，消費(fèi)者連接到RabbitMQ服務(wù)器，并訂閱到隊(duì)列上;
??(5) Queue ：隊(duì)列，RabbitMQ的內(nèi)部對象，用于存儲消息，RabbitMQ的生產(chǎn)者生產(chǎn)消息并最終投遞到隊(duì)列中，消費(fèi)者可以從隊(duì)列中獲取消息并消費(fèi);
??(6) Exchange:交換器，生產(chǎn)者將消息發(fā)送到Exchange，由交換器將消息路由到一個或者多個隊(duì)列中;
??(7) RoutingKey :路由鍵，生產(chǎn)者將消息發(fā)給交換器的時候，一般會指定一個RoutingKey，用來指定這個消息的路由規(guī)則，生產(chǎn)者可以在發(fā)送消息給交換器時，通過指定RoutingKey來決定消息流向哪里。
??AMQP模型：

圖1-1 AMQP模型

??AMQP協(xié)議消息發(fā)送接收流程：
??(1) 在Producer客戶端建立了Channel后，就建立了到Broker上Virtual Host的連接。接下來Producer就可以向這個Virtual Host中的Exchange發(fā)送消息了；
??(2) Exchange能夠處理消息的前提是：它至少已經(jīng)和某個queue或者另外的Exchange形成了綁定關(guān)系，并設(shè)置好了到這些queue和Excahnge的Routing（路由規(guī)則）。Excahnge中的Routing有四種模式。在Exchange收到消息后，會根據(jù)設(shè)置的Routing（路由規(guī)則），將消息發(fā)送到符合要求的queue或者Exchange中（路由規(guī)則還會和RoutingKey屬性配合使用）；
??Excahnge中的Routing的四種模式：
??1)fanout路由模式，fanout路由模式不需要routingKey。當(dāng)設(shè)置為fanout模式的Exchange收到消息后，然后復(fù)制多份，分別發(fā)送到和自己綁定的各個queue中，相當(dāng)于廣播模式。
??2)direct路由模式：

圖1-2 direct路由模式

??以上圖的配置為例，以routingKey=”aa”發(fā)送消息到Exchange，則消息會路由到queueX和queueY，另外路由鍵與隊(duì)列名是完全匹配的，如果一個隊(duì)列綁定到交換機(jī)要求路由鍵為“aa”，不會轉(zhuǎn)發(fā)“aa.bb”，也不會轉(zhuǎn)發(fā)“aa.cc”，它是完全匹配、單播的模式；如果以routingKey=”bb”，則消息只會路由到queueY。
??3)topic路由模式

圖1-3 topic路由模式

??routingKey中可以存在兩種特殊字符“”與“#”，用于做模糊匹配，其中“”用于匹配一個單詞，“#”用于匹配多個單詞（可以是零個），用“.”隔開。以上圖中的配置為例，routingKey=”aa.bb.cc”的消息會同時路由到queueX與queueY，routingKey=”aa.cc.dd”的消息會路由到queueX，routingKey=”cc.dd.ff”的消息會路由到queueY，routingKey=”cc.bb.ee”的消息會路由到queueY（雖然與兩個routingKey都匹配，但只會投遞給一次）。
??4)headers，headers類型的Exchange不依賴于routing key與binding key的匹配規(guī)則來路由消息，而是根據(jù)發(fā)送的消息內(nèi)容中的headers屬性進(jìn)行匹配。(一般不用)
??(3) queue收到消息后，可能會進(jìn)行如下的處理：如果當(dāng)前沒有Consumer的Channel連接到這個queue，那么queue將會把這條消息進(jìn)行存儲直到有Channel被創(chuàng)建；如果已經(jīng)有Channel連接到這個queue，那么消息將會按順序被發(fā)送給這個Channel。
??(4) Consumer收到消息后，就可以進(jìn)行消息的處理了。

2. RabbitMQ實(shí)現(xiàn)RPC請求

??RabbitMQ的RPC的處理流程圖：

圖2-1 RabbitMQ的RPC的處理流程

??RabbitMQ的RPC的處理流程步驟：
??(1) 當(dāng)客戶端啟動時，創(chuàng)建一個匿名的回調(diào)隊(duì)列。
??(2) 客戶端為RPC請求設(shè)置2個屬性：replyTo，設(shè)置回調(diào)隊(duì)列名字；correlationId，標(biāo)記request。
??(3) 請求被發(fā)送到rpc_queue隊(duì)列中。
??(4) RPC服務(wù)器端監(jiān)聽rpc_queue隊(duì)列中的請求，當(dāng)請求到來時，服務(wù)器端會處理并且把帶有結(jié)果的消息發(fā)送給客戶端。接收的隊(duì)列就是replyTo設(shè)定的回調(diào)隊(duì)列。
??(5) 客戶端監(jiān)聽回調(diào)隊(duì)列，當(dāng)有消息時，檢查correlationId屬性，如果與request中匹配，就是結(jié)果了。

3. Oslo_message對RPC的封裝

??而Oslo.messaging庫為OpenStack各個組件使用RPC和事件通知（Event Notification)提供了一套統(tǒng)一的接口，它基于支持 AMQP(Advanced Message Queuing Protocol) 協(xié)議的為同一個項(xiàng)目內(nèi)的各個進(jìn)程之間的通信提供了 API，如 nova-api 和 nova-scheduler 的通信，cinder-api 和 cinder-volume 的通信等。
??下面通過實(shí)例看一下Oslo.messaging庫使用RPC提供的接口。首先，了解一下Oslo.messaging的Transport和Target，Transport(傳輸層)主要實(shí)現(xiàn)RPC底層的通信以及事件循環(huán)、多線程等其他功能，通過URL來獲得指向不同Transport實(shí)現(xiàn)的句柄。URL格式：Transport://user:pass@host1:port[,hostN:portN]/virtual_host，可通過oslo.messaging.get_transport函數(shù)來獲得transport對象實(shí)例的句柄；Target封裝了指定某一個消息最終目的地的所有信息,下表所示為其所具有的屬性:

??在不同的應(yīng)用場景下，構(gòu)造Target對象需要不同的參數(shù):創(chuàng)建一個RPC服務(wù)器時，需要topic和server參數(shù)，exchange參數(shù)可選; 指定一個endpoint時，namespace和version是可選的;客戶端發(fā)送消息時，需要topic參數(shù)，其他可選。

class ServerControlEndpoint(object):
    target = messaging.Target(namespace='controle', version='2.0')
    
    def __init__(self, server):
        self.server = server

    def stop(self, ctx):
        if self.server:
            self.server.stop()
            
            
class TestEndpoint(object):
    def test(self, ctx, arg):
        return arg
    

transport_url = 'rabbit://user:pass@host1:port/virtual_host'
transport = messaging.get_transport(cfg.CONF, transport_url)
target = messaging.Target(topic='test', server='server1')
endpoints = [
    ServerControlEndpoint(None),
    TestEndpoint(),
]

server = messaging.get_rpc_server(transport, target, endpoints, executor='blocking')

??從上面代碼可以看出，創(chuàng)建rpc server對象之前，需要先創(chuàng)建transport和target對象，使用get_transport()函數(shù)來獲得transport對象的句柄，此處構(gòu)建的Target對象是用來建立RPC Server的,所以需指明topic和server參數(shù)，使用get_rpc_server()函數(shù)創(chuàng)建server對象,然后調(diào)用server對象的start方法開始接收遠(yuǎn)程調(diào)用，對于endpoint，一個RPC服務(wù)器可以暴露多個endpoint(本實(shí)例有兩個)，每個endpoint包含一組方法，這組方法是可以被客戶端通過某種Transport對象遠(yuǎn)程調(diào)用的，executor是消息構(gòu)造對象，有blocking的阻塞模式和eventlet采用協(xié)程兩種方式，OpenStack采用eventlet。

transport_url = 'rabbit://user:pass@host1:port/virtual_host'
transport = messaging.get_transport(cfg.CONF, transport_url)
target = messaging.Target(topic='test')
client = messaging.RPCClient(transport, target)
client.call(ctxt, 'test', arg=arg)
cctxt = client.prepare(namespace='control', version ='2.0')

??上述代碼是通過RPC Client，遠(yuǎn)程調(diào)用RPC Sever上的方法，調(diào)用方式有cast和call兩種遠(yuǎn)程調(diào)用方式。通過cast方式遠(yuǎn)程調(diào)用，請求發(fā)送后就直接返回了；通過call方式調(diào)用， 需要等響應(yīng)從服務(wù)器返回。

4. RabbitMQ在Openstack應(yīng)用

??下面以創(chuàng)建虛擬機(jī)為例分析一下消息流程：

圖4-1 創(chuàng)建虛擬機(jī)為例分析一下消息流程

??從上圖能夠看出，以nova-api和nova-conductor之間的通信為例，nova-conductor服務(wù)在啟動時會注冊一個RPC server等待處理請求，nova-api發(fā)送創(chuàng)建虛擬機(jī)的rpc請求時會先創(chuàng)建一個topic publisher用于topic發(fā)布，method為build_instance，然后publisher將消息發(fā)送給exchange，exchange再根據(jù)routingkey轉(zhuǎn)發(fā)給綁定的queue中，最后由topic consumer接收并調(diào)用nova-conductor manager中的build_instance方法處理，對于nova-conductor和nova-scheduler之間的通信，多了一步把目標(biāo)主機(jī)作為返回結(jié)果信息返回到reply_xx隊(duì)列中，然后由nova-conductor接收以后向nova-compute發(fā)起rpc.cast的創(chuàng)建請求。
??OpenStack各個組件內(nèi)部的各個服務(wù)進(jìn)程之間則是通過基于AMPQ的RPC方式進(jìn)行通信，實(shí)現(xiàn)RPC通信需借助Rabbitmq消息隊(duì)列，RPC方式又分為兩種，rpc.cast和rpc.call，rpc.call為request/response方式，多用于同步場景；而使用rpc.cast方式發(fā)出請求后則無需一直等待響應(yīng)，但之后需要定期查詢執(zhí)行結(jié)果，一般用于異步場景，OpenStack將其使用的通信方式都封裝在公有庫oslo_messaging中。

5. RabbitMQ在Openstack中應(yīng)用常見問題之連接數(shù)

??RabbitMQ每增加一個連接，erlang都會給這個連接分配三個erlang進(jìn)程，每個進(jìn)程都會分配一定大小內(nèi)存空間，所以隨著連接數(shù)的增長，內(nèi)存和erlang進(jìn)程數(shù)呈現(xiàn)有規(guī)律的增長，所以RabbitMQ連接數(shù)的無限增大會壓垮mq服務(wù)，導(dǎo)致RabbitMQ服務(wù)崩潰。
??(1) Openstack連接數(shù)增長原因
客戶端與RabbitMQ建立的是長連接，而不是建立短連接，因?yàn)槿绻l繁的建立、銷毀connection，會增加額外的時間開銷，當(dāng)業(yè)務(wù)量比較大時，就會對系統(tǒng)性能產(chǎn)生比較大的影響。OpenStack組件與RabbitMQ的連接使用到了第三方庫oslo_message中的connection pool的概念，在不超過pool size的前提上，當(dāng)有并發(fā)業(yè)務(wù)的時候，如果發(fā)現(xiàn)pool中已有connection正被使用，那么就會在pool中繼續(xù)創(chuàng)建新的connection，直到創(chuàng)建的connection數(shù)量達(dá)到pool的最大值，之后如果再有業(yè)務(wù)需要，會等待之前創(chuàng)建的connection被重新放入connection pool，然后等待被繼續(xù)使用。這種情況下，就會出現(xiàn)connection一直增長的現(xiàn)象。
??(2) Openstack最大連接數(shù)計(jì)算方法：
OpenStack的Connection連接數(shù)包括m臺計(jì)算節(jié)點(diǎn)+n臺控制節(jié)點(diǎn)兩部分之和:

Connection連接數(shù)1（m臺計(jì)算節(jié)點(diǎn)） =    （64+30）* m；
（其中，“m”指的是m臺的計(jì)算節(jié)點(diǎn)，“64”指的是每個Nova-Compute進(jìn)程中配置的
最大rpc_conn_pool_size連接數(shù)上限，“30”指的是每個Ovs-agent進(jìn)程中配置的最
大rpc_conn_pool_size連接數(shù)上限）;  

Connection連接數(shù)2（n臺控制節(jié)點(diǎn)）=     
（nova-api worker進(jìn)程數(shù) + nova-conductor worker進(jìn)程數(shù) ）* n * 30 
+ （cinder-api worker進(jìn)程數(shù) + 1 + 1+backend進(jìn)程數(shù)）* n * 30
+ （neutron worker進(jìn)程數(shù) + 1 + 1）* 3 * 30；
（其中，“n”指的是n臺控制節(jié)點(diǎn)，“30”指的是“nova-api”、“nova-conductor”和“neutron”
三個進(jìn)程中配置的最大rpc_conn_pool_size連接數(shù)上限）;  

??綜合上面兩部分，OpenStack的最大Connection連接數(shù) 等于：Connection連接數(shù)1（m臺計(jì)算節(jié)點(diǎn)）與 Connection連接數(shù)2（n臺控制節(jié)點(diǎn)）之和。此計(jì)算值必須小于RabbitMQ能夠承載的連接數(shù)最大值，才能保證RabbitMQ不被壓垮，正常提供服務(wù)。對此，對RabbitMQ集群（三個節(jié)點(diǎn)）做壓力測試能夠得到RabbitMQ集群能夠承載的連接數(shù)最大值。
??使用的測試環(huán)境配置如下：

??測試目的及用例：
利用壓力測試工具PerfTest逐步增加集群的三個節(jié)點(diǎn)的連接數(shù)，觀察記錄各節(jié)點(diǎn)的內(nèi)存、連接數(shù)、日志輸出，測試并記錄“批量創(chuàng)建虛擬機(jī)”和“批量刪除虛擬機(jī)”兩個典型的Openstack業(yè)務(wù)場景所需時間。找到當(dāng)虛擬機(jī)的創(chuàng)建/刪除操作受到影響（創(chuàng)建/刪除十分耗時或者不能成功）或者RabbitMQ集群出現(xiàn)異常Error的日志時，從而得到RabbitMQ集群所承受的連接數(shù)即為集群所能承受的最大連接數(shù)。
對三個節(jié)點(diǎn)，通過壓力測試工具對每個節(jié)點(diǎn)增加2000個連接，等待5分鐘（等待集群運(yùn)行穩(wěn)定），記錄每個節(jié)點(diǎn)連接數(shù)、內(nèi)存、erlang進(jìn)程數(shù)、消息積壓情況。然后通過腳本批量創(chuàng)建60臺虛擬機(jī)，記錄全部創(chuàng)建完成所需時間、全部刪除完成所需時間，記錄nova-compute的狀態(tài)、RabbitMQ日志有無異常輸出。重復(fù)上述步驟，每次各增加2000個連接，直到發(fā)現(xiàn)虛擬機(jī)的創(chuàng)建或刪除耗時超過30分鐘，或者虛擬機(jī)無法成功創(chuàng)建或刪除，則終止測試。然后觀察RabbitMQ集群情況和nova-compute服務(wù)30分鐘，看是否有RabbitMQ異常日志打印和nova-compute由up變?yōu)閐own。
??測試結(jié)果：
??各個節(jié)點(diǎn)狀態(tài)結(jié)果：

??虛機(jī)創(chuàng)建和節(jié)點(diǎn)狀態(tài)相關(guān)結(jié)果：

??各個節(jié)點(diǎn)及集群統(tǒng)計(jì)圖的趨勢變化：

圖5-1 node1測試結(jié)果

圖5-2 node2測試結(jié)果

圖5-3 node3測試結(jié)果

圖5-4 集群測試結(jié)果

6. 小結(jié)

??本篇文章對AMQP協(xié)議的基本概念以及框架流程做了詳細(xì)的介紹，RabbitMQ使用此協(xié)議能夠很好的實(shí)現(xiàn)RPC請求，Oslo_message對RPC進(jìn)行的封裝能夠使用起來更加優(yōu)雅，然后以創(chuàng)建虛擬機(jī)為例分析了RabbitMQ在Openstack中如何應(yīng)用，針對諸多的RabbitMQ應(yīng)用問題，選取了連接數(shù)問題進(jìn)行了測試并得出結(jié)論，后續(xù)會針對RabbitMQ應(yīng)用的其他問題做出更深入的研究。