通过使用app.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)

绑定的任务
任务被绑定意味着任务的第一个参数将始终是任务实例（self），就像Python绑定方法一样：


多个装饰器
当将多个装饰器与任务装饰器结合使用时，必须确保任务装饰器是最后应用的（奇怪的是，在Python中，这意味着它必须在列表中的第一个）： 
@app.task
@decorator2
@decorator1
def add(x, y):
    return x + y

任务继承
任务装饰器的基参数指定任务的基类：
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))

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

姓名
每个任务都必须有一个唯一的名称。
如果没有提供明确的名称，任务装饰器将为您生成一个，并且这个名称将基于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


目录展示
project/
       /__init__.py
       /celery.py
       /moduleA/
               /__init__.py
               /tasks.py
       /moduleB/
               /__init__.py
               /tasks.py

日志记录
工作人员将自动为您设置日志记录，或者您可以手动配置日志记录。
有一个名为“celener.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

重试延迟
@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)

from django.db import transaction
from django.http import HttpResponseRedirect

@transaction.atomic
def create_article(request):
    article = Article.objects.create()
    expand_abbreviations.delay(article.pk)
    return HttpResponseRedirect('/articles/')

这是一个Django视图，在数据库中创建一个文章对象，然后将主键传递给任务。它使用transaction.atomic装饰器，该装饰器将在视图返回时提交事务，或者在视图引发异常时回滚。
如果任务在事务提交之前就开始执行，则存在竞争条件；数据库对象还不存在！
解决方案是在成功提交所有事务后，使用on_commit回调来启动Celery任务。
from django.db.transaction import on_commit

def create_article(request):
    article = Article.objects.create()
    on_commit(lambda: expand_abbreviations.delay(article.pk))

T.delay(arg, kwarg=value)
.apply_async. (.delay(*args, **kwargs) calls .apply_async(args, kwargs)). 的快捷方式

T.apply_async(countdown=10)
 从现在起10秒内执行。 

T.apply_async(eta=now + timedelta(seconds=10))
 从现在起10秒内执行，使用eta指定 

T.apply_async(countdown=60, expires=120)
 从现在起一分钟内执行，但在2分钟后过期。 

T.apply_async(expires=now + timedelta(days=2))
2天后过期，使用日期时间设置。 

task.delay(arg1, arg2, kwarg1='x', kwarg2='y')

task.apply_async(args=[arg1, arg2], kwargs={'kwarg1': 'x', 'kwarg2': 'y'})


# 链式调用执行完异步后执行自己的函数
add.apply_async((2, 2), link=add.s(16))
这里，第一个任务（4）的结果将被发送到一个新任务，该新任务将前一个结果加16，形成表达式（2+2）+16=20
如果任务引发异常（errback），您也可以导致应用回调。worker实际上不会将errback作为任务调用，而是直接调用errback函数，以便将原始请求、异常和回溯对象传递给它。
这是一个错误回调示例：
@app.task
def error_handler(request, exc, traceback):
    print('Task {0} raised exception: {1!r}\n{2!r}'.format(
          request.id, exc, traceback))

 可以使用link_error执行选项将其添加到任务中： 
add.apply_async((2, 2), link_error=error_handler.s())

[签名官方文档](https://docs.celeryq.dev/en/stable/userguide/canvas.html#guide-canvas "签名官方文档")
您可以使用 添加任务的名称为其创建签名，如下所示：
>>> from celery import signature
>>> signature('tasks.add', args=(2, 2), countdown=10)
tasks.add(2, 2)

也可以使用任务的 签名 方法：
>>> add.signature((2, 2), countdown=10)
tasks.add(2, 2)

还有一个使用星星参数的快捷方式：
>>> add.s(2, 2)
tasks.add(2, 2)



此外， link 和 link_error 选项都可以表示为 作为列表：
add.apply_async((2, 2), link=[add.s(16), other_task.s()])

ETA（预计到达时间）允许您设置特定的日期和时间， 是执行任务的最早时间。倒计时为 一种快捷方式，可将ETA设置为未来的秒数。
>>> result = add.apply_async((2, 2), countdown=3)
>>> result.get()    # this takes at least 3 seconds to return
20

当 倒计时 为整数时， eta 必须为日期时间 对象，指定精确的日期和时间（包括毫秒精度， 和时区信息）：
>>> from datetime import datetime, timedelta

>>> tomorrow = datetime.utcnow() + timedelta(days=1)
>>> add.apply_async((2, 2), eta=tomorrow)

过期
>>> # Task expires after one minute from now.
>>> add.apply_async((10, 10), expires=60)

>>> # Also supports datetime
>>> from datetime import datetime, timedelta
>>> add.apply_async((10, 10), kwargs,
...                 expires=datetime.now() + timedelta(days=1)

指定调用任务时使用的压缩的示例：
results = []
with add.app.pool.acquire(block=True) as connection:
    with add.get_publisher(connection) as publisher:
        try:
            for args in numbers:
                res = add.apply_async((2, 2), publisher=publisher)
                results.append(res)
print([res.get() for res in results])

尽管这个特殊的例子更适合用一个群体来表达：
>>> from celery import group

>>> numbers = [(2, 2), (4, 4), (8, 8), (16, 16)]
>>> res = group(add.s(i, j) for i, j in numbers).apply_async()

>>> res.get()
[4, 8, 16, 32]

task_routes = {
    'celery.ping': 'default',
    'mytasks.add': 'cpu-bound',
    'feed.tasks.*': 'feeds',                           # <-- glob pattern
    re.compile(r'(image|video)\.tasks\..*'): 'media',  # <-- regex
    'video.encode': {
        'queue': 'video',
        'exchange': 'media',
        'routing_key': 'media.video.encode',
    },
}

task_routes = ('myapp.tasks.route_task', {'celery.ping': 'default'})
---------------------------------
def route_task(self, name, args, kwargs, options, task=None, **kw):
        if task == 'celery.ping':
            return {'queue': 'default'}
---------------------------------
Task.apply_async(immediate=False, exchange='video',
                 routing_key='video.compress')
-------------------------------
and a router returns:路由器返回：
{'immediate': True, 'exchange': 'urgent'}
------------------------------
the final message options will be:最终消息选项将是：
immediate=False, exchange='video', routing_key='video.compress'
--------------------------------
定义值 任务路由 优先于中定义的值 任务队列 当合并两者时。
With the follow settings:具有以下设置：
task_queues = {
    'cpubound': {
        'exchange': 'cpubound',
        'routing_key': 'cpubound',
    },
}

task_routes = {
    'tasks.add': {
        'queue': 'cpubound',
        'routing_key': 'tasks.add',
        'serializer': 'json',
    },
}
The final routing options for tasks.add will become:
的最终路由选项 tasks.add 将变为：
{'exchange': 'cpubound',
 'routing_key': 'tasks.add',
 'serializer': 'json'}