如何在Celery的Task子类定义中指定队列？

当然可以！你说的这种直接在类定义里指定队列的方式完全可行，Celery其实支持这个用法，只是官方文档可能没把这个细节放在最显眼的位置。我之前做项目的时候也遇到过同样的需求，下面给你两种靠谱的实现方式：

方法一：直接在Task子类中定义queue属性 #

你可以直接在继承celery.Task的子类里声明queue属性，这样每次调用任务时，只要不手动指定队列，就会自动使用这个默认队列：

from celery import Task

class MyTask(Task):
    # 在这里指定默认队列
    queue = 'my_queue'

    def run(self, *args, **kwargs):
        # 你的任务业务逻辑
        print(f"执行任务，参数：{args}, {kwargs}")

调用的时候就不用再写queue参数了：

# 自动发送到my_queue队列
MyTask.delay('foo', bar='baz')
# 或者用apply_async，同样不需要指定队列
MyTask.apply_async(args=('foo',), kwargs={'bar': 'baz'})

方法二：用@app.task装饰器装饰类任务 #

如果你更习惯函数式任务的写法，也可以用Celery实例的task装饰器来装饰你的任务类，直接在装饰器里指定队列，效果是一样的：

from celery import Celery, Task

app = Celery('my_project')

@app.task(name='custom_class_task', queue='my_queue')
class MyTask(Task):
    def run(self, *args, **kwargs):
        # 任务逻辑
        pass

额外小技巧：临时覆盖队列 #

如果偶尔需要把任务发到其他队列，你还是可以在调用时手动指定queue参数，这会覆盖类里定义的默认值：

# 临时发送到temp_queue队列
MyTask.apply_async(args=('foo',), queue='temp_queue')

原理其实很简单：Celery的Task基类本身就预留了queue这个属性，当调用apply_async或delay时，如果没有显式传入队列参数，Celery会依次查找任务实例、任务类的queue属性，最后才会使用全局默认队列。

内容的提问来源于stack exchange，提问作者Jahongir Rahmonov