[关闭]
@hainingwyx 2020-10-05T07:22:45.000000Z 字数 4331 阅读 142

消息消费

rabbitmq



消费方法


Basic.Get

示例程序

  1. import rabbitpy
  2. with rabbitpy.Connection() as connection:
  3. with connection.channel() as channel:
  4. queue = rabbitpy.Queue(channel, 'test-messages')
  5. queue.declare()
  6. while True:
  7. message = queue.get()
  8. if message:
  9. message.pprint()
  10. # 确认消息
  11. message.ack()
  12. if message.body == 'stop':
  13. break

Basic.Consume

示例程序

  1. import rabbitpy
  2. for message in rabbitpy.consume('amqp://guest:guest@localhost:5672/%2f',
  3. 'test-messages'):
  4. message.pprint()
  5. # 消息确认
  6. message.ack()

消费者标签
应用程序发出Basic.Comsume时,创建唯一字符串(消费者标签),标识应用程序。RabbitMQ每次都会把该字符串和消息一同发送给应用程序。
客户端库对消费者标签封装,以确定如何处理消息。开发者不用处理消费者标签。

示例代码:监听消息直到,收到停止消息

  1. import rabbitpy
  2. with rabbitpy.Connection() as connection:
  3. with connection.channel() as channel:
  4. for message in rabbitpy.Queue(channel, 'test-messages'):
  5. message.pprint()
  6. message.ack()
  7. if message.body == 'stop':
  8. break

对比

Consume吞吐量更大。Get包含了每条消息的同步通信开销。


消费性能优化


1、no-ack

应用程序发送Basic.Comsume请求时,设置no-ack。表明消费者不进行消费确认。

示例代码:消费不确认

  1. import rabbitpy
  2. with rabbitpy.Connection() as connection:
  3. with connection.channel() as channel:
  4. queue = rabbitpy.Queue(channel, 'test-messages')
  5. for message in queue.consume_messages(no_ack=True):
  6. message.pprint()

2、预取

QoS(Quality of service)中,可设置消费者预先接收一定数量的消息。Basic.Qos一般在Basic.Consume之前设置。

示例程序:指定QoS

  1. import rabbitpy
  2. with rabbitpy.Connection() as connection:
  3. with connection.channel() as channel:
  4. #预取数为10
  5. channel.prefetch_count(10)
  6. for message in rabbitpy.Queue(channel, 'test-messages'):
  7. message.pprint()
  8. message.ack()

应用程序不需要确认每条消息,可确认所有以前未读消息。

示例程序:多消息确认

  1. import rabbitpy
  2. with rabbitpy.Connection() as connection:
  3. with connection.channel() as channel:
  4. channel.prefetch_count(10)
  5. for message in rabbitpy.Queue(channel, 'test-messages'):
  6. message.pprint()
  7. unacknowledged += 1
  8. if unacknowledged == 10:
  9. # 确认所有未确认消息
  10. message.ack(all_previous=True)
  11. unacknowledged = 0

3、事务

事务允许消费者应用程序提交和回滚批量操作。不适用QoS时,可以获得轻微的性能提升。


拒绝消息


Basic.Reject

通知rabbitmq无法处理投递的消息(拒绝一个消息),可指示rabbitMQ丢弃消息或使用requeue重发消息。

示例程序:消息拒绝

  1. import rabbitpy
  2. for message in rabbitpy.consume('amqp://guest:guest@localhost:5672/%2f',
  3. 'test-messages'):
  4. message.pprint()
  5. print('Redelivered: %s' % message.redelivered)
  6. message.reject(True)

Basic.Nack

同时拒绝多个消息

死信交换器(DLX)

创建队列时声明该交换器用于保存被拒绝的消息。队列x-dead-letter-exchange参数(RPC请求)指定死信交换器。

示例程序:指定死信交换器

  1. import rabbitpy
  2. with rabbitpy.Connection() as connection:
  3. with connection.channel() as channel:
  4. #死信交换器
  5. rabbitpy.Exchange(channel, 'rejected-messages').declare()
  6. queue = rabbitpy.Queue(channel, 'dlx-example',
  7. dead_letter_exchange='rejected-messages')
  8. queue.declare()

控制队列


临时队列

自动删除队列
消费者完成连接和检索消息,所有消费者断开连接时,队列将被删除。

示例程序:自动删除队列auto_delete=True

  1. import rabbitpy
  2. with rabbitpy.Connection() as connection:
  3. with connection.channel() as channel:
  4. queue = rabbitpy.Queue(channel, 'ad-example', auto_delete=True)
  5. queue.declare()

只允许单个消费者
只有单个消费者能够消费队列中的消息。消费者断开连接后,会自动删除队列。

示例程序:独占队列exclusive

  1. import rabbitpy
  2. with rabbitpy.Connection() as connection:
  3. with connection.channel() as channel:
  4. queue = rabbitpy.Queue(channel, 'exclusive-example',
  5. exclusive=True)
  6. queue.declare()

自动过期队列
如果一段时间没有使用该队列就删除它,一般用于RPC回复队列。

示例程序:自动过期队列

  1. import rabbitpy
  2. import time
  3. with rabbitpy.Connection() as connection:
  4. with connection.channel() as channel:
  5. queue = rabbitpy.Queue(channel, 'expiring-queue',
  6. arguments={'x-expires': 1000})
  7. queue.declare()
  8. messages, consumers = queue.declare(passive=True)
  9. time.sleep(2)
  10. try:
  11. messages, consumers = queue.declare(passive=True)
  12. except rabbitpy.exceptions.AMQPNotFound:
  13. print('The queue no longer exists')

永久队列

队列持久性
服务器重启后队列仍然存在。
示例程序:持久队列

  1. import rabbitpy
  2. with rabbitpy.Connection() as connection:
  3. with connection.channel() as channel:
  4. queue = rabbitpy.Queue(channel, 'durable-queue',
  5. durable=True)
  6. if queue.declare():
  7. print('Queue declared')

队列消息自动过期
同时指定死信交换器和消息TTL,过期消息将成为死信消息。

示例程序:消息TTL

  1. import rabbitpy
  2. with rabbitpy.Connection() as connection:
  3. with connection.channel() as channel:
  4. queue = rabbitpy.Queue(channel, 'expiring-msg-queue',
  5. arguments={'x-message-ttl': 1000})
  6. queue.declare()

最大队列长度
一旦达到最大值,添加新消息时,删除队列前端的消息。声明队列时,如果指定死信交换器,前端移除的消息将成为死信。

示例程序:最大长度队列

  1. import rabbitpy
  2. with rabbitpy.Connection() as connection:
  3. with connection.channel() as channel:
  4. queue = rabbitpy.Queue(channel, 'max-length-queue',
  5. arguments={'x-max-length': 1000})
  6. queue.declare()

队列设置参数

参数 说明
x-dead-letter-exchange 死信交换器,路由不重发且被拒绝的消息
x-dead-letter-routing-key 死信消息的可选路由键
x-expires 队列在指定的毫秒数后删除
x-ha-proxy 创建HA队列
x-ha-nodes HA队列分布的节点
x-max-length 队列的最大消息数
x-message-ttl 毫秒为单位的队列过期时间
x-max-priority 队列优先级排序
添加新批注
在作者公开此批注前,只有你和作者可见。
回复批注