RabbitMQ實踐體驗

最近由于業(yè)務(wù)需要進行性能升級，將原來需要經(jīng)過http進行數(shù)據(jù)交互的方式修改為消息隊列的形式。于是原來的同步處理的方式變成了異步處理，在一定程度上提升我們系統(tǒng)的性能，不過debug的時候，不免哭了出來。因為每個環(huán)節(jié)都需要進行詳細檢查。
對于RabbitMQ，我們知道，其是AMQP的一種代理服服務(wù)器，具有一套嚴格的通信方式，即在核心產(chǎn)品進行通信的各個方面幾乎都采用了RPC(Remote Procedure Call, 遠程過程調(diào)用)模式。

AMQ與RabbitMQ進行交互

RabbitMQ通信時用到的類和方法與AMQP協(xié)議層面的類和方法一一對應(yīng)。因此AMQP本質(zhì)上是RPC的一種傳輸機制

高級消息隊列模型

AMQ(Advanced Message Queuing)模型，這個模型是針對代理服務(wù)器軟件例如(RabbitMQ)設(shè)計的，該模型在邏輯上定義了三種抽象組件用于指定消息的路由行為，分別是：

• 交換器Exchange，消息代理服務(wù)器中用于把消息路由到隊列的組件
  接收/發(fā)送到RabbitMQ中的消息并決定把他們投遞到何處。
  定義消息的路由行為，通常這需要檢查消息所攜帶的數(shù)據(jù)特性或者包含在消息體內(nèi)的各種屬性
• 隊列Queue，用來存儲消息的數(shù)據(jù)結(jié)構(gòu)，位于硬盤或內(nèi)存中，以FIFO的順序進行投遞
  負責存儲接收到的消息，同時也可能包含何如處理消息的配置信息。
• 綁定Binding，一套規(guī)則，用于告訴交換器消息應(yīng)該被存儲到哪個隊列
  • 定義隊列和交換機之間的關(guān)系
  • 告知一個交換器應(yīng)該將消息投遞到哪些隊列中。對于某些交換器類型，綁定同時告知交換器如何對消息進行過濾從而決定能夠投遞到隊列的消息
  • 當發(fā)布一條消息到交換器時，應(yīng)用程序使用路由鍵routing-key屬性。路由可以是隊列名稱，也可以是一串用于描述消息、具有特定語法的字符串。當交換器對一條消息進行評估以決定路由到哪些合適的隊列時，消息的路由就會和綁定進行比對。
  • 綁定是綁定隊列到交換器的粘合劑，而路由則是用于比對的標準。 RabbitMQ的靈活性來自于消息如何通過交換器路由到隊列的動態(tài)特性，介于交換器和隊列之間的綁定，以及他們所創(chuàng)建的動態(tài)消息路由，構(gòu)成了消息通信架構(gòu)的基本組件。為了把消息路由到合適的目標地址，RabbitMQ所需的第一種信息就是用于控制路由的交換

python使用AMQP

在將消息發(fā)布到隊列之前，我們需要經(jīng)歷過以下若干個步驟。至少，必須要設(shè)置交換器和隊列，然后將他們綁定再一起。接下來我們將通過python來實現(xiàn)AMQP機制。
我用到了pika這個庫，需要的話，需要通過以下指令安裝。該庫實現(xiàn)了絕大部分rabbitmq的api以及提供了相關(guān)的調(diào)優(yōu)參數(shù)，后續(xù)有機會不妨可以詳談。

pip install pika

1. 聲明交換器

交換器在AMQ模型中是非常重要的角色存在。因此，在AMQP規(guī)范中都有自己的類。聲明一個交換器，我們可以直接在控制臺界面進行創(chuàng)建。

不過這樣僅僅是在極少數(shù)的情況下才適合，動手調(diào)戲鼠標對開發(fā)工程師的來說實在是太蠢啦，能玩鍵盤就別玩鼠標啊，我們不妨通過以下代碼來聲明（創(chuàng)建）一個交換器。pika內(nèi)置函數(shù)會事先通過get的方式來檢查我們待聲明的交換器是否存在，如果存在則不創(chuàng)建，否則創(chuàng)建一個新的交換器。

 self.channel.exchange_declare(
            exchange=exchange,
            exchange_type="direct",
            passive=False,
            durable=True,
            auto_delete=False)

2. 聲明隊列

一旦交換器創(chuàng)建成功，就可以通過發(fā)送類似queue.declare命令讓rabbitmq創(chuàng)建一個隊列。同樣的，我們?nèi)匀豢梢栽趫D形化界面里面創(chuàng)建隊列。

還是那句話，動手調(diào)戲鼠標對開發(fā)工程師的來說實在是太蠢啦，能玩鍵盤就別玩鼠標啊，我們不妨通過以下代碼來聲明（創(chuàng)建）若干個隊列。pika內(nèi)置函數(shù)會事先通過get的方式來檢查我們待聲明的隊列是否存在，如果存在則不創(chuàng)建，否則創(chuàng)建一個新的隊列。

