参考：https://blog.csdn.net/tichimi3375/article/details/82415412

中文翻译：https://www.celerycn.io/      https://blog.csdn.net/weixin_40475396/article/details/80439781

官网：http://docs.celeryproject.org/en/latest/index.html

注意事项：

1 cd到tasks.py同级目录中，执行命令

2 导入配置的一种方式

3 my_task.apply_async((2, 2), queue='my_queue', countdown=10) 任务my_task将会被发送到my_queue队列中，并且在发送10秒之后执行。

4

5

group: 一组任务并行执行，返回一组返回值，并可以按顺序检索返回值。

  chain: 任务一个一个执行，一个执行完将执行return结果传递给下一个任务函数

from proj.tasks import my_task1
from proj.tasks import my_task2
from proj.tasks import my_task3
from celery import group

将多个signature放入同一组中

my_group = group((my_task1.s(, ), my_task2.s(, ), my_task3.s(, )))
ret = my_group() # 执行组任务
print(ret.get()) # 输出每个任务结果

from proj.tasks import my_task1
from proj.tasks import my_task2
from proj.tasks import my_task3
from celery import chain

将多个signature组成一个任务链

my_task1的运行结果将会传递给my_task2

my_task2的运行结果会传递给my_task3

my_chain = chain(my_task1.s(, ) | my_task2.s() | my_task3.s())
ret = my_chain() # 执行任务链
print(ret.get()) # 输出最终结果

6 my_task1.apply_async(queue='queue1')通过apply_aynsc()方法来设置任务发送到那个队列中

7 celery -A proj worker --loglevel=info -Q queue1,queue2 设置一个worker服务器处理两个队列中的任务

8 celery beat是一个调度器，它可以周期内指定某个worker来执行某个任务。

启动woker处理周期性任务: celery -A proj worker --loglevel=info --beat

beat_schedule = {
'every-5-minute':
{
'task': 'proj.tasks.period_task',
'schedule': 5.0,
'args': (, ),
},
'add-every-monday-morning': {
'task': 'proj.tasks.period_task',
'schedule': crontab(hour=, minute=, day_of_week=),
'args': (, ),
},

}

9

from celery import Celery

创建celery实例

app = Celery('demo')
app.config_from_object('proj.celeryconfig')

自动搜索任务

app.autodiscover_tasks(['proj'])

10 定义app时要指定名称，否则如下

当这个模块运行，任务将以前缀 __main__ 命名，但是当该模块被其他进程引入来运行一个任务，这个任务的名称将以前缀 tasks 命名（即这个模块的真实名称）

【app名称不指定，则task的名称不固定，就不方便根据任务名称映射出实际任务函数】

from celery import Celery
app = Celery() --app未指定名称

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

if __name__ == '__main__':
app.worker_main()

最佳实践如下：

>>> app = Celery('tasks')

app.main
'tasks'

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

add.name
tasks.add

参考：https://blog.csdn.net/libing_thinking/article/details/78541171

11 客户端导入模块 myapp.tasks 时使用 .tasks ，而工作单元导入模块使用 myapp.tasks， 他们产生的名称会不匹配，任务调用时工作单元会报 NotRegistered 错误。

>>> from project.myapp.tasks import mytask

mytask.name
'project.myapp.tasks.mytask'

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

基于这一点，你必须在导入模块时保持一致，这也是 python 的最佳实践

参考：https://blog.csdn.net/libing_thinking/article/details/78547816

12

celery_demo # 项目根目录
├── celery_app # 存放 celery 相关文件
│ ├── __init__.py
│ ├── celeryconfig.py # 配置文件
│ ├── task1.py # 任务文件
│ └── task2.py # 任务文件
└── client.py # 应用程序

执行 python client.py 就生产出了task

在celery_demo 目录下执行 celery worker -A celery_app -l info -Q email 就执行了任务

13

https://www.cnblogs.com/kangoroo/p/6588615.html

from celery.app.task import Task

class CallbackTask(Task):

def \_\_init\_\_(self):  
    super(CallbackTask, self).\_\_init\_\_()

def on_success(self, retval, task_id, args, kwargs):
try:
item_param= json.loads(args[])
logger.info('[task_id] %s, [task_type] %s, finished successfully.' % (task_id, item_param.get('task_type')))
except Exception, ex:
logger.error(traceback.format_exc())

def on\_failure(self, exc, task\_id, args, kwargs, einfo):  
    try:  
        item\_param = json.loads(args\[\])  
        logger.error(('Task {0} raised exception: {1!r}\\n{2!r}'.format(  
                task\_id, exc, einfo.traceback)))  
    except Exception, ex:  
        logger.error(traceback.format\_exc())

from celery import task
from common.callback import CallbackTask

logger = logging.getLogger(__name__)

@task(base=CallbackTask)
def quota_check(item_param):
logger.info('start')
return

14

result = add.delay(``4``, 4``)

>>> result.ready()
False

You can wait for the result to complete, but this is rarely used since it turns the asynchronous call into a synchronous one:

>>> result.get(timeout=1)
8

In case the task raised an exception, get() will re-raise the exception, but you can override this by specifying the propagate argument:

>>> result.get(propagate=False)

If the task raised an exception you can also gain access to the original traceback:

>>> result.traceback

15 添加定时任务的一种方法

from celery import Celery
from celery.schedules import crontab

app = Celery()

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
# Calls test('hello') every 10 seconds.
sender.add_periodic_task(10.0, test.s('hello'), name='add every 10')

# Calls test('world') every 30 seconds  
sender.add\_periodic\_task(30.0, test.s('world'), expires=10)

# Executes every Monday morning at 7:30 a.m.  
sender.add\_periodic\_task(  
    crontab(hour=7, minute=30, day\_of\_week=1),  
    test.s('Happy Mondays!'),  
)

@app.task
def test(arg):
print(arg)

另一种方法：
Example: Run the tasks.add task every seconds.

app.conf.beat_schedule = {
'add-every-30-seconds': {
'task': 'tasks.add',
'schedule': 30.0,
'args': (, )
},
}
app.conf.timezone = 'UTC'

16 查看结果：

from celery.result import AsyncResult
from celery_app_task import cel

async = AsyncResult(id="e919d97d-2938-4d0f-9265-fd8237dc2aa3", app=cel)

if async.successful():
result = async.get()
print(result)
# result.forget() # 将结果删除
elif async.failed():
print('执行失败')
elif async.status == 'PENDING':
print('任务等待中被执行')
elif async.status == 'RETRY':
print('任务异常后正在重试')
elif async.status == 'STARTED':
print('任务已经开始被执行')

==========================

ret=task1.add.apply_async(args=[2, 8],queue="email",routing_key="email")
print('hello world')
print(ret.status)
print(ret.id)

print(ret.result) # NONE
print(ret.get()) #没有加timeout，所以阻塞住了，直到返回结果10
print(ret.result) # 10

17