RabbitMQ实战2.消息轮询、响应、持久化消息轮询分配消息响应防丢失消息持久化参考资料
RabbitMQ实战2.消息轮询、响应、持久化消息轮询分配消息响应防丢失消息持久化参考资料
章鱼喵
发布于 2018-08-10 11:35:15
发布于 2018-08-10 11:35:15
继上篇 RabbitMQ实战1.消息代理
消息轮询分配
如果生产者投递的消息需要运行相当长的时间，且有多个消费者在处理消息，那么RabbitMQ是怎么分配消息的？
新建 new_task.py
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue')
message = ' '.join(sys.argv[1:]) or "Hello World!" # 接收标准输入的参数
channel.basic_publish(exchange='',
routing_key='task_queue',
body=message)
print(" [x] Sent %r" % message)
connection.close()新建 worker.py
import pika
import time
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue')
print(' [*] Waiting for messages. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
time.sleep(body.count(b'.')) # 用 . 来模拟耗时任务，一个 . 表示耗时1秒
print(" [x] Done")
channel.basic_consume(callback,
queue='task_queue',
no_ack=True)
channel.start_consuming()在三个终端开启消费者：
☁ rabbitMq [master] ⚡ python new_task.py First message.
[x] Sent 'First message.'
☁ rabbitMq [master] ⚡ python new_task.py Second message..
[x] Sent 'Second message..'
☁ rabbitMq [master] ⚡ python new_task.py Third message...
[x] Sent 'Third message...'
☁ rabbitMq [master] ⚡ python new_task.py Fourth message....
[x] Sent 'Fourth message....'
☁ rabbitMq [master] ⚡ python new_task.py Fifth message.....
[x] Sent 'Fifth message.....'各个消费者的情况：
☁ rabbitMq [master] ⚡ python worker.py # 消息者1
[*] Waiting for messages. To exit press CTRL+C
[x] Received b'First message.'
[x] Done
[x] Received b'Fourth message....'
[x] Done
☁ rabbitMq [master] ⚡ python worker.py # 消费者2
[*] Waiting for messages. To exit press CTRL+C
[x] Received b'Second message..'
[x] Done
[x] Received b'Fifth message.....'
[x] Done
☁ rabbitMq [master] ⚡ python worker.py # 消费者3
[*] Waiting for messages. To exit press CTRL+C
[x] Received b'Third message...'
[x] Done由此可见，五条消息被轮流分配到不同的消费者
默认来说，RabbitMQ会按顺序把消息发送给每个消费者（consumer）。平均每个消费者都会收到同等数量得消息。这种发送消息得方式叫做——轮询（round-robin）
一对多
消息响应防丢失
在以上的代码中，当消息被RabbitMQ发送给消息者后，就从内存中移除了。如果消费者在执行任务过程中挂掉了，这条消息就相当于被丢失了，不会再被指派到其他消费者。
这对于要求数据完整性的业务来说，是不可忍受的。
怎样防止消息丢失？我们可以延迟RabbitMQ移除消息的时间，当消费者将任务完成后，发送一个响应给RabbitMQ，此时RabbitMQ再移除消息。如果消费者在发送响应前挂掉了，RabbitMQ会将此消息重新发送给其他消费者。
消息没有超时的概念，即使是处理耗时相当长的任务也不会有问题。只有挂掉，RabbitMQ才会重发
消息响应默认是开启的，可使用no_ack=True标识把它关闭
- 将 worker.py 的
no_ack=True取消
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
time.sleep(body.count(b'.'))
print(" [x] Done")
ch.basic_ack(delivery_tag=method.delivery_tag) # 返回响应
channel.basic_consume(callback,
queue='task_queue')- 重新开启两个消费者 worker.py
- 生产者启动一个任务
☁ rabbitMq [master] ⚡ python new_task.py ten message..........
[x] Sent 'ten message..........'- 将正在执行的消费者关闭, 另一个消费者会自动接管此条消息
☁ rabbitMq [master] ⚡ python worker.py # 消费者1执行一半，将进程关闭
[*] Waiting for messages. To exit press CTRL+C
[x] Received b'ten message..........'
^CTraceback (most recent call last):
File "worker.py", line 21, in <module>
channel.start_consuming()
...
☁ rabbitMq [master] ⚡ python worker.py # 消费者2继续执行消费者1未完成的任务
[*] Waiting for messages. To exit press CTRL+C
[x] Received b'ten message..........'
[x] Done消息持久化
默认情况下，当RabbitMQ崩溃时，会丢失所有队列及消息，可通过设置durable=True使消息持久化
在 worker.py 声明队列时，添加持久化的配置参数
☁ rabbitMq [master] ⚡ python worker.py
Traceback (most recent call last):
File "worker.py", line 8, in <module>
channel.queue_declare(queue='task_queue', durable=True)代码报错是因为之前我们声明的队列task_queue是非持久化的，同一个队列，不能以不同的参数重新定义。解决方案：
- 修改队列名
- 将原队列删除
☁ rabbitMq [master] ⚡ rabbitmqadmin delete queue name=task_queue # 根据队列名删除队列
queue deleted
☁ rabbitMq [master] ⚡ python worker.py
[*] Waiting for messages. To exit press CTRL+C参考资料
本文参与 腾讯云自媒体分享计划，分享自作者个人站点/博客。
原始发表：2018.08.09 ，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
目录