路由任务/Tasks¶

Routing Tasks

中文

备注

类似“主题（topic）”与“广播（fanout）”的路由概念并非在所有传输方案中都可用，请参阅 transport comparison table 获取详细信息。

英文

备注

Alternate routing concepts like topic and fanout is not available for all transports, please consult the transport comparison table.

基础知识¶

Basics

自动路由¶

Automatic routing

中文

最简单的路由方式是启用 task_create_missing_queues 配置（默认已开启）。

启用该设置后，在 task_queues 中未显式定义的命名队列会被自动创建。这使得执行简单的路由任务变得非常容易。

假设你有两台处理常规任务的服务器 x 与 y，以及一台专门处理 feed 相关任务的服务器 z，你可以使用如下配置:

task_routes = {'feed.tasks.import_feed': {'queue': 'feeds'}}

启用上述路由后，feed 导入任务将会被路由到 "feeds" 队列，其它任务将会被路由到默认队列（历史原因下默认命名为 "celery"）。

你也可以使用通配符或正则表达式来匹配 feed.tasks 命名空间中的所有任务：

app.conf.task_routes = {'feed.tasks.*': {'queue': 'feeds'}}

如果匹配顺序很重要，你应使用 items 格式来指定路由器：

task_routes = ([
    ('feed.tasks.*', {'queue': 'feeds'}),
    ('web.tasks.*', {'queue': 'web'}),
    (re.compile(r'(video|image)\.tasks\..*'), {'queue': 'media'}),
],)

备注

配置项 task_routes 可以是一个字典，也可以是一个包含路由器对象的列表，所以在本例中我们使用元组包裹该列表以进行指定。

在安装好路由器之后，你可以通过如下命令启动服务器 z，使其只处理 feeds 队列：

user@z:/$ celery -A proj worker -Q feeds

你可以指定任意数量的队列，因此你也可以让该服务器同时处理默认队列：

user@z:/$ celery -A proj worker -Q feeds,celery

英文

The simplest way to do routing is to use the task_create_missing_queues setting (on by default).

With this setting on, a named queue that's not already defined in task_queues will be created automatically. This makes it easy to perform simple routing tasks.

Say you have two servers, x, and y that handle regular tasks, and one server z, that only handles feed related tasks. You can use this configuration:

task_routes = {'feed.tasks.import_feed': {'queue': 'feeds'}}

With this route enabled import feed tasks will be routed to the "feeds" queue, while all other tasks will be routed to the default queue (named "celery" for historical reasons).

Alternatively, you can use glob pattern matching, or even regular expressions, to match all tasks in the feed.tasks name-space:

app.conf.task_routes = {'feed.tasks.*': {'queue': 'feeds'}}

If the order of matching patterns is important you should specify the router in items format instead:

task_routes = ([
    ('feed.tasks.*', {'queue': 'feeds'}),
    ('web.tasks.*', {'queue': 'web'}),
    (re.compile(r'(video|image)\.tasks\..*'), {'queue': 'media'}),
],)

备注

The task_routes setting can either be a dictionary, or a list of router objects, so in this case we need to specify the setting as a tuple containing a list.

After installing the router, you can start server z to only process the feeds queue like this:

user@z:/$ celery -A proj worker -Q feeds

You can specify as many queues as you want, so you can make this server process the default queue as well:

user@z:/$ celery -A proj worker -Q feeds,celery

更改默认队列的名称¶

Changing the name of the default queue

中文

你可以通过以下配置修改默认队列的名称：

app.conf.task_default_queue = 'default'

英文

You can change the name of the default queue by using the following configuration:

app.conf.task_default_queue = 'default'

队列的定义方式¶

How the queues are defined

中文

该特性旨在为用户隐藏复杂的 AMQP 协议，实现基本需求即可使用。但你可能仍然对这些队列是如何声明的感兴趣。

一个名为 "video" 的队列将会使用以下配置被创建：

{'exchange': 'video',
'exchange_type': 'direct',
'routing_key': 'video'}

对于非 AMQP 后端（如 Redis 或 SQS），它们不支持 exchange 的概念，因此要求 exchange 的名称与队列一致。该设计可确保其兼容。

英文

