Celery/Django 工作人员检测何时没有工作

Question

我有一个 Celery/Django worker 通过 RabbitMQ 连接到服务器。当工作人员完成一项工作时，如果没有剩余工作，我希望它终止 - 我如何检查队列中是否没有剩余工作？

Answer 1

任务完成时杀死 pid throw psutil.

例如：

import psutil
import os

@celery.task
def my_task():
    pid=os.getpid() # get the worker pid
    # your code
    return pid # or store it somewhere

def task_caller()
    result = my_task.apply()
    if no_more_jobs('my_queue'): 
        kill_worker(result)
    
def kill_worker(pid):
    try:
        proc = psutil.Process(pid=pid)
        for child in proc.children(recursive=True):
            child.kill()
        proc.kill()
        return True
    except Exception:
        # manage exception
        return False

def no_more_jobs(queue):
    # edit below params
    connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
    channel = connection.channel()
    q = channel.queue_declare(queue)
    return q.method.message_count == 0

注意：这是一个基本示例，需要根据您的生产者/消费者逻辑进行编辑