RabbitMQ 任务分发机制

zoukankan html css js c++ java

RabbitMQ 任务分发机制
   在上篇文章中，我们解决了从发送端（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 plain copy

print ?

import sys



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

channel.basic_publish(exchange='',

                      routing_key='hello',

                      body=message)

print " [x] Sent %r" % (message,)
```
import sys

message = ' '.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(exchange='',
                      routing_key='hello',
                      body=message)
print " [x] Sent %r" % (message,)
```
同样的道理，把receive.py的名字换成worker.py，并且根据Message中的.的数量进行计算密集型模拟：
[python] view plain copy

print ?

import time



def callback(ch, method, properties, body):

    print " [x] Received %r" % (body,)

    time.sleep( body.count('.') )

    print " [x] Done"
```
import time

def callback(ch, method, properties, body):
    print " [x] Received %r" % (body,)
    time.sleep( body.count('.') )
    print " [x] Done"
```
2. Round-robin dispatching 循环分发

RabbitMQ的分发机制非常适合扩展，而且它是专门为并发程序设计的。如果现在load加重，那么只需要创建更多的Consumer来进行任务处理即可。当然了，对于负载还要加大怎么办？我没有遇到过这种情况，那就可以创建多个virtual Host，细化不同的通信类别了。

首先开启两个Consumer，即运行两个worker.py。

Console1：
[python] view plain copy

print ?

shell1$ python worker.py

[*] Waiting for messages. To exit press CTRL+C
```
shell1$ python worker.py
 [*] Waiting for messages. To exit press CTRL+C
```
Consule2：
[python] view plain copy

print ?

shell2$ python worker.py

[*] Waiting for messages. To exit press CTRL+C
```
shell2$ python worker.py
 [*] Waiting for messages. To exit press CTRL+C
```
Producer new_task.py要Publish Message了：
[python] view plain copy

print ?

shell3$ python new_task.py First message.

shell3$ python new_task.py Second message..

shell3$ python new_task.py Third message...

shell3$ python new_task.py Fourth message....

shell3$ python new_task.py Fifth message.....
```
shell3$ python new_task.py First message.
shell3$ python new_task.py Second message..
shell3$ python new_task.py Third message...
shell3$ python new_task.py Fourth message....
shell3$ python new_task.py Fifth message.....
```
注意一下：.代表的sleep(1)。接着开一下Consumer worker.py收到了什么:

Console1：
[python] view plain copy

print ?

shell1$ python worker.py

[*] Waiting for messages. To exit press CTRL+C

[x] Received 'First message.'

[x] Received 'Third message...'

[x] Received 'Fifth message.....'
```
shell1$ python worker.py
 [*] Waiting for messages. To exit press CTRL+C
 [x] Received 'First message.'
 [x] Received 'Third message...'
 [x] Received 'Fifth message.....'
```
Console2：
[python] view plain copy

print ?

shell2$ python worker.py

[*] Waiting for messages. To exit press CTRL+C

[x] Received 'Second message..'

[x] Received 'Fourth message....'
```
shell2$ python worker.py
 [*] Waiting for messages. To exit press CTRL+C
 [x] Received 'Second message..'
 [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 plain copy

print ?

def callback(ch, method, properties, body):

    print " [x] Received %r" % (body,)

    time.sleep( body.count('.') )

    print " [x] Done"

    ch.basic_ack(delivery_tag = method.delivery_tag)



channel.basic_consume(callback,

                      queue='hello')
```
def callback(ch, method, properties, body):
    print " [x] Received %r" % (body,)
    time.sleep( body.count('.') )
    print " [x] Done"
    ch.basic_ack(delivery_tag = method.delivery_tag)

channel.basic_consume(callback,
                      queue='hello')
