长时间运行的芹菜工作者任务即使成功也不会返回结果

问题描述

我正在运行远程芹菜工作者，这大约需要2个小时才能完成一项任务。我使用的代理是Rabbitmq，对于后端，它是rpc。任务完成后，芹菜工作者不会将结果返回给客户。

Celery版本为4.4，操作系统为Windows

有人可以让我知道是什么问题，因为我在芹菜日志中没有看到任何错误。

pipenv run celery worker -A src.celery_app -l info -P solo

 -------------- celery@remote_host v4.4.7 (cliffs)
--- ***** -----
-- ******* ---- Windows-10-10.0.14393-SP0 2020-09-10 20:00:18
- *** --- * ---
- ** ---------- [config]
- ** ---------- .> app:         celery_app:0x2191f95bb38
- ** ---------- .> transport:   amqp://test:**@10.1.9.159:5672/test_host
- ** ---------- .> results:     rpc://
- *** --- * --- .> concurrency: 64 (solo)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** -----
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery


[tasks]
  . doors.routes.abortDoors
  . doors.routes.pushDemTodoors

[2020-09-10 20:00:18,572: INFO/MainProcess] Connected to amqp://test:**@10.1.9.159:5672/test_host
[2020-09-10 20:00:18,595: INFO/MainProcess] mingle: searching for neighbors
[2020-09-10 20:00:19,641: INFO/MainProcess] mingle: all alone
[2020-09-10 20:00:19,658: INFO/MainProcess] celery@remote_host ready.
[2020-09-10 20:00:48,533: INFO/MainProcess] Received task: doors.routes.pushDemTodoors[d175239e-dcfe-4282-8596-7be80cf725fe]

错误日志：-

Task doors.routes.pushDemTodoors[697b9f66-d67e-41c0-972f-ca845212a1fb] succeeded in 9341.906000000075s: 
[2020-09-12 01:13:01,721: CRITICAL/MainProcess] Couldn't ack 1,reason:ConnectionResetError(10054,'An existing connection was forcibly closed by the remote host',None,10054,None)
Traceback (most recent call last):
  File "c:\users\g-us01.test\.virtualenvs\celery-z5n-38vt\lib\site-packages\kombu\message.py",line 131,in ack_log_error
    self.ack(multiple=multiple)
  File "c:\users\g-us01.test\.virtualenvs\celery-z5n-38vt\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-z5n-38vt\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-z5n-38vt\lib\site-packages\amqp\abstract_channel.py",line 59,in send_method
    conn.frame_writer(1,self.channel_id,sig,args,content)
  File "c:\users\g-us01.test\.virtualenvs\celery-z5n-38vt\lib\site-packages\amqp\method_framing.py",line 189,in write_frame
    write(view[:offset])
  File "c:\users\g-us01.test\.virtualenvs\celery-z5n-38vt\lib\site-packages\amqp\transport.py",line 305,in write
    self._write(s)
ConnectionResetError: [WinError 10054] An existing connection was forcibly closed by the remote host

解决方法

您的visibility timeout设置是什么？如果您的任务运行时间超过了可见性超时设置的时间（默认情况下设置为1h），您将一无所获。要将其增加到例如4个小时，请执行类似app.conf.broker_transport_options = {"visibility_timeout": 14400}的操作。

我在celery worker命令中找不到选项-P gevent的问题。在我的情况下，造成连接重置问题的罪魁祸首是-P solo。

celery flask python rpc rpc