Celery/RabbitMQ unacked消息阻塞队列？

Question

我已经调用了一个任务,用urllib2远程获取一些信息几千次.这些任务是使用随机eta(一周内)安排的,因此它们都不会同时命中服务器.有时我会得到404,有时候不会.我正在处理错误,以防它发生.

在RabbitMQ控制台中,我可以看到16条未确认的消息:

我停下芹菜,清理队列并重新启动它.16条未经确认的消息仍在那里.

我有其他任务进入同一队列,也没有执行任何任务.清除之后,我尝试提交另一项任务,状态仍然准备就绪:

我有什么想法可以找出为什么消息仍未被承认？

版本:

celery==3.1.4
{rabbit,"RabbitMQ","3.5.3"}

celeryapp.py

CELERYBEAT_SCHEDULE = {
    'social_grabber': {
        'task': '<django app>.tasks.task_social_grabber',
        'schedule': crontab(hour=5, minute=0, day_of_week='sunday'),
    },
}

tasks.py

@app.task
def task_social_grabber():
    for user in users:
        eta = randint(0, 60 * 60 * 24 * 7) #week in seconds
        task_social_grabber_single.apply_async((user), countdown=eta)

没有定义此任务的路由,因此它进入默认队列:celery.有一个工作人员处理此队列.

supervisord.conf:

[program:celery]
autostart = true
autorestart = true
command = celery worker -A <django app>.celeryapp:app --concurrency=3 -l INFO -n celery

Answer 1

RabbitMQ 在 3.3 版本中破坏了 QoS 设置。您需要将 celery 至少升级到 3.1.11 ( changelog )，将 kombu 至少升级到 3.0.15 ( changelog )。您应该使用最新版本。

当 3.3 发布时，我遇到了完全相同的行为。RabbitMQ 翻转了 prefetch_count 标志的默认行为。在此之前，如果消费者在 etad 消息中达到了 CELERYD_PREFETCH_MULTIPLIER 限制，则工作线程将提高此限制以获取更多消息。该更改破坏了此行为，因为新的默认行为拒绝了此功能。