1

我有 1 个 celery 经纪人和几个 celery 工人，都与 rabbitMQ 通信。在我的设置中，我向我的 celery 工人发送了几个任务，他们将处理所有任务（大约需要 1 小时），然后我将手动终止我的 celery 工人。

我想转向一个系统，如果 celery worker id 'idle'（我定义为：在 0 个时间段内有 0 个活动任务timeout_seconds，我将预先定义），则该 worker 将以编程方式终止。所有工作人员将有大约相同的任务数来运行，并且都会在同一时间“空闲”。

我设置了可以让我终止工作人员的代码，但我不确定如何检测到工作人员处于“空闲”状态并准备好终止工作。我想我想使用一个信号，但看起来没有符合我要求的信号

4

1 回答 1

1

在我工作的地方，我们的任务基本上是在做你想做的事——根据“情况”自动扩大/缩小集群。这个过程的关键是 Celery inspect/control API，所以我建议你熟悉它。这是一个没有充分记录的领域，因此请从以下内容开始：

insp = celery_app.control.inspect()
active_queues = insp.active_queues()
# Note: between these two calls some nodes may shut down and disappear
# from the dictionary so may need to deal with this...
active_stats = insp.active()

您可以在 Celery 集群运行任务时在单独的 IPython 会话中执行此操作，并查看那里有什么......

于 2020-09-01T16:01:57.467 回答