CELERY_ROUTES - 如何根据任务名称进行路由

Question

我试图让 celery 根据任务名称路由任务...基本上，我有名为“worker.some_name”和“web.some_name”的任务，并且我使用两个不同的队列，称为worker和分别是网络。我希望所有工作任务都进入工作队列，反之亦然。目前我有一个像这样的大 CELERY_ROUTES 字典：

CELERY_ROUTES = {
    "web.some_name": {
        "queue": "web"
    },
    "web.some_other_name": {
        "queue": "web"
    },
    etc.... }

但我想要更通用的东西，例如：

CELERY_ROUTES = (MyRouter(), ) 
class MyRouter(object):
    def route_for_task(self, task, args=None, kwargs=None):
        if task.split('.')[0] == "worker":
            return {"queue": "worker"}
        return {"queue": "web"}

但这似乎不起作用。有什么想法吗？谢谢。

Answer 1

您必须对 py 文件中定义的任务使用装饰器“@app.task”。

您可以使用 @app.task(queue='queue_name') 路由您的任务

Answer 2

您应该能够通过将交换类型从直接更改为主题来完成您想要的操作。通过这种方式，您可以将任务指定为 web.* 或 worker.*

您可以在此处阅读：http://ask.github.com/celery/userguide/routing.html#topic-exchanges

Answer 3

Celery 3.x 默认不支持通配符路由，但你可以自己实现。

这是一个复制粘贴解决方案：

class TaskRouter:
    def __init__(self, routes):
        self.routes = {}
        self.glob_routes = {}

        for glob, queue in routes.items():
            if '*' in glob:
                self.glob_routes[glob] = queue
            else:
                self.routes[glob] = queue

    def route_for_task(self, task, args=None, kwargs=None):
        if task in self.routes:
            return self.routes[task]

        for route in self.glob_routes:
            prefix = route.split('*')[0]
            if task.startswith(prefix):
                return self.glob_routes[route]

        return None # for unknown tasks will be used default queue

用法：

# celery.py
CELERY_ROUTES = {
    'web.*':            'web',
    'web.slow_task':    'slow',
    'worker.*':         'worker',
}

app = Celery('config')
app.config_from_object('django.conf:settings') # Django or your app config

app.conf.update(
    CELERY_ROUTES=(TaskRouter(CELERY_ROUTES),),
)

TaskRouter 的工作原理：

In [2]: CELERY_ROUTES = { 
   ...:     'web.*':            'web', 
   ...:     'web.slow_task':    'slow', 
   ...:     'worker.*':         'worker', 
   ...: }                                                                       

In [3]: router = TaskRouter(CELERY_ROUTES)                                      

In [4]: router.route_for_task('web.blabla')                                     
Out[4]: 'web'

In [5]: router.route_for_task('web.slow_task')                                  
Out[5]: 'slow'

In [6]: router.route_for_task('unknown_task')  # None = default queue                                 

In [7]: router.route_for_task('worker.foo')                                     
Out[7]: 'worker'

In [8]: router.route_for_task('worker.bar')                                     
Out[8]: 'worker'