Celery-UserGuide-Tasks

任务是celery的组成要素。
任务是由可调用对象创建的类。任务包含双重角色，因为它既定义了调用任务（发送消息）时发生的事情，也定义了工人/worker接收到该消息时发生的情况。
每个任务类都含有唯一的名称，这个名称会在人物信息中引用，worker通过这个名称执行正确的函数。
任务信息会被保留在队列中，直到worker确认任务信息。一个工人可以预先保留许多信息，即使工人因停电或其他原因死亡，消息也会重新传递给另一个工人。
理想情况下，任务函数应该是幂等的：这意味着即使使用相同的参数多次调用该函数也不会产生意外的效果。由于worker无法检测到任务是否幂等，默认行为是在消息执行之前提前确认消息，这样就不会再执行已经启动的任务调用。
如果用户定义的任务是幂等的，那么可以通过设置acks_late选项使工人在任务返回时再确认任务信息。
即是设置了acks_late，如果执行任务的子进程结束(无论是通过退出信号还是调用sys.exit())，工人就会确认任务信息。

本章介绍如何定义任务

Basic

用户可以通过task()装饰可调用对象创建任务，例：

from .models import User

@app.task
def create_user(username, password):
    User.objects.create(username=username, password=password)

可以通过设置参数定义任务的属性

@app.task(serializer='json')
def create_user(username, password):
    User.objects.create(username=username, password=password)

task装饰器可以通过Celery application实例调用。如果用户使用Django或再定义一套库，那么可以使用shared_task()装饰器：

from celery import shared_task

@shared_task
def add(x, y):
    return x + y

当多个装饰器一起使用时，要确保task装饰器最后生效，即在最上层：

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

任务绑定意味着任务的第一个参数是任务实例本身：

logger = get_task_logger(__name__)

@task(bind=True)
def add(self, x, y):
    logger.info(self.request.id)

任务继承：在task装饰器中使用base参数指定任务的基类

import celery

class MyTask(celery.Task):

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

@task(base=MyTask)
def add(x, y):
    raise KeyError()

Name

每个任务都必须有一个唯一的名字。
如果没有指定名称，那么装饰器会自动生成一个，这个名称通过1任务所在的模块、2任务函数的名称生成。
指定名称:

>>> @app.task(name='sum-of-two-numbers')
>>> def add(x, y):
...     return x + y

>>> add.name
'sum-of-two-numbers'

最好使用模块名作为命名空间以避免和其他模块下的任务冲突：

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

用户可以通过任务名称时行分辨不同任务:

>>> add.name
'tasks.add'

下例中任务定义在名为tasks.py的模块下，会自动生成tasks.add作为任务名：
tasks.py

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

>>> from tasks import add
>>> add.name
'tasks.add'

自动命名和相对路径引用不能很好的兼容，如果要使用相对引用，需要指定，明确的任务名称。比如客户端用’.tasks’引用’myapps.tasks’模块，而工人使用myapp.tasks引入模块，则自动生成的名称不能正确匹配任务，会抛出NotRegistered错误。
在Django中使用project.myapp这种格式的引入也会导致同样的问题：

INSTALLED_APPS = ['project.myapp']

如果用户在project.myapp中定义应用，那任务模块必须以project.myapp.tasks的形式引入。

>>> from project.myapp.tasks import mytask   # << GOOD

>>> from myapp.tasks import mytask    # << BAD!!!

第二种情况会因为客户端和worker引入的模块名不同

>>> from project.myapp.tasks import mytask
>>> mytask.name
'project.myapp.tasks.mytask'

>>> from myapp.tasks import mytask
>>> mytask.name
'myapp.tasks.mytask'

任务请求

app.Task.request包含了当前执行任务的信息和状态。
request定义了如下属性:

