当前位置 知且行 其他 正文 下一篇:

RabbitMQ实战2.消息轮询、响应、持久化

消息轮询分配

如果生产者投递的消息需要运行相当长的时间,且有多个消费者在处理消息,那么RabbitMQ是怎么分配消息的?

新建 new_task.py

  1. import pika
  2. import sys
  3. connection = pika.BlockingConnection(pika.ConnectionParameters(
  4. host='localhost'))
  5. channel = connection.channel()
  6. channel.queue_declare(queue='task_queue')
  7. message = ' '.join(sys.argv[1:]) or "Hello World!" # 接收标准输入的参数
  8. channel.basic_publish(exchange='',
  9. routing_key='task_queue',
  10. body=message)
  11. print(" [x] Sent %r" % message)
  12. connection.close()

新建 worker.py

  1. import pika
  2. import time
  3. connection = pika.BlockingConnection(pika.ConnectionParameters(
  4. host='localhost'))
  5. channel = connection.channel()
  6. channel.queue_declare(queue='task_queue')
  7. print(' [*] Waiting for messages. To exit press CTRL+C')
  8. def callback(ch, method, properties, body):
  9. print(" [x] Received %r" % body)
  10. time.sleep(body.count(b'.')) # 用 . 来模拟耗时任务,一个 . 表示耗时1秒
  11. print(" [x] Done")
  12. channel.basic_consume(callback,
  13. queue='task_queue',
  14. no_ack=True)
  15. channel.start_consuming()

在三个终端开启消费者:

  1. rabbitMq [master] python new_task.py First message.
  2. [x] Sent 'First message.'
  3. rabbitMq [master] python new_task.py Second message..
  4. [x] Sent 'Second message..'
  5. rabbitMq [master] python new_task.py Third message...
  6. [x] Sent 'Third message...'
  7. rabbitMq [master] python new_task.py Fourth message....
  8. [x] Sent 'Fourth message....'
  9. rabbitMq [master] python new_task.py Fifth message.....
  10. [x] Sent 'Fifth message.....'

各个消费者的情况:

  1. rabbitMq [master] python worker.py # 消息者1
  2. [*] Waiting for messages. To exit press CTRL+C
  3. [x] Received b'First message.'
  4. [x] Done
  5. [x] Received b'Fourth message....'
  6. [x] Done
  7. rabbitMq [master] python worker.py # 消费者2
  8. [*] Waiting for messages. To exit press CTRL+C
  9. [x] Received b'Second message..'
  10. [x] Done
  11. [x] Received b'Fifth message.....'
  12. [x] Done
  13. rabbitMq [master] python worker.py # 消费者3
  14. [*] Waiting for messages. To exit press CTRL+C
  15. [x] Received b'Third message...'
  16. [x] Done

由此可见,五条消息被轮流分配到不同的消费者

默认来说,RabbitMQ会按顺序把消息发送给每个消费者(consumer)。平均每个消费者都会收到同等数量得消息。这种发送消息得方式叫做——轮询(round-robin)

一对多

消息响应防丢失

在以上的代码中,当消息被RabbitMQ发送给消息者后,就从内存中移除了。如果消费者在执行任务过程中挂掉了,这条消息就相当于被丢失了,不会再被指派到其他消费者。

这对于要求数据完整性的业务来说,是不可忍受的。

怎样防止消息丢失?我们可以延迟RabbitMQ移除消息的时间,当消费者将任务完成后,发送一个响应给RabbitMQ,此时RabbitMQ再移除消息。如果消费者在发送响应前挂掉了,RabbitMQ会将此消息重新发送给其他消费者。

消息没有超时的概念,即使是处理耗时相当长的任务也不会有问题。只有挂掉,RabbitMQ才会重发

消息响应默认是开启的,可使用no_ack=True标识把它关闭

  1. 将 worker.py 的 no_ack=True 取消
  1. def callback(ch, method, properties, body):
  2. print(" [x] Received %r" % body)
  3. time.sleep(body.count(b'.'))
  4. print(" [x] Done")
  5. ch.basic_ack(delivery_tag=method.delivery_tag) # 返回响应
  6. channel.basic_consume(callback,
  7. queue='task_queue')
  1. 重新开启两个消费者 worker.py

  2. 生产者启动一个任务

  1. rabbitMq [master] python new_task.py ten message..........
  2. [x] Sent 'ten message..........'
  1. 将正在执行的消费者关闭, 另一个消费者会自动接管此条消息
  1. rabbitMq [master] python worker.py # 消费者1执行一半,将进程关闭
  2. [*] Waiting for messages. To exit press CTRL+C
  3. [x] Received b'ten message..........'
  4. ^CTraceback (most recent call last):
  5. File "worker.py", line 21, in <module>
  6. channel.start_consuming()
  7. ...
  8. rabbitMq [master] python worker.py # 消费者2继续执行消费者1未完成的任务
  9. [*] Waiting for messages. To exit press CTRL+C
  10. [x] Received b'ten message..........'
  11. [x] Done

消息持久化

默认情况下,当RabbitMQ崩溃时,会丢失所有队列及消息,可通过设置durable=True使消息持久化

在 worker.py 声明队列时,添加持久化的配置参数

  1. rabbitMq [master] python worker.py
  2. Traceback (most recent call last):
  3. File "worker.py", line 8, in <module>
  4. channel.queue_declare(queue='task_queue', durable=True)

代码报错是因为之前我们声明的队列task_queue是非持久化的,同一个队列,不能以不同的参数重新定义。解决方案:

  1. 修改队列名

  2. 将原队列删除

  1. rabbitMq [master] rabbitmqadmin delete queue name=task_queue # 根据队列名删除队列
  2. queue deleted
  3. rabbitMq [master] python worker.py
  4. [*] Waiting for messages. To exit press CTRL+C

参考资料

转载必须注明出处:https://www.zhiqiexing.com/58.html

关于我

我希望能成为一个认真、有趣、创造更多价值的人
关注微信
微信扫一扫关注我

微信扫一扫关注我

返回顶部