self.channel.queue_declare(queue=queue, durable=True)

當隊列同名時，即如果我們多次發(fā)送同一個queue.declare命令并不會有任何副作用，因為RabbitMQ并不會處理后續(xù)的隊列聲明，究其原因，每次創(chuàng)建都會先通過get的方式調(diào)用消息隊列引擎查詢隊列是否存在。如果需要返回隊列相關(guān)的有用信息，則將會返回隊列中待處理消息的數(shù)量以及該隊列的消費者數(shù)量。當然了如果隊列同名，而且新隊列的屬性與原有的隊列不一樣，那么RabbitMQ將關(guān)閉發(fā)出的RPC請求的信道，返回403錯誤

3. 綁定隊列到交換器

一旦創(chuàng)建了交換器和隊列，之后就可以將它們綁定在一起了，如同queue.declare命令，將隊列綁定到交換器Queue.Bind每次只能指定一個隊列。我們既可以通過圖形化界面進行綁定，也可以通過代碼實現(xiàn)這個效果

 self.channel.queue_bind(
            queue=queue, exchange=exchange, routing_key=rk)

4. 發(fā)布消息

發(fā)布消息到RabbitMQ時，多個幀封裝了發(fā)送到服務(wù)器的消息數(shù)據(jù)。在實際的消息內(nèi)容到達rabbitMQ之前，客戶端應(yīng)用程序會發(fā)送一個basic.publish方法幀、一個內(nèi)容頭幀和至少一個消息體幀。
默認情況下，只要沒有消費者正在監(jiān)聽隊列，消息就會被存儲在隊列中。當添加更多消息時，隊列大小也會隨之增加。RabbitMQ可以將這些消息保存在內(nèi)存或者寫入磁盤。

def produce(self, body):
        self.channel.basic_publish(exchange=self.exchange, routing_key=self.route_key, body=body,
                                   properties=pika.BasicProperties(content_type='text/plain', delivery_mode=1)
                                   )

5. 消費消息

一旦發(fā)布消息被路由并且保存在一個或者多個隊列中，剩下的就是如何對其進行消費。注意到，發(fā)送和消費是異步的。 消費時，可以讓RabbitMQ知道如何消費他們
Basic.Consume命令中
no_ack為true時，RabbitMQ將連續(xù)發(fā)送消息直到消費者發(fā)送一個Basic.Cancel命令或者斷開連接為止
如果為false，則需要發(fā)送一個Basic.Ack來確認收到每條消息的請求

def on_message(chan, method_frame, _header_frame, body, userdata=None):
            """Called when a message is received. Log message and ack it."""
            # LOGGER.info('Userdata: %s Message body: %s', userdata, body)
            # print(" [x] Received %r" % body.decode())
            data = body.decode()
            result = alarmFun(data)
            publish = Publish(exchange='spider', queue='alarm', rk='rk-alarm')
            publish.produce(result)
            # chan.basic_ack(delivery_tag=method_frame.delivery_tag)

on_message_callback = functools.partial(on_message)
self.channel.basic_consume(on_message_callback=on_message_callback,
                                   queue=self.queue,
                                   auto_ack=True
                                   )

基于python開發(fā)

經(jīng)過前面的描述，我們需要理論聯(lián)系實踐，讓我們通過python開發(fā)消費者角色和發(fā)布者角色。

發(fā)布者

按照配置流程，我們需要初始化連接、配置交換器、隊列、綁定，然后才能通過連接件信息推送（publish）到隊列中。

import logging
from random import randint

import pika

BROKER_USER = os.environ.get('BROKER_USER', 'guest')
BROKER_PASSWD = os.environ.get('BROKER_PASSWD', 'guest')
BROKER_IP = os.environ.get('BROKER_IP', '127.0.0.1')
BROKER_PORT = os.environ.get('BROKER_PORT', '5672')
BROKER_VHOST = os.environ.get('BROKER_VHOST', 'my_vhost')
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'
BROKER_URL = 'amqp://{}:{}@{}:{}/{}'.format(BROKER_USER, BROKER_PASSWD, BROKER_IP, BROKER_PORT, BROKER_VHOST)

# logging.basicConfig(level=logging.DEBUG)
# LOG_FORMAT = ('%(levelname) -10s %(asctime)s %(name) -30s %(funcName) '
#               '-35s %(lineno) -5d: %(message)s')
# LOGGER = logging.getLogger(__name__)

