rabbitmq篇---python的pika 库常用函数及参数说明

python的pika 库常用函数及参数说明

1. pika.PlainCredentials(username, password, erase_on_connect)

• 功能:创建连接时的登录凭证
• 参数:
  • username: MQ 账号
  • password: MQ 密码
  • erase_on_connect: 删除连接上的凭据, 默认为 False

credentials = pika.PlainCredentials(username="guest", password="guest")

2.pika.ConnectionParameters(host, port, virtual_host, credentials)

• 功能: 连接 MQ 的参数设置
• 参数:
  • host: RabbitMQ IP 地址
  • port: RabbitMQ 端口
  • virtual_host: RabbitMQ 虚拟主机
  • credentials: 登录凭证

pika.ConnectionParameters(host='127.0.0.1', port=5672, virtual_host='/', credentials=credentials)

3. pika.BlockingConnection(parameters)

• 功能: 阻塞式连接 MQ
• 参数:
  • parameters: 连接参数(包含主机/端口/虚拟主机/账号/密码等凭证信息)

connect = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1', port=5672, virtual_host='/', credentials=credentials))

4 pika.channel(channel_number)

• 功能: 创建信道
• 参数:
  • channel_number: 信道个数, 一般采用默认值 None

channel = connect.channel()

5. channel.exchange_declare(callback,exchange,exchange_type,passive,durable,auto_delete,internal,nowait,arguments)

• 功能: 声明交换器
• 参数:
  • callback=None : 当 Exchange.DeclareOk 时 调用该方法, 当 nowait=True 该值必须为 None
  • exchange=None: 交换器名称,保持非空,由字母、数字、连字符、下划线、句号组成
  • exchange_type=‘direct’: 交换器类型
  • passive=False: 执行一个声明或检查它是否存在
  • durable=False: RabbitMQ 重启时保持该交换器的持久性,即不会丢失
  • auto_delete=False: 没有队列绑定到该交换器时,自动删除该交换器
  • internal=False: 只能由其它交换器发布-Can only be published to by other exchanges
  • nowait=False: 不需要 Exchange.DeclareOk 的响应-Do not expect an Exchange.DeclareOk response
  • arguments=None: 对该交换器自定义的键/值对, 默认为空

channel.exchange_declare(exchange='hello', exchange_type='direct', passive=False, durable=True, auto_delete=False)

6. channel.queue_declare(callback,queue,passive,durable,exclusive,auto_delete,nowait,arguments)

• 功能: 声明队列
• 参数:
  • callback : 当 Queue.DeclareOk 时的回调方法; 当 nowait=True 时必须为 None.
  • queue=’’ : 队列名称
  • passive=False : 只检查队列是否存在
  • durable=False : 当 RabbitMQ 重启时,队列保持持久性
  • exclusive=False : 仅仅允许当前的连接访问
  • auto_delete=False : 当消费者取消或者断开连接时, 自动删除该队列
  • nowait=False : 当 Queue.DeclareOk 时不需要等待
  • arguments=None : 对该队列自定义键/值对
  • channel.queue_declare(queue=‘hello’)

channel.queue_declare(queue='hello')

7. channel.queue_bind(callback, queue, exchange,routing_key,nowait,arguments)

• 功能: 通过路由键将队列和交换器绑定
• 参数:
  • callback: 当 Queue.BindOk 时的回调函数, 当 nowait=True 时必须为 None
  • queue: 要绑定到交换器的队列名称
  • exchange: 要绑定的源交换器
  • routing_key=None: 绑定的路由键
  • nowait=False: 不需要 Queue.BindOk 的响应
  • arguments=None: 对该绑定自定义键/值对
  • channel.queue_bind(queue=‘hello’, exchange=‘hello’, routing_key=‘world’)

channel.queue_bind(queue='hello', exchange='hello', routing_key='world')

8. channel.basic_publish(exchange, routing_key, body, properties, mandatory, immediate)

• 功能: 将消息发布到 RabbitMQ 交换器上
• 参数:
  • exchange: 要发布的目标交换器
  • routing_key: 该交换器所绑定的路由键
  • body: 携带的消息主体
  • properties=None: 消息的属性,即文本/二进制等等
  • mandatory=False: 当 mandatory 参数设置为 true 时，交换机无法根据自身的路由键找到一个符合的队列，
    那么 RabbitMQ 会调用 Basic.Return 命令将消息返回给生产者，
    当 mandatory 参数设置为 false 时，出现上述情况，消息会被丢弃
  • immediate=False: 立即性标志

channel.basic_publish(exchange='hello',  routing_key='world',  properties=msg_props, body=message)

9. channel.basic_consume(consumer_callback, queue, no_ack, exclusive, consumer_tag, arguments)

• 功能: 从队列中拿到消息开始消费
• 参数:
  • consumer_callback:
  • 先来看官方的解释

The function to call when consuming with the signature 
consumer_callback(channel, method, properties,body),where
                                channel: pika.Channel
                                method: pika.spec.Basic.Deliver
                                properties: pika.spec.BasicProperties
                                body: str, unicode, or bytes (python 3.x)