id	当前执行任务的唯一id
group	任务分组的唯一id
chord
correlation_id	自定义的id，用于重复数据消除之类
args	位置参数
kwargs	关键字参数
origin	发送任务的主机
retries	当前任务重试的次数，从0开始
is_eager	如果任务在客户端本地执行而不是由worker执行，则设置为True
eta	UTC格式的任务预测时间
expires	任务的过期时间
hostname	执行任务的worker所在的主机
delivery_info	额外的任务信息的传递信息。存储了用于传递此任务的交换密钥和路由密钥的映射。比如app.Task.retry()使用这个参数把任务重新发送相同的队列。可用的键取决于使用的消息中间件。
reply_to	用于返回响应的队列名称。
called_directly	标记任务是否不是被worker执行
timelimit	任务的时间限制元组
callback	任务运行成功后需要调用的信号函数列表
errbacks	任务失败后需要调用的信号函数列表
utc	标记任务使用使用UTC

例：

@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))

bind设置为True意味着函数会变成绑定方法，这样用户就可以访问任务类型实例的属性和方法。

日志

worker会自动设置日志，用户也可以自己配置日志。
celery内置的日志器名为celery.task，继承这个日志器可以自动获取任务名称和唯一id。
最好在模块中先定义任务通用的日志器

from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@app.task
def add(x, y):
    logger.info('Adding {0} + {1}'.format(x, y))
    return x + y

celery使用python标准库中的logger模块。
用户也可以使用print或其他输出到标注输出或标注错误的方法，这些都会被重定向到日志系统。

参数检查

celery会在用户调用函数时检查参数：

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

# Calling the task with two arguments works:
>>> add.delay(8, 8)
<AsyncResult: f59d71ca-1549-43e0-be41-4e8821a83c0c>

# Calling the task with only one argument fails:
>>> add.delay(8)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "celery/app/task.py", line 376, in delay
    return self.apply_async(args, kwargs)
  File "celery/app/task.py", line 485, in apply_async
    check_arguments(*(args or ()), **(kwargs or {}))
TypeError: add() takes exactly 2 arguments (1 given)

设置typing参数为False可以禁用参数检查

>>> @app.task(typing=False)
... def add(x, y):
...     return x + y

# Works locally, but the worker receiving the task will raise an error.
>>> add.delay(8)
<AsyncResult: f59d71ca-1549-43e0-be41-4e8821a83c0c>

使用task_protocol1或更高的设置，用户可以通过argsrepe和kwargsrepr指定位置参数和关键字参数如何在日志中和监控事件中展示:

>>> add.apply_async((2, 3), argsrepr='(<secret-x>, <secret-y>)')

>>> charge.s(account, card='1234 5678 1234 5678').set(
...     kwargsrepr=repr({'card': '**** **** **** 5678'})
... ).delay()

重试

使用app.Task.retyr()可以重新执行任务。
调用retry方法时，该方法会给像是任务所在的队列发送新的信息和相同的任务id。
当任务重试时，会设置任务状态为重试。
例：

@app.task(bind=True)
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)

bind参数将允许访问任务实例。
exc参数用于传递日志中使用的异常信息，以及保存任务结果的时间。如果设置了result abckend，那么任务的异常和回溯都可以被访问。
如果任务具有max_retries值，且已超过最大重试次数，则将重新引发当前异常。除非：

• 没有设置exc参数，这时会抛出MaxRetriesExceededError
• 当前没有异常。如果没有原始的异常，则会抛出exc参数指定的异常，如self.retry(exc=Twitter.LoginError())。

使用自定义的延迟重试

任务重试前可以等待一段时间，默认时间通过default_retry_delay参数指定。默认是3min。这个参数的单位是秒。
使用countdown参数可以覆盖重试时间值

@app.task(bind=True, default_retry_delay=30 * 60)  # retry in 30 minutes.
def add(self, x, y):
    try:
        something_raising()
    except Exception as exc:
        # overrides the default delay to retry after 1 minute
        raise self.retry(exc=exc, countdown=60)

Task装饰器的参数

给task装饰器设置的参数最终会被设置为task类的属性。

通用参数