class Publish(object):
    def __init__(self, exchange, queue, rk):
        # LOGGER.info('Connecting to %s', BROKER_URL)
        # logging.basicConfig(level=logging.DEBUG)
        self.credentials = pika.PlainCredentials(BROKER_USER, BROKER_PASSWD)
        # 通過這個方式設(shè)置備用鏈路，保證connection穩(wěn)定性
        self.parameters = (
            pika.ConnectionParameters(BROKER_IP, BROKER_PORT, BROKER_VHOST, self.credentials),
            pika.ConnectionParameters(BROKER_IP, BROKER_PORT, BROKER_VHOST, self.credentials, connection_attempts=5,
                                      retry_delay=1))
        self.connection = pika.BlockingConnection(self.parameters)
        self.channel = self.connection.channel()
        self.exchange = exchange
        self.channel.exchange_declare(
            exchange=exchange,
            exchange_type="direct",
            passive=False,
            durable=True,
            auto_delete=False)
        self.channel.queue_declare(queue=queue, durable=True)
        self.route_key = rk

    def produce(self, body):
        self.channel.basic_publish(exchange=self.exchange, routing_key=self.route_key, body=body,
                                   properties=pika.BasicProperties(content_type='text/plain', delivery_mode=1)
                                   )

    def close(self):
        self.connection.close()

def test():
    publish = Publish(exchange='test_yerik', queue='test_test', rk='rk-test_test')
    for i in range(1, 10000):
        publish.produce(randint(1, 100).__str__())
    publish.close()

if __name__ == '__main__':
    test()

消費者

消費者的設(shè)計和生產(chǎn)者在初始化的時候設(shè)計大致相同，都是通過建立連接、開啟channel、exange、queue、bind等過程，主要的區(qū)別在于commsum

import functools
import logging
import pika

BROKER_USER = os.environ.get('BROKER_USER', 'guest')
BROKER_PASSWD = os.environ.get('BROKER_PASSWD', 'guest')
BROKER_IP = os.environ.get('BROKER_IP', '127.0.0.1')
BROKER_PORT = os.environ.get('BROKER_PORT', '5672')
BROKER_VHOST = os.environ.get('BROKER_VHOST', 'my_vhost')
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'
BROKER_URL = 'amqp://{}:{}@{}:{}/{}'.format(BROKER_USER, BROKER_PASSWD, BROKER_IP, BROKER_PORT, BROKER_VHOST)

# print('pika version: %s' % pika.__version__)

# logging.basicConfig(level=logging.DEBUG)
# LOG_FORMAT = ('%(levelname) -10s %(asctime)s %(name) -30s %(funcName) '
#               '-35s %(lineno) -5d: %(message)s')
# LOGGER = logging.getLogger(__name__)
from apps.alarm.alarmfun import alarmFun
from apps.utils.rabbitmq.publish import Publish

class Consummer(object):
    def __init__(self, exchange, queue, rk):
        # LOGGER.info('Connecting to %s', BROKER_URL)
        self.credentials = pika.PlainCredentials(BROKER_USER, BROKER_PASSWD)
        self.parameters = (
            pika.ConnectionParameters(BROKER_IP, BROKER_PORT, BROKER_VHOST, self.credentials),
            pika.ConnectionParameters(BROKER_IP, BROKER_PORT, BROKER_VHOST, self.credentials, connection_attempts=5,
                                      retry_delay=1))
        self.connection = pika.BlockingConnection(self.parameters)

        self.channel = self.connection.channel()
        self.exchange = exchange
        self.channel.basic_qos(prefetch_count=1)
        self.exchange = exchange
        self.queue = queue
        self.channel.exchange_declare(
            exchange=exchange,
            exchange_type="direct",
            passive=False,
            durable=True,
            auto_delete=False)
        self.channel.queue_declare(queue=queue, durable=True)
        self.channel.queue_bind(
            queue=queue, exchange=exchange, routing_key=rk)
        self.channel.basic_qos(prefetch_count=1)

    def consum_message(self):
        # LOGGER.info('Comsummer by {}'.format(name))
        def on_message(chan, method_frame, _header_frame, body, userdata=None):
            """Called when a message is received. Log message and ack it."""
            # LOGGER.info('Userdata: %s Message body: %s', userdata, body)
            # print(" [x] Received %r" % body.decode())
            data = body.decode()
            result = alarmFun(data)
            publish = Publish(exchange='spider', queue='alarm', rk='rk-alarm')
            publish.produce(result)
            # chan.basic_ack(delivery_tag=method_frame.delivery_tag)

        on_message_callback = functools.partial(on_message)

        self.channel.basic_consume(on_message_callback=on_message_callback,
                                   queue=self.queue,
                                   auto_ack=True
                                   )
        try:
            self.channel.start_consuming()

        except KeyboardInterrupt:
            self.channel.stop_consuming()

    def cancel(self):
        self.connection.close()

def test():
    consummer = Consummer('test_yerik', 'test_test', 'rk-test_test')
    consummer.consum_message()
    print(consummer.receive)

if __name__ == '__main__':
    test()

參考文檔：

1. 深入RabbitMQ， Gavin M.Roy 著 汪佳南 鄭天民 譯