前文列表

分布式任务队列 Celery
分布式任务队列 Celery —— 详解工作流
分布式任务队列 Celery —— 应用基础

前言

紧接前文，继续深入了解 Celery Tasks。示例代码依旧在前文的基础上进行修改。

​Tasks 是 Celery 的基石，原型类为 celery.app.task:Task，它提供了两个核心功能：

• 将任务消息发送到队列
• 声明 Worker 接收到消息后需要执行的具体函数

Task 的实例化

使用装饰器 app.task 来装饰一个普通函数，就可以轻松创建出一个任务函数。

from proj.celery import app

@app.task
def add(x, y):
    return x + y

值得注意的是，任务函数本质上已经不再是一个普通函数，而是一个 celery.app.task:Task 实例对象。

>>> from proj.task.tasks import add
>>> dir(add)
['AsyncResult', 'MaxRetriesExceededError', ..., 'apply_async', 'delay', u'name', 'on_bound', 'on_failure', 'on_retry', 'on_success', 'request', 'retry', 'subtask', ...]

所以任务函数才可以调用 delay/apply_async 等属于 Task 的实例属性和方法。

>>> add.apply_async
<bound method add.apply_async of <@task: proj.task.tasks.add of proj at 0x7fedb363a790>>
>>> add.delay
<bound method add.delay of <@task: proj.task.tasks.add of proj at 0x7fedb363a790>>

这是一个非常重要的认识「app.task 的 “装饰” 动作，其实是 Task 的实例化过程」，而装饰器的参数就是 Task 初始化的参数。
官方文档（http://docs.celeryproject.org/en/latest/userguide/tasks.html#list-of-options）提供了完整的 app.task 装饰器参数列表。

除此之外，还需要注意「多装饰器顺序」的坑，app.task 应该始终放到最后使用，才能保证其稳定有效。

app.task
@decorator2
@decorator1
def add(x, y):
    return x + y

任务的名字

每个任务函数都具有唯一的名字，这个名字被包含在任务消息中，Worker 通过该名字来找到具体执行的任务函数。默认的，Task 会启用自动命名，将函数的全路径名作为任务名。

>>> add.name
u'proj.task.tasks.add'

当然了，也可以通过指定装饰器参数 name 来指定任务名。

@app.task(name='new_name')
def add(x, y):
    return x + y

>>> from proj.task.tasks import add
>>> add.name
'new_name'

但为了避免命名冲突的问题，一般不建议这么做，除非你很清楚自己的做什么。

任务的绑定

既然任务函数本质是一个 Task 实例对象，那么当然也可以应用 self 绑定特性。

# 启用绑定：
@app.task(bind=True)
def add(self, x, y):
    print("self: ", self)
    return x + y

>>> add.delay(1, 2)
<AsyncResult: 1982dc85-694b-4ceb-849b-5f69e40b4fe9>

绑定对象 self 十分重要，Task 的很多高级功能都是依靠它作为载体来调用的。例如：任务重试功能，请求上下文功能。

任务的重试

任务重试功能的实现为 Task.retry，它将任务消息重新发送到同一个的队列中，以此来重启任务。

@app.task(bind=True, max_retries=3)
def send_twitter_status(self, oauth, tweet):
    try:
        twitter = Twitter(oauth)
        twitter.update_status(tweet)
    except (Twitter.FailWhaleError, Twitter.LoginError) as exc:
        raise self.retry(exc=exc)

• max_retries 指定了最大的重试次数
• exc 指定将异常信息输出到日志，需要开启 result backend。

如果你仅希望触发特定异常时才进行重试，可以应用 Task 的「Automatic retry for known exceptions」特性。

# 只有在触发 FailWhaleError 异常时，才会重试任务，且最多重试 5 次。
@app.task(autoretry_for=(FailWhaleError,), retry_kwargs={'max_retries': 5})
def refresh_timeline(user):
    return twitter.refresh_timeline(user)

