如何在芹菜中将任务从一个队列移动到另一个队列

【问题标题】：How do I move tasks from one queue to another in celery如何在芹菜中将任务从一个队列移动到另一个队列
【发布时间】：2017-02-08 14:07:03
【问题描述】：

我需要能够做两件事，这两件事都取决于我能否将任务从一个队列移动到另一个队列。：

当任务重试失败一定次数时，将其移至另一个队列
当任务满足特定条件时，将其移动到不同的队列（由不同的工作人员处理）

请注意，我的意思是将工作人员的一项任务中的内容放入另一个队列中 - 而不是来自主应用程序。

我能找到的唯一一段代码在这里被引用 - https://*.com/a/27144119/112050 有人可以指出正确的 api 吗？

【问题讨论】：

对于案例“2”。上面，您打算何时/如何检测“任务满足某些条件”？它来自正在运行的任务吗？它是在相关任务运行之前来自另一个任务吗？您能否澄清或举例说明您在案例“2”中想要什么？

标签： python redis rabbitmq celery

【解决方案1】：

您移动任务的想法实际上归结为使用相同的参数运行相同的任务，但将其发送到不同的队列。

apply_async 有一个queue 参数

from celery.exceptions import MaxRetriesExceededError

@shared_task(default_retry_delay = 1 * 60, max_retries = 10)
def some_task(arg1, arg2):
    try:
        # task logic

        if some_condition:
            some_task.apply_async([arg1, arg2], queue='different_queue')
            return

    except MaxRetriesExceededError:
        some_task.apply_async([arg1, arg2], queue='different_queue')

    except Exception, exc:
        raise some_task.retry(exc=exc)

如果你真的想在队列之间移动任务，假设你使用的是 RabbitMQ，你可以使用Shovel Plugin。例如将消息从队列q1 移动到本地代理上的队列q2：

rabbitmqctl set_parameter shovel my-shovel \
'{"src-uri": "amqp://", "src-queue": "q1", \
 "dest-uri": "amqp://", "dest-queue": "q2"}'

【讨论】：