Prefetch count--预取数量

一、前言

前面提到如果有多个消费者同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者。这时如果每个消息的处理时间不同，就有可能会导致某些消费者一直在忙，而另外一些消费者很快就处理完手头工作并一直空闲的情况。我们可以通过设置prefetchCount来限制Queue每次发送给每个消费者的消息数，比如我们设置prefetchCount=1，则Queue每次给每个消费者发送一条消息；消费者处理完这条消息后Queue会再给该消费者发送一条消息。

二、事例

生产端：

# -*- coding: UTF-8 -*-

import pika

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

channel = connection.channel()

# 声明队列，并进行队列持久化
channel.queue_declare(queue='task_queue', durable=True)

# 信息内容
message = "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()

消费端：

# -*- coding: UTF-8 -*-

import pika
import time
import random

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(random.randint(1, 20))
    print('[x] Done')
    # 当消息处理完成后，主动通知rabbitmq，之后rabbitmq才会删除
    # 队列中的这条消息
    ch.basic_ack(delivery_tag=method.delivery_tag)

# prefetch_count = 1 如果消费者中有一条消息没有
# 处理完，就不会继续给这个消费者发送新消息
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
                      queue='task_queue')

# 永远收下去，没有就在这卡住
channel.start_consuming()

可以运行多个consumer，让它们不断接受消息，可以看到只有当consumer中的消息处理完成了，才会接受下一个消息。