Task.name: 注册的任务名称。用户可以手动设置任务，也可以自动生成。
Task.request: 如果任务被执行，request会保存本次执行请求的任务信息。使用线程本地存储。Task.max_retries: 当任务调用self.retry方法或装饰器使用了autoretry_for参数时会使用此参数。在任务失败前充实的最大次数。超出后会抛出MaxRetriesExceededError异常。
retry()方法必须手动调用。
Task.throws: 异常类元组，发生元组中的异常时不会被当作真的异常，会被报告为任务失败，但不会在日志中作为error记录，也不会保留回溯。
例：

@task(throws=(KeyError, HttpNotFound)):
def get_foo():
    something()

异常类型:
预测到的异常——包含在Task.throws中：
使用INFO级别的日志，不保留回溯
未预测到的异常：
使用ERROR级别的日志，保留回溯。
Task.default_retry_delay: 任务重试前等待的秒数。默认3分钟。
Task.rate_limit: 设置此任务类型的速率限制（限制在给定时间范围内可以运行的任务数）。当速率限制生效时，任务仍将完成，但可能需要一段时间才能开始。
设为None时，不做限制。设为整数或浮点数时，视为每秒执行的任务数。
添加“/s”, “/m” or “/h” 可以设置每秒、每分钟或每小时这时的任务数。
比如 “100/m”会强制每两个任务之间最少间隔600ms。
默认值是task_defauilt_rate_limit设置。
这设置的是每个worker实例的速率而不是全局速率。
Task.time_limit: 当前任务的硬性时间限制，以秒为单位。
Task.soft_time_limit: 任务的软性时间限制。
Task.ignore_result: 不保留任务状态。如果设置为True就不能通过AsyncResult检查任务是否就绪或获取任务结果。
Task.store_errors_even_if_ignored: 设为True时，即是设置了ignore_result为True，也依然会保留异常。
Task.serializer: 定义默认的序列化方法的字符串。默认值是task_serializer设置。可选值为pickle/json/yaml或其他自定义的可被kombu.serialization.registry注册的序列化方法。
Task.compression: 字符创，定义默认的压缩框架。默认是task_compression设置。可以是gzip/bzip2或其他在kombu.compression.regitry中注册的压缩框架。
Task.backend: 结果存储后台。celery.backends中任一实例。默认是app.backend，由reault_backend参数设置。
Task.acks_late: 设为True时，任务信息会在任务被执行后确认。默认是在执行前确认。这意味着如果工作线程在执行过程中崩溃，任务可能会执行多次。
Task.track_started: 设为True时，任务被worker执行时会把状态设置为started。默认值为False，因为正常行为是不报告该粒度级别。任务只有pending\finished或waiting to be retried三种状态。执行任务的worker所在的主机名称和进程id可以通过state meta-data访问。

State

celery可以追溯任务的当前状态，也包含成功任务的结果，失败任务的异常和回溯。
在任务的生命周期中，任务可以转换多种状态，每种状态可能有不同的meta-data。当任务进入新的状态，原状态会被舍弃。
还有一些状态集，比如故障状态集和就绪状态集。
客户端使用状态即中的成员判断是否需要再次抛出异常或缓存结果。

内置状态

PENDING

任务等待被执行或未知任务。

STARTED

任务开始。需要设置app.Task.track_started。
meta-data: 执行任务的worker所在主机和进程。

SUCCESS

任务执行成功。
meta-data: 任务返回的结果

FAILURE

任务执行失败
meta-data: 抛出的异常和回溯。

RETRY

任务在重试。
meta-data: 导致重试的异常和异常回溯。

REVOKED

任务被驱逐。

自定义状态

设置一个唯一的名称即可定义状态。状态通常有大写字母组成。可以参考abortable tasks种定义的ABORTED状态。
使用updated_state()更新任务状态：

@app.task(bind=True)
def upload_files(self, filenames):
    for i, file in enumerate(filenames):
        if not self.request.called_directly:
            self.update_state(state='PROGRESS',
                meta={'current': i, 'total': len(filenames)})

这里定义了PROGRESS状态，通过将当前计数和总计数作为状态元数据的一部分，告诉任何知道此状态的应用程序任务当前正在进行中，以及任务在进程中的位置。可用于创建进度条。