我正在寻找一种方法来对每个芹菜工人执行特定类型的任务。确切的用例是一个周期性的健康作业，它确保成功的任务初始化和进展的各种先决条件（这些指标报告给不同的服务）。例如，确保可以建立与 DB 的连接。
我发现远程控制和检查命令可用于该目的（有一些固定的调度），但当 AWS SQS 用作后端代理时，它们不受支持。
知道如何在不向分叉进程任务添加任何内存占用的情况下实现这一目标吗？也许通过在工作进程中启动另一个线程？
【问题讨论】：
为了解决这个问题，我使用了Celery custom worker bootstep。它在启动时注册一个计划任务，每 X 秒将一个健康检查任务省略到工作执行池中。
此解决方案与后端代理无关，并利用定制的工作执行池。
class WorkerHealthMonitor(bootsteps.StartStopStep):
requires = {'celery.worker.components:Timer',
'celery.worker.components:Pool'}
def __init__(self, worker, **kwargs):
self.tref = None
self.interval = 60
def start(self, worker):
logger.info("Registering health monitor timer with %d seconds interval", self.interval)
self.tref = worker.timer.call_repeatedly(
self.interval, schedule_health_check, (worker,), priority=10,
)
def stop(self, worker):
if self.tref:
self.tref.cancel()
self.tref = None
def schedule_health_check(worker):
worker.pool.apply_async(health_check, callback=health_check_completed)
def health_check(**kwargs):
logger.info('Running Health Check...')
return 'I am alive'
def health_check_completed(result):
logger.info('Health check completed with msg: %s', result)
任务注册：
app = Celery('tasks', broker=BROKER_URL, backend=BACKEND_URL)
app.steps['worker'].add(WorkerHealthMonitor)
【讨论】：
虽然您可以这样做（我将在下面进一步解释），但我建议您使用出色的 Celery monitoring and management API。 status 命令（以某种方式进行基本的健康检查）就是以这种方式实现的示例，因此您应该将其用作您自己实现的基础。
在了解“检查/控制 API”（我就是这么称呼它）之前，我通过为每个工作人员分配一个唯一的队列名称来实现类似于您想要做的事情。然后，您的“向所有工作人员发送任务”将遍历所有唯一的队列名称（例如 worker_1、worker_2 等），并将任务发送到这些队列中的每一个。
【讨论】：