我正在使用celery(使用concurrency=1的solo池),我希望能够在特定任务运行后关闭worker。但需要注意的是,我希望避免工作线程在完成该任务后接收到其他任务。
from __future__ import absolute_import, unicode_literals
from celery import Celery
from celery.exceptions import WorkerShutdown
from celery.signals import task_postrun
app = Celery()
app.config_from_object('celeryconfig')
@app.task
def add(x, y):
return x + y
@task_postrun.connect(sender=add)
def shutdown(*args, **kwargs):
raise WorkerShutdown()
然而,当我运行工作线程时
celery -A celeryapp worker --concurrency=1 --pool=solo
并运行该任务
add.delay(1,4)
我得到以下内容:
-------------- celery@sam-APOLLO-2000 v4.0.2 (latentcall)
---- **** -----
--- * *** * -- Linux-4.4.0-116-generic-x86_64-with-Ubuntu-16.04-xenial 2018-03-18 14:08:37
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app: __main__:0x7f596896ce90
- ** ---------- .> transport: redis://localhost:6379/0
- ** ---------- .> results: redis://localhost/
- *** --- * --- .> concurrency: 4 (solo)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** -----
-------------- [queues]
.> celery exchange=celery(direct) key=celery
[2018-03-18 14:08:39,892: WARNING/MainProcess] Restoring 1 unacknowledged message(s)
任务被重新排队,将在另一个工作进程上再次运行,从而形成循环。
当我把WorkerShutdown
异常放在任务本身中时,也会发生这种情况。
@app.task
def add(x, y):
print(x + y)
raise WorkerShutdown()
有没有一种方法可以在特定任务完成后关闭worker,同时避免不良副作用?
os.kill(os.getpid(), signal.SIGTERM)
看看是否有帮助。请尝试您已经尝试过的两种方法。 - Tarun Lalwanicelery
进程的子进程,则应尝试使用os.getppid()
。 - Tarun Lalwani