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

Question

我是新来的芹菜，有一个问题。我有这个简单的任务：

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

我稍后会在类似的测试用例中调用此任务

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

任务在队列 @9​​87654325@ 中成功运行（因为我在 celery 日志中看到它，并且它完成得很好。但我想知道一种以编程方式查找任务 test_install_queue 运行在哪个队列中的方法，来自result中存储的对象。

谢谢！

编辑：

我已将任务更改为：

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

然后我使用apply_async 的结果如下：

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

解决方法是工作人员（主机名）与工作人员中初始化的唯一队列具有相同的名称。

Answer 1

您可以尝试以下方法：

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

这种方法是建立在假设您使用默认 celery 设置并且您的交换是直接的。如果您需要更通用的扇出和主题交换解决方案，则必须检查 app.amqp.queues 中每个已声明队列的路由键。

Answer 2

我自己刚刚遇到了这个问题，我真的很怀疑是否需要一个复杂的解决方案，比如来自“lexabug”的一个已经被接受的解决方案...... 因此，由于即使 Celery 文档也没有提供有效的替代方案，我自己使用反射进行了调查，以了解哪个对象包含我需要的信息，我想出了一个超级简单直接的解决方案。具体来说，我正在用 Celery 术语编写一个钩子，或者更好的是一个信号，这就是我根据任务名称检索队列名称的方式：

    @signals.after_task_publish.connect()
    def on_task_publish(sender=None, headers=None, body=None, **kwargs):

        # "sender" is a string containing task name 
        # ("celery" here is the celery app)
        task: Task = celery.tasks.get(sender)

        # once we have the task object, we can access the "queue" property 
        # which contains the name of the queue 
        # (it' a dynamic property so don't expect support by your IDE)
        queue_name: str = task.queue if task is not None else 'unknown'

ps。我正在使用 Celery 4.4