同名的 Celery 任务答案

【问题标题】：Celery tasks with the same name同名的 Celery 任务
【发布时间】：2016-12-31 09:22:28
【问题描述】：

我正在实现一个文件树遍历脚本，它根据文件扩展名或目录名调用任务。

我的行走代码如下所示：

from celery import Celery
import os


app = Celery('tasks', broker='amqp://guest@localhost//')
app.conf.update(
    CELERY_DEFAULT_EXCHANGE = 'path_walker',
    CELERY_DEFAULT_EXCHANGE_TYPE = 'topic',
)   

for root, dirs, files in os.walk("/"):
    for filename in files:
        name, ext = os.path.splitext(filename)
        if ext:
            app.send_task("process", args=(os.path.join(root, filename),), routing_key="file" + ext)
    for dirname in dirs:
        app.send_task("process", args=(dirname,), routing_key="directory." + dirname)

您可以看到我正在调用相同的任务 (process)，但使用不同的 routing_keys。

在我的工人中，我有：

from celery import Celery
from kombu import Queue, Exchange
import uuid

app = Celery(broker='amqp://guest@localhost//')

file_queue = Queue(str(uuid.uuid4()), routing_key="file.py")
dir_queue = Queue(str(uuid.uuid4()), routing_key="directory.tmp")

app.conf.update(
    CELERY_DEFAULT_EXCHANGE="path_walker",
    CELERY_DEFAULT_EXCHANGE_TYPE="topic",
    CELERY_QUEUES=(
        dir_queue,
        file_queue,
    ),
)


@app.task(name="process", ignore_result=True, queue=dir_queue)
def process_dir(dir_name):
    print("Found a tmp dir: {}".format(dir_name))


@app.task(name="process", ignore_result=True, queue=file_queue)
def process_file(file_name):
    print("Found a python file: {}".format(file_name))

上面的代码创建了两个具有不同路由键的队列。然后这两个任务绑定到各个队列，但是当我运行 tree walker 时，只有第二个任务（process_file 函数）被调用。

是否可以有同名但在不同队列上的任务，由同一个工作人员运行。或者，如果我想坚持这种方法，我是否只需要每个工作人员执行一项任务？

【问题讨论】：

标签： python celery

【解决方案1】：

回答我自己的问题：

在一个 celery 应用中不可能有两个同名的任务。我可以将以上内容拆分为两个单独的应用程序并使用不同的工作人员运行它们，或者为任务提供唯一的名称。

celery 的关键代码在这里：

https://github.com/celery/celery/blob/8455b0c56797c22ba52abf59f4467ccc19eb9d20/celery/app/base.py

【讨论】：