python - celery 长时间不活动后意外关闭

Question

所以我使用 RabbitMQ + Celery 来创建一个简单的 RPC 架构。我有一个 RabbitMQ 消息代理和一个运行 Celery 守护进程的远程 worker 。

还有第三台服务器公开了一个瘦 RESTful API。当它收到 HTTP 请求时，它会向远程工作人员发送任务，等待响应并返回响应。

这在大多数情况下都非常有效。但是我注意到，在长时间不活动后(比如 5 分钟没有传入请求)，Celery worker 的行为很奇怪。长时间不活动后收到的前 3 个任务返回此错误:

exchange.declare: connection closed unexpectedly

在三个错误的任务之后它再次工作。如果长时间没有任务，也会发生同样的事情。有什么想法吗？

Celery worker 的初始化脚本:

# description "Celery worker using sync broker"

console log

start on runlevel [2345]
stop on runlevel [!2345]

setuid richard
setgid richard

script
chdir /usr/local/myproject/myproject
exec /usr/local/myproject/venv/bin/celery worker -n celery_worker_deamon.%h -A proj.sync_celery -Q sync_queue -l info --autoscale=10,3 --autoreload --purge
end script

respawn

我的 celery 配置:

# Synchronous blocking tasks
BROKER_URL_SYNC = 'amqp://guest:guest@localhost:5672//'
# Asynchronous non blocking tasks
BROKER_URL_ASYNC = 'amqp://guest:guest@localhost:5672//'

#: Only add pickle to this list if your broker is secured
#: from unwanted access (see userguide/security.html)
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'UTC'
CELERY_ENABLE_UTC = True
CELERY_BACKEND = 'amqp'

# http://docs.celeryproject.org/en/latest/userguide/tasks.html#disable-rate-limits-if-they-re-not-used
CELERY_DISABLE_RATE_LIMITS = True

# http://docs.celeryproject.org/en/latest/userguide/routing.html
CELERY_DEFAULT_QUEUE = 'sync_queue'
CELERY_DEFAULT_EXCHANGE = "tasks"
CELERY_DEFAULT_EXCHANGE_TYPE = "topic"
CELERY_DEFAULT_ROUTING_KEY = "sync_task.default"
CELERY_QUEUES = {
    'sync_queue': {
        'binding_key':'sync_task.#',
    },
    'async_queue': {
        'binding_key':'async_task.#',
    },
}

有什么想法吗？

编辑:

好的，现在它似乎是随机发生的。我在 RabbitMQ 日志中注意到了这一点:

=WARNING REPORT==== 6-Jan-2014::17:31:54 ===
closing AMQP connection <0.295.0> (some_ip_address:36842 -> some_ip_address:5672):
connection_closed_abruptly

Answer 1

你的 RabbitMQ 服务器或你的 Celery worker 是否有机会在负载均衡器后面？如果是，则负载平衡器在一段时间不活动后关闭 TCP 连接。在这种情况下，您将必须从客户端(工作人员)端启用心跳。如果这样做，我不建议为此使用纯 Python amqp 库。相反，将其替换为 librabbitmq .