python – 停止芹菜工人处理任务或监控Flower中的“未消耗”任务

2019年7月28日 221次阅读

假设芹菜队列上的所有任务都在使用第三方API.但是,API有一个速率限制,我正在跟踪(有一个限制和每小时限制,我需要尊重).一旦达到速率限制,我想暂停新任务的消耗,然后在我知道自己很好的时候恢复.

我通过使用以下两个任务来实现这一目标：

@celery.task()
def cancel_api_queue(minutes_to_resume):
    resume_api_queue.apply_async(countdown=minutes_to_resume*60, queue='celery')
    celery.control.cancel_consumer('third_party', reply=True)

@celery.task(default_retry_delay=300, max_retries=5)
def resume_api_queue():
    celery.control.add_consumer('third_party', destination=['y@local'])

然后我可以继续提交我的第三方API任务,并且一旦我的消费者被添加回来,我的所有任务都会被消耗掉.大.

但是,由于此队列中没有消费者,这似乎意味着我无法再看到Flower中提交的作业(直到我的消费者被添加).

有什么我做错了吗？我可以通过另一种方式实现这种“暂停”,让我继续看到提交的工作花朵吗？

附：也许这与这个问题有关,但不是100％肯定：https://github.com/celery/celery/issues/1452

我正在使用amqp经纪人,如果这有所作为.

谢谢女孩和男孩.

最佳答案 我怀疑在工人拿起它们之前偷看队列消息的内容并不是Flower的预期设计的一部分.因此,如果您停止从队列中消耗任务,那么最好的Flower可以做的是显示在“Broker”窗格中它们中有多少已作为单个数字排队.

观察传入任务内部的一种hackish方式可能是添加一个中间虚拟“转发”任务,它只是将消息从一个队列(我们称之为query_inbox)转发给另一个队列(例如,query_processing).

例如.就像是：

@celery.task(queue='query_inbox')
def query(params):
    process_query.delay(params)

@celery.task(queue='query_processing')
def process_query(params):
    ... do rate-limited stuff ...

现在您可以停止从query_processing中消耗任务,但是当它们流经query_inbox worker时,您仍然可以观察它们的参数.

点赞