Python操作rabbitmq系列(二):多个接收端消费消息

今天,我们要逐步开始讨论rabbitmq稍微高级点的耍法了。了解这一步,对我们设计高并发的系统非常有用。当然,还可以使用kafka。不过还是算了,有几个硬性条件不支持,还是用rabbitmq吧。

循环分发:

启动一个发送端往队列发消息,此时启动多个接收端。发送的消息会对接收端一个一个挨着发送消息。如图:

这就是默认情况下,多个接收端轮流消费消息。队列发送给消费端后,就立即删除。那么问题来了,当某个消费者在处理消息的时候,异常终止了怎么办?此时,我们更希望这样:若是那个消费者挂掉了,消息自动转给另一个消费者处理。

幸好,rabbitmq就有效确认机制。消费者收到消息后,正常处理完成,此时才通知队列可以自由删除。那么问题又来了,消费者挂掉了连确认消息都发不出,该怎么办?rabbitmq维持了消费者的连接信息。消费者挂掉,与server的连接通道会关闭或tcp连接丢失。这时server知道了这个情况,就自动重发消息。

这里还有个问题,就是server挂掉了怎么办?注意: durable=True。这个就是,当server挂了队列还存在。delivery_mode=2:server挂了消息还存在。若是保证消息不丢,这两个参数都要设置。

发送端:

import pika

import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))

channel = connection.channel()

# durable:server挂了队列仍然存在

channel.queue_declare(queue='task_queue', durable=True)

# 使用默认的交换机发送消息。exchange为空就使用默认的。delivery_mode=2:使消息持久化。和队列名称绑定routing_key

message = ' '.join(sys.argv[1:]) or "Hello World!"

channel.basic_publish(exchange='',

routing_key='task_queue',

body=message,

properties=pika.BasicProperties(

delivery_mode=2,

))

print(" [x] Sent %r" % message)

connection.close()

接收端:

import pika

import time

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))

channel = connection.channel()

channel.queue_declare(queue='task_queue', durable=True)

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'.'))

print(" [x] Done")

# 手动对消息进行确认

ch.basic_ack(delivery_tag=method.delivery_tag)

# basic_consume:这个函数有no_ack参数。该参数默认为false。表示:需要对message进行确认。怎么理解:no设置成false,表示要确认

channel.basic_consume(callback, queue='task_queue')

channel.start_consuming()

公平派遣:

此刻,我们已经知道如何保证消息不丢,那么问题又来了。有的消费干得快,有的干得慢。这样分发消息,有的累死有的没事干。这个问题如何解决?

rabbitmq已经考虑到了。那就是:那个干完了,通知给server,server就发送给那个。

在上面的接收端的

channel.basic_consume(callback, queue='task_queue')

代码前加:

channel.basic_qos(prefetch_count=1)

即可

现在,我们的消息都是一个给一个消费者。接下来,我们要讨论,向多个消费者发送相同的消息。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容

  • Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具(例如配置管理,服务发现,断路器,智...
    卡卡罗2017阅读 135,131评论 19 139
  • 来源 RabbitMQ是用Erlang实现的一个高并发高可靠AMQP消息队列服务器。支持消息的持久化、事务、拥塞控...
    jiangmo阅读 10,424评论 2 34
  • 什么叫消息队列 消息(Message)是指在应用间传送的数据。消息可以非常简单,比如只包含文本字符串,也可以更复杂...
    lijun_m阅读 1,374评论 0 1
  • 本文章翻译自http://www.rabbitmq.com/api-guide.html,并没有及时更新。 术语对...
    joyenlee阅读 7,737评论 0 3
  • AMQP大致内容就是,将消息和队列绑定起来,规定让进入到交换机中的具有某个路由键的消息进入到指定队列中去。 Rab...
    StevenMD阅读 1,923评论 0 3