python - celery worker - 连接已关闭错误

Question

我正在使用 Flask 和远程 celery worker，为了 celery 通信，我使用 rabbitmq 作为消息代理。远程 celery worker 随机抛出如下错误:-

[2020-09-03 13:49:59,390: CRITICAL/MainProcess] Couldn't ack 20, reason:RecoverableConnectionError(None, 'connection already closed', None, '')
Traceback (most recent call last):
  File "c:\users\g-us01.test\.virtualenvs\celery_doors_integration-o4w-mxzx\lib\site-packages\kombu\message.py", line 131, in ack_log_error
    self.ack(multiple=multiple)
  File "c:\users\g-us01.test\.virtualenvs\celery_doors_integration-o4w-mxzx\lib\site-packages\kombu\message.py", line 126, in ack
    self.channel.basic_ack(self.delivery_tag, multiple=multiple)
  File "c:\users\g-us01.test\.virtualenvs\celery_doors_integration-o4w-mxzx\lib\site-packages\amqp\channel.py", line 1394, in basic_ack
    spec.Basic.Ack, argsig, (delivery_tag, multiple),
  File "c:\users\g-us01.test\.virtualenvs\celery_doors_integration-o4w-mxzx\lib\site-packages\amqp\abstract_channel.py", line 56, in send_method
    raise RecoverableConnectionError('connection already closed')
amqp.exceptions.RecoverableConnectionError: connection already closed

我正在使用 celery 版本 4。有关如何避免此错误的任何指示都会有所帮助。

Answer 1

在运行非常长的任务时，我在使用 celery 4.4.6 版和 rabbitmq 时遇到了同样的问题。然后我使用以下配置更改运行相同的任务，现在它可以工作了(我以单人模式运行工作人员)。重要的配置似乎是代理心跳:https://www.rabbitmq.com/heartbeats.html .这应该会禁用心跳，并且不应因错过的心跳而重置连接。

CELERY_BROKER_HEARTBEAT = 0

celery 文档链接:https://docs.celeryproject.org/en/v4.4.6/userguide/configuration.html#std:setting-broker_heartbeat

与 Flask 的集成应该像这样工作:

from flask import Flask
from celery import Celery

app = Flask(__name__)
app.config['CELERY_BROKER_URL'] =
                            'amqp://myuser:mypassword@localhost:5672/myvhost'
app.config['CELERY_BROKER_HEARTBEAT'] = 0

celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)

我可以在这里推荐这个博客(代码片段的来源):https://blog.miguelgrinberg.com/post/using-celery-with-flask