意思是说,当要消费时,调用该回调函数 consumer_callback, 函数的参数有channel, method, properties,body
+ queue=’’: 要消费的消息队列
+ no_ack=False: 自动确认已经消费成功
+ exclusive=False: 不允许其它的消费者消费该队列
+ consumer_tag=None: 指定自己的消费标记
+ arguments=None: 对该消费者自定义设置键值对

def callback(channel, method, properties, body):
    # 消息确认
    channel.basic_ack(delivery_tag=method.delivery_tag)

    if body.decode('utf-8') == "quit":
        # 停止消费，并退出
        channel.basic_cancel(consumer_tag='hello-consumer')
        channel.close()
        connect.close()
    else:
        print("msg is {}".format(body))

10. channel.basic_ack()

• 功能: 消息确认
• 参数:
  • delivery_tag=0 : 服务端分配的传递标识
  • multiple=False:

If set to True, the delivery tag is treated as “up to and including”, so that multiple messages can be acknowledged with a single method. If set to False, the delivery tag refers to a single message. If the multiple field is 1, and the delivery tag is zero, this indicates acknowledgement of all outstanding messages.

channel.basic_ack(delivery_tag=method.delivery_tag)

11. channel.basic_cancel(callback, consumer_tag, nowait)

• 功能: 取消消费, 该方法不会影响已经发送的消息,但是不会再发送新的消息给消费者
• 参数:
  • callback=None : 当 Basic.CancelOk 响应时的回调函数; 当 nowait=True 时必须为 None. 当 nowait=False 时必须是可回调的函数
  • consumer_tag=’’: 消费标识
  • nowait=False : 不期望得到 Basic.CancelOk response

If set to True, the delivery tag is treated as “up to and including”, so that multiple messages can be acknowledged with a single method. If set to False, the delivery tag refers to a single message. If the multiple field is 1, and the delivery tag is zero, this indicates acknowledgement of all outstanding messages.

channel.basic_cancel(consumer_tag='hello-consumer')

12. channel.start_consuming()

• 功能: 处理 I/O 事件和 basic_consume 的回调, 直到所有的消费者被取消
• 参数:
  • NA

channel.start_consuming()

13. channel.basic_reject(delivery_tag, requeue=True)

• 功能: 拒绝单条消息
• 参数:
  • delivery_tag : 传递标签
  • requeue=True : 是否重新放回到队列中去
  • channel.basic_reject()

# requeue True表示拒绝了消息后重新放回队列，False表示丢弃消息
channel.basic_reject(delivery_tag=method.delivery_tag, requeue=True)

14. channel.basic_nack(delivery_tag=None, multiple=False, requeue=True)

• 功能: 拒绝单条或者多条消息
• 参数:
  • delivery_tag=None : 传递标签
  • multiple=False : 是否批量,即多条消息

If set to True, the delivery tag is treated as “up to and including”, so that multiple messages can be acknowledged with a single method. If set to False, the delivery tag refers to a single message. If the multiple field is 1, and the delivery tag is zero, this indicates acknowledgement of all outstanding messages.

+ requeue=True: 是否重新放回到队列中去

channel.basic_nack()

14. channel.queue_delete(callback=None,queue=’’,if_unused=False,if_empty=False,nowait=False)

• 功能: 删除已声明的队列
• 参数:
  • callback=None: The callback to call on Queue.DeleteOk; MUST be None when nowait=True
  • queue=’’: The queue to delete
  • if_unused=False: only delete if it’s unused
  • if_empty=False: only delete if the queue is empty
  • nowait=False: Do not wait for a Queue.DeleteOk
    channel.queue_delete()

channel.queue_delete()

15. channel.exchange_delete(callback=None,exchange=None,if_unused=False,nowait=False)

• 功能: 删除已声明的交换器
• 参数:
  • callback=None: The function to call on Exchange.DeleteOk; MUST be None when nowait=True.
  • exchange=None: The exchange name
  • if_unused=False: only delete if the exchange is unused
  • nowait=False: Do not wait for an Exchange.DeleteOk

channel.exchange_delete()

16. pika.BasicProperties()

• 功能: 发送消息的属性
• 参数:
  • content_type=None,
  • content_encoding=None,
  • headers=None,
  • delivery_mode=None, 声明信息持久化, 使信息持久化，需要声明queue持久化和delivery_mode=2信息持久化
  • priority=None,
  • correlation_id=None,
  • reply_to=None,
  • expiration=None,
  • message_id=None,
  • timestamp=None,
  • type=None,
  • user_id=None,
  • app_id=None,
  • cluster_id=None

msg_props = pika.BasicProperties()
msg_props.content_type = 'text/plain'

17. 回调函数 callback

• 功能: 回调函数
• 参数:
  • channel: 包含channel的一切属性和方法
  • method: 包含 consumer_tag, delivery_tag, exchange, redelivered, routing_key
  • properties: basic_publish 通过 properties 传入的参数
  • body: basic_publish发送的消息

def callback(channel, method, properties, body):
# 消息确认
channel.basic_ack(delivery_tag=method.delivery_tag)

if body.decode('utf-8') == "quit":
    # 停止消费，并退出
    channel.basic_cancel(consumer_tag='hello-consumer')
    channel.close()
    connect.close()
else:
    print("msg is {}".format(body))

参考链接：https://blog.csdn.net/wohu1104/article/details/105598159