The point with this feature is to hide the complex AMQP protocol for users with only basic needs. However -- you may still be interested in how these queues are declared.

A queue named "video" will be created with the following settings:

{'exchange': 'video',
'exchange_type': 'direct',
'routing_key': 'video'}

The non-AMQP backends like Redis or SQS don't support exchanges, so they require the exchange to have the same name as the queue. Using this design ensures it will work for them as well.

手动路由¶

Manual routing

中文

再次假设你有两台处理常规任务的服务器 x 与 y，以及一台专门处理 feed 任务的服务器 z，你可以使用如下配置：

from kombu import Queue

app.conf.task_default_queue = 'default'
app.conf.task_queues = (
    Queue('default',    routing_key='task.#'),
    Queue('feed_tasks', routing_key='feed.#'),
)
app.conf.task_default_exchange = 'tasks'
app.conf.task_default_exchange_type = 'topic'
app.conf.task_default_routing_key = 'task.default'

task_queues 是由 Queue 实例组成的列表。如果你未为某项设置 exchange 或 exchange_type，这些值将从配置项 task_default_exchange 与 task_default_exchange_type 中继承。

要将任务路由至 feed_tasks 队列，可以在 task_routes 配置中添加如下条目：

task_routes = {
        'feeds.tasks.import_feed': {
            'queue': 'feed_tasks',
            'routing_key': 'feed.import',
        },
}

你也可以通过 Task.apply_async() 或 send_task() 的 routing_key 参数进行覆盖：

>>> from feeds.tasks import import_feed
>>> import_feed.apply_async(args=['http://cnn.com/rss'],
...                         queue='feed_tasks',
...                         routing_key='feed.import')

要让服务器 z 专门消费 feed 队列，可以使用 celery worker -Q 启动：

user@z:/$ celery -A proj worker -Q feed_tasks --hostname=z@%h

服务器 x 与 y 应配置为消费默认队列：

user@x:/$ celery -A proj worker -Q default --hostname=x@%h
user@y:/$ celery -A proj worker -Q default --hostname=y@%h

如果你希望 feed 处理节点在高负载时也能处理常规任务，可以这样启动：

user@z:/$ celery -A proj worker -Q feed_tasks,default --hostname=z@%h

如果你想添加一个使用不同 exchange 的队列，只需显式指定 exchange 与类型：

from kombu import Exchange, Queue

app.conf.task_queues = (
    Queue('feed_tasks',    routing_key='feed.#'),
    Queue('regular_tasks', routing_key='task.#'),
    Queue('image_tasks',   exchange=Exchange('mediatasks', type='direct'),
                        routing_key='image.compress'),
)

如果你对上述术语感到困惑，建议你阅读 AMQP 相关文档。

参见

除了下面的 Redis 消息优先级外，还有一篇优秀的博文 Rabbits and Warrens 介绍了队列与交换机的概念；同时还有 CloudAMQP 教程，以及面向 RabbitMQ 用户的 RabbitMQ FAQ，都可以作为信息来源。

英文

Say you have two servers, x, and y that handle regular tasks, and one server z, that only handles feed related tasks, you can use this configuration:

from kombu import Queue

app.conf.task_default_queue = 'default'
app.conf.task_queues = (
    Queue('default',    routing_key='task.#'),
    Queue('feed_tasks', routing_key='feed.#'),
)
app.conf.task_default_exchange = 'tasks'
app.conf.task_default_exchange_type = 'topic'
app.conf.task_default_routing_key = 'task.default'

task_queues is a list of Queue instances. If you don't set the exchange or exchange type values for a key, these will be taken from the task_default_exchange and task_default_exchange_type settings.

To route a task to the feed_tasks queue, you can add an entry in the task_routes setting:

task_routes = {
        'feeds.tasks.import_feed': {
            'queue': 'feed_tasks',
            'routing_key': 'feed.import',
        },
}

You can also override this using the routing_key argument to Task.apply_async(), or send_task():

>>> from feeds.tasks import import_feed
>>> import_feed.apply_async(args=['http://cnn.com/rss'],
...                         queue='feed_tasks',
...                         routing_key='feed.import')

To make server z consume from the feed queue exclusively you can start it with the celery worker -Q option:

user@z:/$ celery -A proj worker -Q feed_tasks --hostname=z@%h

