如何获取运行任务的队列-celery

问题描述

我是新来的芹菜人，有一个问题.我有一个简单的任务:

I'm new using celery and have a question. I have this simple task:

@app.task(name='test_install_queue')
def test_install_queue():
    return subprocess.call("exit 0",shell=True)

我稍后会在一个测试用例中调用此任务

and I am calling this task later in a test case like

result = tasks.test_default_queue.apply_async(queue="install")

该任务在队列 install 中成功运行(因为我在celery日志中看到了该任务，并且可以正常完成.但是我想以编程方式知道在哪个队列中找到任务 test_install_queue 从存储在结果中的对象运行.

The task run successfully in the queue install (because I am seeing it in the celery log, and it completes fine. But I would like to know a programmatically way of finding in which queue was the task test_install_queue run, from the object stored in result.

谢谢！

我将任务更改为:

@app.task(name='test_install_queue',bind=True)
def test_install_queue(self):
    return self.request.__dict__

然后我使用 apply_async 的结果，如下所示:

and then I'm using the result of apply_async as follows:

result = tasks.test_install_queue.apply_async(queue="install")
assert "install" in result.get()["hostname"]

，解决方法是该工作程序(主机名)与在该工作程序中初始化的唯一队列的名称相同.

and the workaround is that the worker (hostname) has the same name as the only queue that is initialized in the worker.

推荐答案

您可以尝试以下方法:

delivery_info = app.current_task.request.delivery_info
# by default celery uses the same name for queues and exchanges
original_queue = delivery_info['exchange']
for queue in app.amqp.queues.itervalues():
    if queue.exchange.name == delivery_info['exchange']
        and queue.routing_key == delivery_info['routing_key']:
            original_queue = queue.name
            break

该方法基于以下假设:您使用默认的芹菜设置，并且您的交换是直接的.如果您需要用于扇出和主题交换的通用解决方案，则必须检查 app.amqp.queues 中每个已声明队列的路由键.

That approach is built on assumption that you use default celery settings and your exchanges are direct. If you need more universal solution for fanout and topic exchanges then you will have to check routing keys of every declared queue in app.amqp.queues.

这篇关于如何获取运行任务的队列-celery的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！