python - celery :如何限制队列中的任务数量并在满时停止喂食？

Question

我对 Celery 很陌生，这是我的问题:

假设我有一个脚本，它应该不断地从数据库中获取新数据并将其发送给使用 Celery 的工作人员。

任务.py

# Celery Task
from celery import Celery

app = Celery('tasks', broker='amqp://guest@localhost//')

@app.task
def process_data(x):
    # Do something with x
    pass

获取数据库.py

# Fetch new data from DB and dispatch to workers.
from tasks import process_data

while True:
    # Run DB query here to fetch new data from DB fetched_data

    process_data.delay(fetched_data)

    sleep(30);

我担心的是:数据每 30 秒获取一次。 process_data() 函数可能需要更长的时间，并且根据我的理解，队列可能会受到限制，这取决于工作人员的数量(特别是如果太少的话)。

1. 我不能增加 worker 的数量。
2. 我可以修改代码，避免在队列已满时向队列添加数据。

问题是如何设置队列大小以及如何知道队列已满？一般情况下，如何处理这种情况？

Answer 1

你可以设置rabbitmq x-max-length在队列中预先声明使用 kombu

例子:

import time
from celery import Celery
from kombu import Queue, Exchange

class Config(object):
    BROKER_URL = "amqp://guest@localhost//"

    CELERY_QUEUES = (
        Queue(
            'important',
            exchange=Exchange('important'),
            routing_key="important",
            queue_arguments={'x-max-length': 10}
        ),
    )

app = Celery('tasks')
app.config_from_object(Config)


@app.task(queue='important')
def process_data(x):
    pass

或使用 Policies

rabbitmqctl set_policy Ten "^one-meg$" '{"max-length-bytes":1000000}' --apply-to queues