pika详解 (一)

àì夳堔傛蜴生んèń 2021-12-14 23:05 483阅读 0赞

pika

pika处理消息可以简单分为以下几个步骤：

1.  我们首先创建连接对象，然后启动事件循环。
2.  当有连接时，调用on\_connected方法。在该方法中创建channel
3.  channel创建完成，将调用on\_channel\_open方法。在该方法中，声明了一个queue。
4.  queue声明成功后，将调用on\_queue\_declared。在该方法中，调用channel.basic\_consume，为RabbitMQ传递的每条消息调用handle\_delivery。
5.  当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()

认证：

认证使用PlainCredentials ，传递给credentials参数

import pika
    credentials = pika.PlainCredentials('username', 'password')
    parameters = pika.ConnectionParameters(credentials=credentials)

连接器的参数传递：

有两种方式：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')

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()

拥有4种连接器

1.  pika.BlockingConnection - 同步模式， 简单易用
2.  pika.SelectConnection - 没有第三方依赖包的异步模式
3.  pika.adapters.tornado\_connection.TornadoConnection - 基于Tornado 的异步IO请求模式
4.  pika.adapters.twisted\_connection.TwistedProtocolConnection - 基于Twisted’的异步IO请求模式