celery 在执行特定任务后关闭工作进程

Question

我正在使用 celery（并发=1 的单池），并且我希望能够在特定任务运行后关闭工作线程。需要注意的是，我想避免工作人员在此之后承担任何进一步任务的可能性。

这是我在大纲中的尝试：

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()

有没有一种方法可以在完成特定任务后关闭工作线程，同时避免这种不幸的副作用？

Answer 1

关闭工作线程的推荐过程是发送

TERM

信号。这将导致 celery 工作线程在完成任何当前正在运行的任务后关闭。如果您向工作进程的主进程发送

QUIT

信号，工作进程将立即关闭。

但是，celery 文档通常会从从命令行或通过 systemd/initd 管理 celery 的角度来讨论这一点，但 celery 还通过

celery.app.control

提供了远程工作人员控制 API。
您可以“撤销”任务以阻止工作人员执行该任务。这应该可以防止您遇到的循环。此外，控制也以这种方式支持工作人员的“关闭”。 所以我想下面的内容会让你得到你想要的行为。

@app.task(bind=True)
def shutdown(self):
    app.control.revoke(self.id) # prevent this task from being executed again
    app.control.shutdown() # send shutdown signal to all workers

由于目前无法从任务内部确认任务，然后继续执行该任务，因此使用

revoke

 的方法规避了这个问题，这样，即使任务再次排队，新的工作人员也会忽略它。 

或者，以下方法也可以阻止重新交付的任务再次执行...

@app.task(bind=True)
def some_task(self):
    if self.request.delivery_info['redelivered']:
        raise Ignore() # ignore if this task was redelivered
    print('This should only execute on first receipt of task')

还值得注意的是

AsyncResult

还有一个 revoke

 方法可以为您调用 

self.app.control.revoke。 如果关闭worker，任务完成后，它不会再次重新排队。

Answer 2

这将在任务完成后正常关闭工作线程。

[2018-04-01 18:44:14,627: INFO/MainProcess] Connected to redis://localhost:6379/0
[2018-04-01 18:44:14,656: INFO/MainProcess] mingle: searching for neighbors
[2018-04-01 18:44:15,719: INFO/MainProcess] mingle: all alone
[2018-04-01 18:44:15,742: INFO/MainProcess] celery@foo ready.
[2018-04-01 18:46:28,572: INFO/MainProcess] Received task: celery_worker_stop.add[ac8a65ff-5aad-41a6-a2d6-a659d021fb9b]
[2018-04-01 18:46:28,585: INFO/ForkPoolWorker-4] Task celery_worker_stop.add[ac8a65ff-5aad-41a6-a2d6-a659d021fb9b] succeeded in 0.005628278013318777s: 3   
[2018-04-01 18:46:28,665: WARNING/MainProcess] Got shutdown from remote

注意：广播将关闭所有工作人员。如果您想关闭特定的工作程序，请使用名称启动工作程序

celery -A celeryapp  worker -n self_killing --concurrency=1 --pool=solo

现在您可以使用目标参数关闭它。

app.control.broadcast('shutdown', destination=['celery@self_killing'])

如果您需要关闭特定的worker并且事先不知道它的名称，您可以从任务属性中获取它。根据上面的答案，您可以使用：

Answer 3

或

app.control.broadcast('shutdown', destination=[self.request.hostname])

注：

工人应该以名称开始（选项'-n'

）；

• 任务应使用

  bind=True

  参数定义。
添加到

sytech

Answer 4

answer

，如果您使用注释@shared_task而不是@app.task，或者/并且您不想使用

(bind=True)

，则可以使用此选项而不使用

self 

参数：

@shared_task
def some_function_name():
    app.control.revoke(some_function_name.request.id)
    app.control.shutdown()

参考：

https://stackoverflow.com/a/18876650/1856563