原文：https://blog.csdn.net/comprel/article/details/94592316
pika
pika處理消息可以簡單分為以下幾個步驟：
我們首先創建連接對象，然后啟動事件循環。
當有連接時，調用on_connected方法。在該方法中創建channel
channel創建完成，將調用on_channel_open方法。在該方法中，聲明了一個queue。
queue聲明成功后，將調用on_queue_declared。在該方法中，調用channel.basic_consume，為RabbitMQ傳遞的每條消息調用handle_delivery。
當RabbitMQ有發送消息，將調用handle_delivery方法傳遞AMQP Method框架，Header框架和Body
官方給出的示例如下：
import pika
# Create a global channel variable to hold our channel object in
channel = None
# Step #2
def on_connected(connection):
"""Called when we are fully connected to RabbitMQ"""
# Open a channel
connection.channel(on_channel_open)
# Step #3
def on_channel_open(new_channel):
"""Called when our channel has opened"""
global channel
channel = new_channel
channel.queue_declare(queue="test", durable=True, exclusive=False, auto_delete=False, callback=on_queue_declared)
# Step #4
def on_queue_declared(frame):
"""Called when RabbitMQ has told us our Queue has been declared, frame is the response from RabbitMQ"""
channel.basic_consume('test', handle_delivery)
# Step #5
def handle_delivery(channel, method, header, body):
"""Called when we receive a message from RabbitMQ"""
print(body)
# Step #1: Connect to RabbitMQ using the default parameters
parameters = pika.ConnectionParameters()
connection = pika.SelectConnection(parameters, on_connected)
try:
# Loop so we can communicate with RabbitMQ
connection.ioloop.start()
except KeyboardInterrupt:
# Gracefully close the connection
connection.close()
# Loop until we're fully closed, will stop on its own
connection.ioloop.start()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
認證：
認證使用PlainCredentials ，傳遞給credentials參數
import pika
credentials = pika.PlainCredentials('username', 'password')
parameters = pika.ConnectionParameters(credentials=credentials)
1
2
3
連接器的參數傳遞：
有兩種方式：ConnectionParameters 和 URLParameters
TCP Backpressure
TCP背壓， 流控機制的一種， 在rabbitmq 2.0 channel.flow移除，使用tcp backpresssure進行限流。 參數backpressure_detection。 如果pika發現有太多消息積壓， 將調用通過add_backpressure_callback注冊的回調函數。
默認超過平常10倍的積壓將會調用，這個參數也可以設置， 設置方法為set_backpressure_multiplier 值為整數
例如：
import pika
parameters = pika.URLParameters('amqp://guest:guest@rabbit-server1:5672/%2F?backpressure_detection=t')
1
2
3
pika擴展時， 需要啟動監聽，使用connection.ioloop.start()方法
import pika
def on_open(connection):
# Invoked when the connection is open
pass
# Create our connection object, passing in the on_open method
connection = pika.SelectConnection(on_open_callback=on_open)
try:
# Loop so we can communicate with RabbitMQ
connection.ioloop.start()
except KeyboardInterrupt:
# Gracefully close the connection
connection.close()
# Loop until we're fully closed, will stop on its own
connection.ioloop.start()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
擁有4種連接器
pika.BlockingConnection - 同步模式， 簡單易用
pika.SelectConnection - 沒有第三方依賴包的異步模式
pika.adapters.tornado_connection.TornadoConnection - 基於Tornado 的異步IO請求模式
pika.adapters.twisted_connection.TwistedProtocolConnection - 基於Twisted’的異步IO請求模式