任务的请求上下文

在 Celery 请求 Worker 执行任务函数时，提供了请求的上下文，这是为了让任务函数在执行过程中能够访问上下文所包含的任务状态和信息。

@app.task(bind=True)
def dump_context(self, x, y):
    print('Executing task id {0.id}, args: {0.args!r} kwargs: {0.kwargs!r}'.format(
            self.request))

>>> from proj.task.tasks import dump_context
>>> dump_context.delay(1, 2)
<AsyncResult: 00bc9f96-98df-4bca-a4a3-4774c535a44c>

捕获请求上下文中的有用信息，有利于我们去分辨和调查一个任务的执行状况。
完整的上下文属性列表，可以查阅官方文档（http://docs.celeryproject.org/en/latest/userguide/tasks.html#task-request）。

任务的继承

使用 app.task 装饰器默认会实例化原生的 Task 类，这虽然能满足大部分的应用场景需求，但并非全部。所以 Celery 允许我们通过继承 Task 类，来衍生出特异化的基类。这一特性在复杂的应用场景中将会十分有效。

• 重新指定适用于的所有任务的默认基类

def make_app(context):
    app = Celery('proj')
    app.config_from_object('proj.celeryconfig')

    default_exchange = Exchange('default', type='direct')
    web_exchange = Exchange('task', type='direct')
    app.conf.task_default_queue = 'default'
    app.conf.task_default_exchange = 'default'
    app.conf.task_default_routing_key = 'default'

    app.conf.task_queues = (
        Queue('default', default_exchange, routing_key='default'),
        Queue('high_queue', web_exchange, routing_key='hign_task'),
        Queue('low_queue', web_exchange, routing_key='low_task'),
    )

    app.conf.timezone = 'Asia/Shanghai'
    app.conf.beat_schedule = {
        'periodic_task_add': {
            'task': 'proj.task.tasks.add',
            'schedule': crontab(minute='*/1'),
            'args': (2, 2)
        },
    }

    TaskBase = app.Task

    class ContextTask(TaskBase):

        abstract = True
        context = ctx

        def __call__(self, *args, **kwargs):
            """Will be execute when create the instance object of ContextTesk.
            """
            LOG.info(_LI("Invoked celery task starting: %(name)s[%(id)s]"),
                     {'name': self.name, 'id': self.request.id})
            return super(ContextTask, self).__call__(*args, **kwargs)

        # 任务执行成功前做什么
        def on_success(self, retval, task_id, args, kwargs):
            """Invoked after the task is successfully execute.
            """
            LOG.info(_LI("Task %(id)s success: [%(ret)s]."),
                     {'id': task_id, 'ret': retval})
            return super(ContextTask, self).on_success(retval, task_id,
                                                       args, kwargs)

        # 任务执行失败后做什么
        def on_failure(self, exc, task_id, args, kwargs, einfo):
            """Invoked after the task failed to execute.
            """
            msg = _LE("Task [%(id)s] failed:
"
                      "args : %(args)s
"
                      "kwargs : %(kw)s
"
                      "detail :%(err)s") % {
                          'id': task_id, 'args': args,
                          'kw': kwargs, 'err': six.text_type(exc)}
            LOG.exception(msg)
            return super(ContextTask, self).on_failure(exc, task_id, args,
                                                       kwargs, einfo)

    # 重新赋予默认基类
    app.Task = ContextTask
    return app

• 为不同类型的任务继承出具有偏向属性的基类

Import celery


class JSONTask(celery.Task):

    serializer = 'json'

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        print('{0!r} failed: {1!r}'.format(task_id, exc))

class XMLTask(celery.Task):

    serializer = 'xml'

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        print('{0!r} failed: {1!r}'.format(task_id, exc))


# 指定不同的基类
@task(base=JSONTask)
def add_json(x, y):
    raise KeyError()

# 指定不同的基类
@task(base=XMLTask)
def add_xml(x, y):
    raise KeyError()