最近由于業(yè)務(wù)需要進行性能升級,將原來需要經(jīng)過http進行數(shù)據(jù)交互的方式修改為消息隊列的形式。于是原來的同步處理的方式變成了異步處理,在一定程度上提升我們系統(tǒng)的性能,不過debug的時候,不免哭了出來。因為每個環(huán)節(jié)都需要進行詳細(xì)檢查。
對于RabbitMQ,我們知道,其是AMQP的一種代理服服務(wù)器,具有一套嚴(yán)格的通信方式,即在核心產(chǎn)品進行通信的各個方面幾乎都采用了RPC(Remote Procedure Call, 遠(yuǎn)程過程調(diào)用)
模式。
RabbitMQ通信時用到的類和方法與AMQP協(xié)議層面的類和方法一一對應(yīng)。因此AMQP本質(zhì)上是RPC的一種傳輸機制
AMQ(Advanced Message Queuing)
模型,這個模型是針對代理服務(wù)器軟件例如(RabbitMQ)設(shè)計的,該模型在邏輯上定義了三種抽象組件用于指定消息的路由行為,分別是:
Exchange
,消息代理服務(wù)器中用于把消息路由到隊列的組件Queue
,用來存儲消息的數(shù)據(jù)結(jié)構(gòu),位于硬盤或內(nèi)存中,以FIFO的順序進行投遞Binding
,一套規(guī)則,用于告訴交換器消息應(yīng)該被存儲到哪個隊列
在將消息發(fā)布到隊列之前,我們需要經(jīng)歷過以下若干個步驟。至少,必須要設(shè)置交換器和隊列,然后將他們綁定再一起。接下來我們將通過python來實現(xiàn)AMQP機制。
我用到了pika這個庫,需要的話,需要通過以下指令安裝。該庫實現(xiàn)了絕大部分rabbitmq的api以及提供了相關(guān)的調(diào)優(yōu)參數(shù),后續(xù)有機會不妨可以詳談。
pip install pika
交換器在AMQ模型中是非常重要的角色存在。因此,在AMQP規(guī)范中都有自己的類。聲明一個交換器,我們可以直接在控制臺界面進行創(chuàng)建。
不過這樣僅僅是在極少數(shù)的情況下才適合,動手調(diào)戲鼠標(biāo)對開發(fā)工程師的來說實在是太蠢啦,能玩鍵盤就別玩鼠標(biāo)啊,我們不妨通過以下代碼來聲明(創(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)
一旦交換器創(chuàng)建成功,就可以通過發(fā)送類似queue.declare命令讓rabbitmq創(chuàng)建一個隊列。同樣的,我們?nèi)匀豢梢栽趫D形化界面里面創(chuàng)建隊列。
還是那句話,動手調(diào)戲鼠標(biāo)對開發(fā)工程師的來說實在是太蠢啦,能玩鍵盤就別玩鼠標(biāo)啊,我們不妨通過以下代碼來聲明(創(chuàng)建)若干個隊列。pika內(nèi)置函數(shù)會事先通過get的方式來檢查我們待聲明的隊列是否存在,如果存在則不創(chuàng)建,否則創(chuàng)建一個新的隊列。
self.channel.queue_declare(queue=queue, durable=True)
當(dāng)隊列同名時,即如果我們多次發(fā)送同一個queue.declare命令并不會有任何副作用,因為RabbitMQ并不會處理后續(xù)的隊列聲明,究其原因,每次創(chuàng)建都會先通過get的方式調(diào)用消息隊列引擎查詢隊列是否存在。如果需要返回隊列相關(guān)的有用信息,則將會返回隊列中待處理消息的數(shù)量以及該隊列的消費者數(shù)量。當(dāng)然了如果隊列同名,而且新隊列的屬性與原有的隊列不一樣,那么RabbitMQ將關(guān)閉發(fā)出的RPC請求的信道,返回403錯誤
一旦創(chuàng)建了交換器和隊列,之后就可以將它們綁定在一起了,如同queue.declare命令,將隊列綁定到交換器Queue.Bind每次只能指定一個隊列。我們既可以通過圖形化界面進行綁定,也可以通過代碼實現(xiàn)這個效果
self.channel.queue_bind(
queue=queue, exchange=exchange, routing_key=rk)
發(fā)布消息到RabbitMQ時,多個幀封裝了發(fā)送到服務(wù)器的消息數(shù)據(jù)。在實際的消息內(nèi)容到達rabbitMQ之前,客戶端應(yīng)用程序會發(fā)送一個basic.publish方法幀、一個內(nèi)容頭幀和至少一個消息體幀。
默認(rèn)情況下,只要沒有消費者正在監(jiān)聽隊列,消息就會被存儲在隊列中。當(dāng)添加更多消息時,隊列大小也會隨之增加。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)
)
一旦發(fā)布消息被路由并且保存在一個或者多個隊列中,剩下的就是如何對其進行消費。注意到,發(fā)送和消費是異步的。 消費時,可以讓RabbitMQ知道如何消費他們
Basic.Consume命令中
no_ack為true時,RabbitMQ將連續(xù)發(fā)送消息直到消費者發(fā)送一個Basic.Cancel命令或者斷開連接為止
如果為false,則需要發(fā)送一個Basic.Ack來確認(rèn)收到每條消息的請求
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
)
經(jīng)過前面的描述,我們需要理論聯(lián)系實踐,讓我們通過python開發(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()
參考文檔:
- 深入RabbitMQ, Gavin M.Roy 著 汪佳南 鄭天民 譯
另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)cdcxhl.cn,海內(nèi)外云服務(wù)器15元起步,三天無理由+7*72小時售后在線,公司持有idc許可證,提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國服務(wù)器、虛擬主機、免備案服務(wù)器”等云主機租用服務(wù)以及企業(yè)上云的綜合解決方案,具有“安全穩(wěn)定、簡單易用、服務(wù)可用性高、性價比高”等特點與優(yōu)勢,專為企業(yè)上云打造定制,能夠滿足用戶豐富、多元化的應(yīng)用場景需求。
當(dāng)前文章:RabbitMQ實踐體驗-創(chuàng)新互聯(lián)
URL地址:http://muchs.cn/article24/ddscje.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供移動網(wǎng)站建設(shè)、企業(yè)建站、標(biāo)簽優(yōu)化、用戶體驗、虛擬主機、網(wǎng)站營銷
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請盡快告知,我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場,如需處理請聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時需注明來源: 創(chuàng)新互聯(lián)
猜你還喜歡下面的內(nèi)容