RabbitMQ異步收發

pika提供了支持異步發送模式的selectconnection方法支持異步發送接收（通過回調的方式）。

connectioon建立時回調建立channel， channel建立時一次回調各種declare方法，declare建立時依次回調publish。

同使用blockconnection方法相比，通過wireshark抓包來看，使用 異步的方式會對發包進行一些優化，會將幾個包合並成一個大包，然后做一次ack應答從而提高效率，與之相反使用blockconnection時將會做至少兩次ack，head一次content一次等。

因此再試用異步的方式時會獲得一定的優化。

　　1.異步發送publish:

import pika
import time as timer
from time import time

connection = None


def on_open(conn):
    conn.channel(on_open_callback=on_channel_open)


def on_channel_open(channel):
    message = 'message body value'
    for i in range(5):
        channel.basic_publish('',
                              'hello-01',
                              message,
                              pika.BasicProperties(content_type='text/plain',
                                                   delivery_mode=1))

    connection.close()


credentials = pika.PlainCredentials("wjq", "1234")
parameters = pika.ConnectionParameters(host='192.168.139.128', credentials=credentials)

connection = pika.SelectConnection(
    parameters=parameters, on_open_callback=on_open)
try:
    connection.ioloop.start()
    connection.ioloop.poller.open = False
    connection.close()
except KeyboardInterrupt:
    connection.close()
    connection.ioloop.start()

　　2.異步接收consum:

import pika
import logging
import traceback
import time as timer
from time import time

mylog = logging.getLogger('pika')
mylog.setLevel(logging.ERROR)
ch = logging.StreamHandler()
ch.setLevel(logging.ERROR)
mylog.addHandler(ch)


def on_open(connection):
    connection.channel(on_open_callback=on_channel_open)


channelg = None
begin_time = time()


# 回調處理函數
def on_message(unused_channel, basic_deliver, properties, body):
    print("body>>>", body.decode("utf-8"))
    timer.sleep(0.1)
    unused_channel.basic_ack(delivery_tag=basic_deliver.delivery_tag)


def on_channel_open(channel):
    try:
        channel.basic_consume('hello-01', on_message, False)
        # channel.start_consuming()
    except (Exception,):
        print(traceback.format_exc(), ">>>")


credentials = pika.PlainCredentials("wjq", "1234")
parameters = pika.ConnectionParameters(host='192.168.139.128', credentials=credentials)
connection = pika.SelectConnection(
    parameters=parameters, on_open_callback=on_open)

try:
    connection.ioloop.start()
except KeyboardInterrupt:
    connection.close()
    connection.ioloop.start()

 

參考地址：

　　https://www.jianshu.com/p/a4671c59351a   -- 建議

　　https://www.cnblogs.com/cwp-bg/p/8426188.html

　　https://blog.51cto.com/8415580/1351328