Servers x and y must be configured to consume from the default queue:

user@x:/$ celery -A proj worker -Q default --hostname=x@%h
user@y:/$ celery -A proj worker -Q default --hostname=y@%h

If you want, you can even have your feed processing worker handle regular tasks as well, maybe in times when there's a lot of work to do:

user@z:/$ celery -A proj worker -Q feed_tasks,default --hostname=z@%h

If you have another queue but on another exchange you want to add, just specify a custom exchange and exchange type:

from kombu import Exchange, Queue

app.conf.task_queues = (
    Queue('feed_tasks',    routing_key='feed.#'),
    Queue('regular_tasks', routing_key='task.#'),
    Queue('image_tasks',   exchange=Exchange('mediatasks', type='direct'),
                        routing_key='image.compress'),
)

If you're confused about these terms, you should read up on AMQP.

参见

In addition to the Redis 消息优先级 below, there's Rabbits and Warrens, an excellent blog post describing queues and exchanges. There's also The CloudAMQP tutorial, For users of RabbitMQ the RabbitMQ FAQ could be useful as a source of information.

特殊路由选项¶

Special Routing Options

RabbitMQ 消息优先级¶

RabbitMQ Message Priorities

中文

supported transports:: RabbitMQ

Added in version 4.0.

可以通过设置 x-max-priority 参数来配置队列以支持优先级：

from kombu import Exchange, Queue

app.conf.task_queues = [
    Queue('tasks', Exchange('tasks'), routing_key='tasks',
        queue_arguments={'x-max-priority': 10}),
]

可以使用 task_queue_max_priority 设置为所有队列配置一个默认优先级上限：

app.conf.task_queue_max_priority = 10

也可以使用 task_default_priority 为所有任务设置默认优先级：

app.conf.task_default_priority = 5

英文

supported transports:: RabbitMQ

Added in version 4.0.

Queues can be configured to support priorities by setting the x-max-priority argument:

from kombu import Exchange, Queue

app.conf.task_queues = [
    Queue('tasks', Exchange('tasks'), routing_key='tasks',
        queue_arguments={'x-max-priority': 10}),
]

A default value for all queues can be set using the task_queue_max_priority setting:

app.conf.task_queue_max_priority = 10

A default priority for all tasks can also be specified using the task_default_priority setting:

app.conf.task_default_priority = 5

Redis 消息优先级¶

Redis Message Priorities

中文

supported transports:: Redis

虽然 Celery 的 Redis 传输支持读取任务的优先级字段，但 Redis 本身并不具备原生的优先级概念。在尝试使用 Redis 实现优先级功能前，请务必阅读以下说明，因为这可能会导致某些意料之外的行为。

要启用基于优先级的任务调度，需配置传输选项中的 queue_order_strategy：

app.conf.broker_transport_options = {
    'queue_order_strategy': 'priority',
}

该优先级支持机制是通过为每个队列创建多个列表（list）实现的。尽管理论上支持 10 个（0 到 9）优先级等级，但为了节省资源，默认会将其压缩为 4 个等级。也就是说，一个名为 celery 的队列实际上会被拆分为 4 个内部队列。

优先级最高的队列仍然命名为 celery，其他的队列则会使用一个分隔符（默认是 x06x16）加上优先级数字附加在原始队列名后构成：

['celery', 'celery\x06\x163', 'celery\x06\x166', 'celery\x06\x169']

如果你希望使用更多的优先级等级，或想更改默认分隔符，可以通过配置 priority_steps 与 sep 参数实现：

app.conf.broker_transport_options = {
    'priority_steps': list(range(10)),
    'sep': ':',
    'queue_order_strategy': 'priority',
}

上述配置将生成如下的队列名称：

['celery', 'celery:1', 'celery:2', 'celery:3', 'celery:4', 'celery:5', 'celery:6', 'celery:7', 'celery:8', 'celery:9']

需要注意的是，这种机制永远无法达到由消息代理服务器层实现的优先级那样的精确性，充其量是一种近似的实现。但对于大多数应用场景而言，这种机制可能已经足够用。

英文

supported transports:: Redis

