3rabbitmq消息队列(三):任务分发机制

anjie2017 贡献于2017-01-06

作者 Administrator  创建于2014-10-29 12:08:00   修改者Administrator  修改于2017-01-04 08:19:25字数6356

文档摘要:从本篇文章开始,我们将结合更加实际的应用场景来讲解更多的高级用法。  
关键词:

 在上篇文章中,我们解决了从发送端(Producer)向接收端(Consumer)发送“Hello World”的问题。在实际的应用场景中,这是远远不够的。从本篇文章开始,我们将结合更加实际的应用场景来讲解更多的高级用法。    当有Consumer需要大量的运算时,RabbitMQ Server需要一定的分发机制来balance每个Consumer的load。试想一下,对于web application来说,在一个很多的HTTP request里是没有时间来处理复杂的运算的,只能通过后台的一些工作线程来完成。接下来我们分布讲解。     应用场景就是RabbitMQ Server会将queue的Message分发给不同的Consumer以处理计算密集型的任务: 1. 准备    在上一篇文章中,我们简单在Message中包含了一个字符串"Hello World"。现在为了是Consumer做的是计算密集型的工作,那就不能简单的字符串了。在现实应用中,Consumer有可能做的是一个图片的 resize,或者是pdf文件的渲染或者内容提取。但是作为Demo,还是用字符串模拟吧:通过字符串中的.的数量来决定计算的复杂度,每个.都会消耗 1s,即sleep(1)。     还是复用上篇文章中的code,根据“计算密集型”做一下简单的修改,为了辨别,我们把send.py 的名字换成new_task.py [python]view plaincopy 1. import sys   2.    3. message = ' '.join(sys.argv[1:]) or "Hello World!"   4. channel.basic_publish(exchange='',   5.                       routing_key='hello',   6.                       body=message)   7. print " [x] Sent %r" % (message,)   同样的道理,把receive.py的名字换成worker.py,并且根据Message中的.的数量进行计算密集型模拟: [python]view plaincopy 1. import time   2.    3. def callback(ch, method, properties, body):   4.     print " [x] Received %r" % (body,)   5.     time.sleep( body.count('.') )   6.     print " [x] Done"   2. Round-robin dispatching 循环分发         RabbitMQ的分发机制非常适合扩展,而且它是专门为并发程序设计的。如果现在load加重,那么只需要创建更多的Consumer来进行任务处理即 可。当然了,对于负载还要加大怎么办?我没有遇到过这种情况,那就可以创建多个virtual Host,细化不同的通信类别了。      首先开启两个Consumer,即运行两个worker.py。 Console1: [python]view plaincopy 1. shell1$ python worker.py   2.  [*] Waiting for messages. To exit press CTRL+C   Consule2: [python]view plaincopy 1. shell2$ python worker.py   2.  [*] Waiting for messages. To exit press CTRL+C   Producer new_task.py要Publish Message了: [python]view plaincopy 1. shell3$ python new_task.py First message.   2. shell3$ python new_task.py Second message..   3. shell3$ python new_task.py Third message...   4. shell3$ python new_task.py Fourth message....   5. shell3$ python new_task.py Fifth message.....   注意一下:.代表的sleep(1)。接着开一下Consumer worker.py收到了什么: Console1: [python]view plaincopy 1. shell1$ python worker.py   2.  [*] Waiting for messages. To exit press CTRL+C   3.  [x] Received 'First message.'   4.  [x] Received 'Third message...'   5.  [x] Received 'Fifth message.....'   Console2: [python]view plaincopy 1. shell2$ python worker.py   2.  [*] Waiting for messages. To exit press CTRL+C   3.  [x] Received 'Second message..'   4.  [x] Received 'Fourth message....'   默认情况下,RabbitMQ 会顺序的分发每个Message。当每个收到ack后,会将该Message删除,然后将下一个Message分发到下一个Consumer。这种分发方式叫做round-robin。这种分发还有问题,接着向下读吧。 3. Message acknowledgment 消息确认       每个Consumer可能需要一段时间才能处理完收到的数据。如果在这个过程中,Consumer出错了,异常退出了,而数据还没有处理完成,那么非常不 幸,这段数据就丢失了。因为我们采用no-ack的方式进行确认,也就是说,每次Consumer接到数据后,而不管是否处理完成,RabbitMQ Server会立即把这个Message标记为完成,然后从queue中删除了。      如果一个Consumer异常退出了,它处理的数据能够被另外的Consumer处理,这样数据在这种情况下就不会丢失了(注意是这种情况下)。       为了保证数据不被丢失,RabbitMQ支持消息确认机制,即acknowledgments。为了保证数据能被正确处理而不仅仅是被Consumer收到,那么我们不能采用no-ack。而应该是在处理完数据后发送ack。     在处理数据后发送的ack,就是告诉RabbitMQ数据已经被接收,处理完成,RabbitMQ可以去安全的删除它了。     如果Consumer退出了但是没有发送ack,那么RabbitMQ就会把这个Message发送到下一个Consumer。这样就保证了在Consumer异常退出的情况下数据也不会丢失。     这里并没有用到超时机制。RabbitMQ仅仅通过Consumer的连接中断来确认该Message并没有被正确处理。也就是说,RabbitMQ给了Consumer足够长的时间来做数据处理。     默认情况下,消息确认是打开的(enabled)。在上篇文章中我们通过no_ack = True 关闭了ack。重新修改一下callback,以在消息处理完成后发送ack: [python]view plaincopy 1. def callback(ch, method, properties, body):   2.     print " [x] Received %r" % (body,)   3.     time.sleep( body.count('.') )   4.     print " [x] Done"   5.     ch.basic_ack(delivery_tag = method.delivery_tag)   6.    7. channel.basic_consume(callback,   8.                       queue='hello')        这样即使你通过Ctr-C中断了worker.py,那么Message也不会丢失了,它会被分发到下一个Consumer。         如果忘记了ack,那么后果很严重。当Consumer退出时,Message会重新分发。然后RabbitMQ会占用越来越多的内存,由于 RabbitMQ会长时间运行,因此这个“内存泄漏”是致命的。去调试这种错误,可以通过一下命令打印un-acked Messages: [python]view plaincopy 1. $ sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged   2. Listing queues ...   3. hello    0       0   4. ...done.   4. Message durability消息持久化      在上一节中我们知道了即使Consumer异常退出,Message也不会丢失。但是如果RabbitMQ Server退出呢?软件都有bug,即使RabbitMQ Server是完美毫无bug的(当然这是不可能的,是软件就有bug,没有bug的那不叫软件),它还是有可能退出的:被其它软件影响,或者系统重启 了,系统panic了。。。     为了保证在RabbitMQ退出或者crash了数据仍没有丢失,需要将queue和Message都要持久化。 queue的持久化需要在声明时指定durable=True: [python]view plaincopy 1. channel.queue_declare(queue='hello', durable=True)   上述语句执行不会有什么错误,但是确得不到我们想要的结果,原因就是RabbitMQ Server已经维护了一个叫hello的queue,那么上述执行不会有任何的作用,也就是hello的任何属性都不会被影响。这一点在上篇文章也讨论过。 那么workaround也很简单,声明一个另外的名字的queue,比如名字定位task_queue: [python]view plaincopy 1. channel.queue_declare(queue='task_queue', durable=True)   再次强调,Producer和Consumer都应该去创建这个queue,尽管只有一个地方的创建是真正起作用的: 接下来,需要持久化Message,即在Publish的时候指定一个properties,方式如下: [python]view plaincopy 1. channel.basic_publish(exchange='',   2.                       routing_key="task_queue",   3.                       body=message,   4.                       properties=pika.BasicProperties(   5.                          delivery_mode = 2, # make message persistent   6.                       ))   关于持久化的进一步讨论:     为了数据不丢失,我们采用了: 1. 在数据处理结束后发送ack,这样RabbitMQ Server会认为Message Deliver 成功。 2. 持久化queue,可以防止RabbitMQ Server 重启或者crash引起的数据丢失。 3. 持久化Message,理由同上。     但是这样能保证数据100%不丢失吗?     答案是否定的。问题就在与RabbitMQ需要时间去把这些信息存到磁盘上,这个time window虽然短,但是它的确还是有。在这个时间窗口内如果数据没有保存,数据还会丢失。还有另一个原因就是RabbitMQ并不是为每个 Message都做fsync:它可能仅仅是把它保存到Cache里,还没来得及保存到物理磁盘上。     因此这个持久化还是有问题。但是对于大多数应用来说,这已经足够了。当然为了保持一致性,你可以把每次的publish放到一个transaction中。这个transaction的实现需要user defined codes。     那么商业系统会做什么呢?一种可能的方案是在系统panic时或者异常重启时或者断电时,应该给各个应用留出时间去flash cache,保证每个应用都能exit gracefully。 5. Fair dispatch 公平分发     你可能也注意到了,分发机制不是那么优雅。默认状态下,RabbitMQ将第n个Message分发给第n个Consumer。当然n是取余后的。它不管Consumer是否还有unacked Message,只是按照这个默认机制进行分发。    那么如果有个Consumer工作比较重,那么就会导致有的Consumer基本没事可做,有的Consumer却是毫无休息的机会。那么,RabbitMQ是如何处理这种问题呢?   通过 basic.qos 方法设置prefetch_count=1 。这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message。换句话说,在接收到该Consumer的ack前,他它不会将新的Message分发给它。 设置方法如下: [python]view plaincopy 1. channel.basic_qos(prefetch_count=1)   注意,这种方法可能会导致queue满。当然,这种情况下你可能需要添加更多的Consumer,或者创建更多的virtualHost来细化你的设计。 6. 最终版本 new_task.py script: [python]view plaincopy 1. #!/usr/bin/env python   2. import pika   3. import sys   4.    5. connection = pika.BlockingConnection(pika.ConnectionParameters(   6.         host='localhost'))   7. channel = connection.channel()   8.    9. channel.queue_declare(queue='task_queue', durable=True)   10.    11. message = ' '.join(sys.argv[1:]) or "Hello World!"   12. channel.basic_publish(exchange='',   13.                       routing_key='task_queue',   14.                       body=message,   15.                       properties=pika.BasicProperties(   16.                          delivery_mode = 2, # make message persistent   17.                       ))   18. print " [x] Sent %r" % (message,)   19. connection.close()   worker.py script: [python]view plaincopy 1. #!/usr/bin/env python   2. import pika   3. import time   4.    5. connection = pika.BlockingConnection(pika.ConnectionParameters(   6.         host='localhost'))   7. channel = connection.channel()   8.    9. channel.queue_declare(queue='task_queue', durable=True)   10. print ' [*] Waiting for messages. To exit press CTRL+C'   11.    12. def callback(ch, method, properties, body):   13.     print " [x] Received %r" % (body,)   14.     time.sleep( body.count('.') )   15.     print " [x] Done"   16.     ch.basic_ack(delivery_tag = method.delivery_tag)   17.    18. channel.basic_qos(prefetch_count=1)   19. channel.basic_consume(callback,   20.                       queue='task_queue')   21.    22. channel.start_consuming()  

下载文档到电脑,查找使用更方便

文档的实际排版效果,会与网站的显示效果略有不同!!

需要 10 金币 [ 分享文档获得金币 ] 0 人已下载

下载文档