Ограничение выборки данных в django-celery, действующем как потребитель, из очереди rabbitmq, действующей как производитель

Я хочу получать 10 сообщений данных в каждый период периодической задачи django-celery из очереди rabbitmq, содержащей 100000 сообщений данных. все работает хорошо. но я не знаю, как я могу остановить получение данных, если 10 сообщений данных были получены в определенный период. вот мой task.py

@periodic_task(run_every=(crontab(minute='*/1')), name="task", ignore_result=True)
def task():

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

    def callback(ch, method, properties, body):
       # do sth


    channel.basic_qos(prefetch_count=10, global_qos=True)
                                                                       
    channel.basic_consume(queue='name', on_message_callback=callback)
    channel.start_consuming()

и вот мой производитель rabbitmq

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='name')
for i in range(10000000):
    channel.basic_publish(exchange='', routing_key='name', body=msg)

Вернуться на верх