rabbitmq - 芹菜工人从 systemd 运行的 celerybeat 接收未注册的任务

问题描述

在我的登台服务器上，我让我的 celery worker (4.3.0) 通过 systemd 启动并运行 celery beat 作为守护进程，RabbitMQ 作为代理。几周前一切都很好，直到 4 天前的那一刻，当 celery 和 amqp 之间通过 kombu 出现某种连接错误时。[Errno 104] Connection reset by peer after started

由于该项目处于 WiP 阶段，因此我并没有过多关注服务器日志，但是当我尝试部署最新版本的代码时，我意识到工作人员出了点问题。

我用谷歌搜索了这个问题，这就是弹出的内容： https ://github.com/celery/celery/issues/4867

简单的解决方案是将 celery 降级到 4.1.1 并等待在未来的稳定版本中修复。

我从我的 venv 中删除了 celery、amqp、billiard 和 kombu，安装了 celery.4.1.1，它以适当的版本安装了上述软件包。

celery 和 celerybeat 的 atm 服务处于活动状态，celerybeat 将任务发送给 celery worker，但是 celery 日志显示错误消息（请参阅降级后的 celery 错误代码）。这很奇怪，因为我没有更改任务声明或我的设置中的任何内容（这可能是这里的问题）。

最奇怪的是，如果我关闭 systemd 服务并使用以下命令运行它们：
celery -A celery_cfg:app worker -B --loglevel=DEBUG

当前的所有任务都像过去一样进行。所以 celery 和 celerybeat 的配置似乎可以正常工作。

我尝试了一些有针对性的方法：
1）确保导入所有模块而不导入亲属。
2) 过去遇到过 venv 中缺少软件包的问题 --> 它们是最新的
3) 重新启动 celery/celerybeat/gunicorn/systemd/rabbitmq 和服务器本身
4) 仔细检查了 systemd 服务中的路径（但也许我正在调试这太长了，我只是看不到错字什么的）
5）尝试使用开发版本4.4.0rc2，（芹菜工人不会站起来）6）安装的应用程序包含所有必需的应用程序

celery 版本降级后的错误信息

    `2019-06-16 19:35:00,092: ERROR/MainProcess] Received unregistered task of type 'apps.mailing.tasks.execute_sending_system_mail'.
The message has been ignored and discarded.

Did you remember to import the module containing this task?
Or maybe you're using relative imports?

Please see
http://docs.celeryq.org/en/latest/internals/protocol.html
for more information.

The full contents of the message body was:
'[[], {}, {"callbacks": null, "errbacks": null, "chain": null, "chord": null}]' (77b)
Traceback (most recent call last):
 File "/home/user/apps/venv/loans/lib/python3.7/site-packages/celery/worker/consumer/consumer.py", line 557, in on_task_received
   strategy = strategies[type_]
KeyError: 'apps.mailing.tasks.execute_sending_system_mail'

芹菜服务系统代码

Description=Celery Service
After=network.target

[Service]
Type=forking
User=<user>
Group=<user>
EnvironmentFile=/etc/default/celery
WorkingDirectory=/home/<user>/apps/loans
ExecStart=/bin/sh -c '${CELERY_BIN} multi start ${CELERYD_NODES} \
 -A ${CELERY_APP} --pidfile=${CELERYD_PID_FILE} \
 --logfile=${CELERYD_LOG_FILE} --loglevel=${CELERYD_LOG_LEVEL} ${CELERYD_OPTS}'
ExecStop=/bin/sh -c '${CELERY_BIN} multi stopwait ${CELERYD_NODES} \
 --pidfile=${CELERYD_PID_FILE}'
ExecReload=/bin/sh -c '${CELERY_BIN} multi restart ${CELERYD_NODES} \
 -A ${CELERY_APP} --pidfile=${CELERYD_PID_FILE} \

Celery Beat 服务系统代码

Description=Celery Beat Service
After=network.target

[Service]
Type=simple
User=user
Group=user
EnvironmentFile=/etc/default/celery
WorkingDirectory=/home/user/apps/loans
ExecStart=/bin/sh -c '${CELERY_BIN} beat  \
  -A ${CELERY_APP} --pidfile=${CELERYBEAT_PID_FILE} \
  --logfile=${CELERYBEAT_LOG_FILE} --loglevel=${CELERYD_LOG_LEVEL}'

[Install]
WantedBy=multi-user.target

变量配置文件

CELERYD_NODES="w1"

CELERY_BIN="/home/user/apps/venv/loans/bin/celery"

CELERY_APP="celery_cfg:app"

CELERYD_MULTI="multi"

CELERYD_OPTS=""

CELERYD_PID_FILE="/home/user/apps/pids/celery/%n.pid"
CELERYD_LOG_FILE="/home/user/apps/logs/celery/%n%I.log"
CELERYD_LOG_LEVEL="INFO"

CELERYBEAT_PID_FILE="/home/user/apps/pids/celery/beat.pid"
CELERYBEAT_LOG_FILE="/home/user/apps/logs/celery/beat.log"

celery_cfg 文件



app = Celery('loans_apps')

app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
app.set_default()

# <====CELERY BEAT PERIODIC TASKS ====>
app.conf.beat_schedule = {
    'execute_sending_system_mail': {
        'task': 'apps.mailing.tasks.execute_sending_system_mail',
        'schedule': crontab(minute='*/5'),
        'args': (),
    },
}


@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

包含 celery cfg 变量的设置的小部分

BROKER_URL = 'amqp://localhost//',
CELERY_ENABLE_UTC = True

我知道我可以尝试在没有 systemd 的情况下设置 celery 和 celerybeat，但是我将此视为最后的解决方案。我想保持原样，即使我不知道那里出了什么问题。

编辑由于错误并在我朋友的指导下我刚刚发现，celery 和 celerybeat 服务似乎在用户 root 上运行良好，这显然不是解决方案，但可以减少可能的缺陷数量

标签： rabbitmqcelerysystemddjango-celerycelerybeat

rabbitmq - 芹菜工人从 systemd 运行的 celerybeat 接收未注册的任务

问题描述

解决方案

推荐阅读