While the Celery Redis transport does honor the priority field, Redis itself has no notion of priorities. Please read this note before attempting to implement priorities with Redis as you may experience some unexpected behavior.

To start scheduling tasks based on priorities you need to configure queue_order_strategy transport option.

app.conf.broker_transport_options = {
    'queue_order_strategy': 'priority',
}

The priority support is implemented by creating n lists for each queue. This means that even though there are 10 (0-9) priority levels, these are consolidated into 4 levels by default to save resources. This means that a queue named celery will really be split into 4 queues.

The highest priority queue will be named celery, and the the other queues will have a separator (by default x06x16) and their priority number appended to the queue name.

['celery', 'celery\x06\x163', 'celery\x06\x166', 'celery\x06\x169']

If you want more priority levels or a different separator you can set the priority_steps and sep transport options:

app.conf.broker_transport_options = {
    'priority_steps': list(range(10)),
    'sep': ':',
    'queue_order_strategy': 'priority',
}

The config above will give you these queue names:

['celery', 'celery:1', 'celery:2', 'celery:3', 'celery:4', 'celery:5', 'celery:6', 'celery:7', 'celery:8', 'celery:9']

That said, note that this will never be as good as priorities implemented at the broker server level, and may be approximate at best. But it may still be good enough for your application.

路由任务¶

Routing Tasks

定义队列¶

Defining queues

中文

在 Celery 中，可用的队列由 task_queues 设置定义。

以下是一个包含三个队列的示例队列配置；一个用于视频，一个用于图像，还有一个默认队列，用于处理其他所有任务：

default_exchange = Exchange('default', type='direct')
media_exchange = Exchange('media', type='direct')

app.conf.task_queues = (
    Queue('default', default_exchange, routing_key='default'),
    Queue('videos', media_exchange, routing_key='media.video'),
    Queue('images', media_exchange, routing_key='media.image')
)
app.conf.task_default_queue = 'default'
app.conf.task_default_exchange = 'default'
app.conf.task_default_routing_key = 'default'

在上述配置中，task_default_queue 将用于路由那些未显式指定路由的任务。

默认交换器、交换器类型和路由键将作为任务的默认路由参数，同时也作为 task_queues 项中条目的默认值。

还支持将多个绑定连接到同一个队列。以下是一个将两个路由键绑定到同一个队列的示例：

from kombu import Exchange, Queue, binding

media_exchange = Exchange('media', type='direct')

CELERY_QUEUES = (
    Queue('media', [
        binding(media_exchange, routing_key='media.video'),
        binding(media_exchange, routing_key='media.image'),
    ]),
)

英文

In Celery available queues are defined by the task_queues setting.

Here's an example queue configuration with three queues; One for video, one for images, and one default queue for everything else:

default_exchange = Exchange('default', type='direct')
media_exchange = Exchange('media', type='direct')

app.conf.task_queues = (
    Queue('default', default_exchange, routing_key='default'),
    Queue('videos', media_exchange, routing_key='media.video'),
    Queue('images', media_exchange, routing_key='media.image')
)
app.conf.task_default_queue = 'default'
app.conf.task_default_exchange = 'default'
app.conf.task_default_routing_key = 'default'

Here, the task_default_queue will be used to route tasks that doesn't have an explicit route.

The default exchange, exchange type, and routing key will be used as the default routing values for tasks, and as the default values for entries in task_queues.

Multiple bindings to a single queue are also supported. Here's an example of two routing keys that are both bound to the same queue:

from kombu import Exchange, Queue, binding

media_exchange = Exchange('media', type='direct')

CELERY_QUEUES = (
    Queue('media', [
        binding(media_exchange, routing_key='media.video'),
        binding(media_exchange, routing_key='media.image'),
    ]),
)

指定任务目标¶

Specifying task destination

中文

任务的投递目标由以下内容决定（按顺序）：

调用 Task.apply_async() 时传入的路由参数。
在 Task 上定义的与路由相关的属性。
在 task_routes 中定义的路由器。

最佳实践是不硬编码这些设置，而是通过使用路由器留作配置选项；这是一种最灵活的方式，但仍可以通过任务属性设置合理的默认值。

英文

The destination for a task is decided by the following (in order):

The routing arguments to Task.apply_async().
Routing related attributes defined on the Task itself.
The 路由器 defined in task_routes.

