RabbitMQ使用介绍2—Work queues

Posted 2020-11-16 venvive

Work queues

接下来是part2

在这一项中,我们创建一个工作队列，用于在多个工作者之间分配耗时的任务。

Work Queues的主要思想是，避免立即执行资源密集的任务而不得不等待其执行完成。我们将任务封装为消息并将其发送到队列中，在后台运行的一个工作进程将会弹出任务并最终执行该任务，当你管理许多工作节点时，任务就会在他们之间共享。

这个概念在web应用程序中尤其有用，因为在一个短HTTP请求窗口中不可能处理复杂的任务

下面我们发送一个特殊的String,用Thread.sleep()辅助，来模拟一些耗时的工作，用点来简单表示一个任务的复杂度，例如Hello.表示此任务需要两秒进行处理。

我们将在原来生产者的基础上修改一些代码，文件名叫send_work.py:

import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
    'localhost'))#默认端口5672,可不写

#创建通道，声明一个管道，在管道里发送消息
channel = connection.channel()
#在管道里声明queue
channel.queue_declare(queue='hello')
message = ''.join(sys.argv[1:]) or "Hello World"
#一条消息永远不能直接发送到队列,它总需要经过一个交换exchange
channel.basic_publish(exchange='',
                      routing_key='hello',
                      body=message)#设置routing_key(消息队列的名称)和body(发送的内容)
print("[x] Sent %r" % message)
connection.close()#关闭连接,队列关闭

我们将在原来消费者的基础上修改一些代码，文件名叫receive_work.py:

#receiving(消费者接收者)
import pika
import time
#创建一个连接
connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost'))#默认端口5672,可不写
#创建通道，声明一个管道，在管道里发送消息
channel = connection.channel()

#把消息队列的名字为hello,把消费者和queue绑定起来，生产者和queue的也是hello
#为什么又声明了一个hello队列
#如果确定已经声明了，可以不声明。但是你不知道那个机器先运行，所以要声明两次
channel.queue_declare(queue='hello')

#回调函数get消息体
def callback(ch,method,properties,body):#四个参数为标准格式
    #管道内存对象，内容相关信息
    print("打印看下是什么:",ch,method,properties) #打印看下是什么
    print(" [x] Received %r" % body)
    time.sleep(body.count(b'.'))
    print("[x] Done")
    
#消费消息
channel.basic_consume(
    queue='hello',#你要从那个队列里收消息
    on_message_callback=callback,#如果收到消息，就调用callback函数来处理消息
    auto_ack=True #写的话，如果接收消息，机器宕机消息就丢了
    #一般不写，宕机则生产者检测到发给其他消费者
)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming() #创建死循环，监听消息队列,可使用CTRL+C结束监听

对于上面的，开启两个控制台的消费者，开启一个生产者的控制台，运行的结果如下：

默认情况下，RabbitMQ将按顺序将每个消息发送给下一个消费者，平均每个消费者将得到相同数量的消息。这种分发消息的方式称为循环(平均分配)

消息确认(消息答复)

完成一项任务可能需要一定的时间。你可能会想，如果一个消费者开始一项长时间的任务，并且只完成了一部分，那么会发生什么。在我们当前的代码中，一旦RabbitMQ向客户发送一条消息，它立即将其标记为删除。在这种情况下。如果我们强制关闭了一个工作节点。我们将丢失它正在处理的消息。我们还将丢失发送给这个特定工作者的所有消息，但是还没有处理

通常我们不希望因为一个节点挂掉而丢失任何消息，而希望能够将这些消息传递给其他存活节点进行处理

确保消息能够不丢失，

RabbitMQ支持message acknowledgments(消息确认)，一条特定的消息被接收后，返回一个ack告诉RabbitMQ可以随意地进行删除

如果一个消费者挂了(通道关闭，TCP连接关闭)就不能发送ack给RabbitMQ,此时RabbitMQ就会意识到，某条消息没有被处理完成，那么就会将其重新发送到其他消费者。这种处理流程保证了信息不会丢失，即使偶尔有消费者挂掉

没有任何消息超时，

RabbitMQ将在某个消费者挂掉后重新传递消息，即使处理消息需要很长时间