```
这样即使你通过Ctr-C中断了worker.py，那么Message也不会丢失了，它会被分发到下一个Consumer。

如果忘记了ack，那么后果很严重。当Consumer退出时，Message会重新分发。然后RabbitMQ会占用越来越多的内存，由于RabbitMQ会长时间运行，因此这个“内存泄漏”是致命的。去调试这种错误，可以通过一下命令打印un-acked Messages：
[python] view plain copy

print ?

$ sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged

Listing queues ...

hello    0       0

...done.
```
$ sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged
Listing queues ...
hello    0       0
...done.
```
4. Message durability消息持久化

在上一节中我们知道了即使Consumer异常退出，Message也不会丢失。但是如果RabbitMQ Server退出呢？软件都有bug，即使RabbitMQ Server是完美毫无bug的（当然这是不可能的，是软件就有bug，没有bug的那不叫软件），它还是有可能退出的：被其它软件影响，或者系统重启了，系统panic了。。。

为了保证在RabbitMQ退出或者crash了数据仍没有丢失，需要将queue和Message都要持久化。

queue的持久化需要在声明时指定durable=True：
[python] view plain copy

print ?

channel.queue_declare(queue='hello', durable=True)
```
channel.queue_declare(queue='hello', durable=True)
```
上述语句执行不会有什么错误，但是确得不到我们想要的结果，原因就是RabbitMQ Server已经维护了一个叫hello的queue，那么上述执行不会有任何的作用，也就是hello的任何属性都不会被影响。这一点在上篇文章也讨论过。

那么workaround也很简单，声明一个另外的名字的queue，比如名字定位task_queue：
[python] view plain copy

print ?

channel.queue_declare(queue='task_queue', durable=True)
```
channel.queue_declare(queue='task_queue', durable=True)
```
再次强调，Producer和Consumer都应该去创建这个queue，尽管只有一个地方的创建是真正起作用的：

接下来，需要持久化Message，即在Publish的时候指定一个properties，方式如下：
[python] view plain copy

print ?

channel.basic_publish(exchange='',

                      routing_key="task_queue",

                      body=message,

                      properties=pika.BasicProperties(

                         delivery_mode = 2, # make message persistent

                      ))
```
channel.basic_publish(exchange='',
                      routing_key="task_queue",
                      body=message,
                      properties=pika.BasicProperties(
                         delivery_mode = 2, # make message persistent
                      ))
```
关于持久化的进一步讨论：

为了数据不丢失，我们采用了：
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 plain copy

print ?

channel.basic_qos(prefetch_count=1)
```
channel.basic_qos(prefetch_count=1)
```
注意，这种方法可能会导致queue满。当然，这种情况下你可能需要添加更多的Consumer，或者创建更多的virtualHost来细化你的设计。

6. 最终版本

new_task.py script:
[python] view plain copy

print ?

#!/usr/bin/env python

import pika

import sys



connection = pika.BlockingConnection(pika.ConnectionParameters(

        host='localhost'))

channel = connection.channel()



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



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

channel.basic_publish(exchange='',

                      routing_key='task_queue',

                      body=message,

                      properties=pika.BasicProperties(

                         delivery_mode = 2, # make message persistent

                      ))

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

connection.close()
```
#!/usr/bin/env python
import pika
import sys

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

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

message = ' '.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(exchange='',
                      routing_key='task_queue',
                      body=message,
                      properties=pika.BasicProperties(
                         delivery_mode = 2, # make message persistent
                      ))
print " [x] Sent %r" % (message,)
connection.close()
```
worker.py script:
[python] view plain copy

print ?

#!/usr/bin/env python

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

    print " [x] Done"

    ch.basic_ack(delivery_tag = method.delivery_tag)



channel.basic_qos(prefetch_count=1)

channel.basic_consume(callback,

                      queue='task_queue')



channel.start_consuming()
```
#!/usr/bin/env python
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('.') )
    print " [x] Done"
    ch.basic_ack(delivery_tag = method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
                      queue='task_queue')

channel.start_consuming()
```
转载自：anzhsoft http://blog.csdn.net/anzhsoft/article/details/19607841

参考资料：

1. http://www.rabbitmq.com/tutorials/tutorial-two-python.html
查看全文

相关阅读:
Spark Scala 读取GBK文件的方法
 Mac OS X 系统下自带的文本文件格式转换工具iconv
报到
 java 字符串中含有双引号" "与单引号' '问题
 div1嵌套div2，div2居中的解决办法
 ionic4 创建 angular项目 ReactNative下载第三方库出错解决Error: EPERM: operation not permitted, rename
ionic+cordova 创建项目+打包
 jxl读取excel文件异常：Unable to recognize OLE stream 的解决方法
 学习 javascript （一）javascript 简介
 学习 JavaScript （四）核心概念：操作符

原文地址：https://www.cnblogs.com/liyongbin/p/8192331.html

RabbitMQ 任务分发机制

1. 准备

2. Round-robin dispatching 循环分发

3. Message acknowledgment 消息确认

4. Message durability消息持久化

5. Fair dispatch 公平分发

6. 最终版本