It's considered best practice to not hard-code these settings, but rather leave that as configuration options by using 路由器; This is the most flexible approach, but sensible defaults can still be set as task attributes.

路由器¶

Routers

中文

路由器是决定任务路由选项的函数。

定义一个新的路由器函数，只需要定义一个具有如下签名的函数： (name, args, kwargs, options, task=None, **kw)：

def route_task(name, args, kwargs, options, task=None, **kw):
        if name == 'myapp.tasks.compress_video':
            return {'exchange': 'video',
                    'exchange_type': 'topic',
                    'routing_key': 'video.compress'}

如果你返回了 queue 键，它将会展开成该队列在 task_queues 中定义的设置：

{'queue': 'video', 'routing_key': 'video.compress'}

展开为 -->

{'queue': 'video',
'exchange': 'video',
'exchange_type': 'topic',
'routing_key': 'video.compress'}

你可以通过将路由器类添加到 task_routes 设置中来安装它们：

task_routes = (route_task,)

路由器函数也可以通过名称添加：

task_routes = ('myapp.routers.route_task',)

对于简单的任务名 -> 路由映射，你可以直接将字典传递给 task_routes，以实现与上述示例相同的行为：

task_routes = {
    'myapp.tasks.compress_video': {
        'queue': 'video',
        'routing_key': 'video.compress',
    },
}

之后路由器将按顺序遍历，只要遇到一个返回了真值的路由器，就会停止，并使用该路由作为任务的最终路由。

你也可以将多个路由器按顺序定义在列表中：

task_routes = [
    route_task,
    {
        'myapp.tasks.compress_video': {
            'queue': 'video',
            'routing_key': 'video.compress',
    },
]

这些路由器将依次被调用，选择第一个返回值的路由器作为结果。

如果你使用的是 Redis 或 RabbitMQ，也可以在路由中指定队列的默认优先级：

task_routes = {
    'myapp.tasks.compress_video': {
        'queue': 'video',
        'routing_key': 'video.compress',
        'priority': 10,
    },
}

类似地，调用任务的 apply_async 方法也可以覆盖默认优先级：

task.apply_async(priority=0)

优先级顺序与集群响应性

需要注意的是，由于 worker 预取（prefetching）机制的存在，如果一批任务在同一时间提交，可能最初不会按优先级顺序执行。禁用 worker 预取可以避免该问题，但对于短小快速的任务，这可能会导致性能下降。在大多数情况下，仅将 worker_prefetch_multiplier 降为 1 就是更简单且更优雅的方式，它能提高系统响应性而不需完全禁用预取。

注意，在使用 Redis broker 时，优先级值是逆序排序的：0 表示最高优先级。

英文

A router is a function that decides the routing options for a task.

All you need to define a new router is to define a function with the signature (name, args, kwargs, options, task=None, **kw):

def route_task(name, args, kwargs, options, task=None, **kw):
        if name == 'myapp.tasks.compress_video':
            return {'exchange': 'video',
                    'exchange_type': 'topic',
                    'routing_key': 'video.compress'}

If you return the queue key, it'll expand with the defined settings of that queue in task_queues:

{'queue': 'video', 'routing_key': 'video.compress'}

becomes -->

{'queue': 'video',
'exchange': 'video',
'exchange_type': 'topic',
'routing_key': 'video.compress'}

You install router classes by adding them to the task_routes setting:

task_routes = (route_task,)

Router functions can also be added by name:

task_routes = ('myapp.routers.route_task',)

For simple task name -> route mappings like the router example above, you can simply drop a dict into task_routes to get the same behavior:

task_routes = {
    'myapp.tasks.compress_video': {
        'queue': 'video',
        'routing_key': 'video.compress',
    },
}

The routers will then be traversed in order, it will stop at the first router returning a true value, and use that as the final route for the task.

You can also have multiple routers defined in a sequence:

task_routes = [
    route_task,
    {
        'myapp.tasks.compress_video': {
            'queue': 'video',
            'routing_key': 'video.compress',
    },
]

The routers will then be visited in turn, and the first to return a value will be chosen.

If you're using Redis or RabbitMQ you can also specify the queue's default priority in the route.

task_routes = {
    'myapp.tasks.compress_video': {
        'queue': 'video',
        'routing_key': 'video.compress',
        'priority': 10,
    },
}

Similarly, calling apply_async on a task will override that default priority.

task.apply_async(priority=0)

Priority Order and Cluster Responsiveness

It is important to note that, due to worker prefetching, if a bunch of tasks submitted at the same time they may be out of priority order at first. Disabling worker prefetching will prevent this issue, but may cause less than ideal performance for small, fast tasks. In most cases, simply reducing worker_prefetch_multiplier to 1 is an easier and cleaner way to increase the responsiveness of your system without the costs of disabling prefetching entirely.

Note that priorities values are sorted in reverse when using the redis broker: 0 being highest priority.

广播¶

Broadcast

中文

Celery 也支持广播路由（broadcast routing）。下面是一个名为 broadcast_tasks 的交换器示例，它会将任务的副本发送给所有连接到它的 worker：

from kombu.common import Broadcast

app.conf.task_queues = (Broadcast('broadcast_tasks'),)
app.conf.task_routes = {
    'tasks.reload_cache': {
        'queue': 'broadcast_tasks',
        'exchange': 'broadcast_tasks'
    }
}

现在， tasks.reload_cache 任务将会被发送给所有从该队列消费的 worker。

以下是另一个使用广播路由的示例，这次是结合 celery beat 调度器：

from kombu.common import Broadcast
from celery.schedules import crontab

app.conf.task_queues = (Broadcast('broadcast_tasks'),)

app.conf.beat_schedule = {
    'test-task': {
        'task': 'tasks.reload_cache',
        'schedule': crontab(minute=0, hour='*/3'),
        'options': {'exchange': 'broadcast_tasks'}
    },
}

广播与结果存储

请注意，Celery 的任务结果机制并未定义当两个任务具有相同 task_id 时会发生什么。如果同一个任务被分发给多个 worker，那么任务状态的历史记录可能无法被保留。

在这种情况下，建议设置 task.ignore_result 属性以避免不一致。

英文

Celery can also support broadcast routing. Here is an example exchange broadcast_tasks that delivers copies of tasks to all workers connected to it:

from kombu.common import Broadcast

app.conf.task_queues = (Broadcast('broadcast_tasks'),)
app.conf.task_routes = {
    'tasks.reload_cache': {
        'queue': 'broadcast_tasks',
        'exchange': 'broadcast_tasks'
    }
}

Now the tasks.reload_cache task will be sent to every worker consuming from this queue.

Here is another example of broadcast routing, this time with a celery beat schedule:

from kombu.common import Broadcast
from celery.schedules import crontab

app.conf.task_queues = (Broadcast('broadcast_tasks'),)

app.conf.beat_schedule = {
    'test-task': {
        'task': 'tasks.reload_cache',
        'schedule': crontab(minute=0, hour='*/3'),
        'options': {'exchange': 'broadcast_tasks'}
    },
}

Broadcast & Results

Note that Celery result doesn't define what happens if two tasks have the same task_id. If the same task is distributed to more than one worker, then the state history may not be preserved.

It's a good idea to set the task.ignore_result attribute in this case.

路由任务/Tasks¶

基础知识¶

自动路由¶

更改默认队列的名称¶

队列的定义方式¶

手动路由¶

特殊路由选项¶

RabbitMQ 消息优先级¶

Redis 消息优先级¶

AMQP 入门¶

消息¶

生产者、消费者和代理¶

交换器、队列和路由键¶

交换器类型¶

直接交换器¶

主题交换器¶

相关 API 命令¶

API 实践¶

路由任务¶

定义队列¶

指定任务目标¶

路由器¶

广播¶

路由任务/Tasks¶

基础知识¶

自动路由¶

更改默认队列的名称¶

队列的定义方式¶

手动路由¶

特殊路由选项¶

RabbitMQ 消息优先级¶

Redis 消息优先级¶

AMQP 入门¶

消息​​¶

生产者、消费者和代理¶

交换器、队列和路由键¶

交换器类型¶

直接交换器¶

主题交换器¶

相关 API 命令¶

API 实践¶

路由任务¶

定义队列¶

指定任务目标¶

路由器¶

广播¶

消息¶