celery介绍

Celery 官网：http://www.celeryproject.org/

Celery 官方文档英文版：http://docs.celeryproject.org/en/latest/index.html

Celery 官方文档中文版：http://docs.jinkan.org/docs/celery/

celery配置：http://docs.jinkan.org/docs/celery/configuration.html#configuration

参考：http://www.cnblogs.com/landpack/p/5564768.html    http://blog.csdn.net/happyAnger6/article/details/51408266

http://www.cnblogs.com/forward-wang/p/5970806.html

分布式队列神器 Celery：https://segmentfault.com/a/1190000008022050

celery最佳实践：https://my.oschina.net/siddontang/blog/284107

Celery 分布式任务队列快速入门：http://www.cnblogs.com/alex3714/p/6351797.html

异步任务神器 Celery 快速入门教程：https://blog.csdn.net/chenqiuge1984/article/details/80127446

定时任务管理之python篇celery使用：http://student-lp.iteye.com/blog/2093397

异步任务神器 Celery：http://python.jobbole.com/87086/

celery任务调度框架实践：https://blog.csdn.net/qq_28921653/article/details/79555212

Celery-4.1 用户指南: Monitoring and Management Guide：https://blog.csdn.net/libing_thinking/article/details/78592801

Celery安装及使用：https://blog.csdn.net/u012325060/article/details/79292243

Celery学习笔记（一）：https://blog.csdn.net/sdulsj/article/details/73741350

 

 

celery简介

1.什么是Clelery

'''
Celery是一个简单、灵活且可靠的，处理大量消息的分布式系统

专注于实时处理的异步任务队列

同时也支持任务调度
'''

　

Celery架构

 Celery的架构由三部分组成，消息中间件（message broker），任务执行单元（worker）和任务执行结果存储（task result store）组成。

 

消息中间件 broker

'''
Celery本身不提供消息服务，但是可以方便的和第三方提供的消息中间件集成。包括，RabbitMQ, Redis等等
broker是一个消息传输的中间件，可以理解为一个邮箱。每当应用程序调用celery的异步任务的时候，会向broker传递消息，
而后celery的worker将会取到消息，进行对于的程序执行。
这个邮箱可以看成是一个消息队列。其中Broker的中文意思是 经纪人 ，其实就是一开始说的 消息队列 ，
用来发送和接受消息。这个Broker有几个方案可供选择：RabbitMQ 
(消息队列)，<a href="http://lib.csdn.net/base/redis" rel="nofollow">Redis</a>（缓存数据库），
<a href="http://lib.csdn.net/base/mysql" rel="nofollow">数据库</a>（不推荐），等等
'''

　　

 

任务执行单元 Worker

Worker是Celery提供的任务执行的单元，worker并发的运行在分布式的系统节点中。　

任务结果存储

Task result store用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括AMQP, redis等

　　

2.使用场景

'''
异步任务：将耗时操作任务提交给Celery去异步执行，比如发送短信/邮件、消息推送、音视频处理等等

定时任务：定时执行某件事情，比如每天数据统计
'''

　　

3.Celery的安装配置

'''
pip install celery

消息中间件：RabbitMQ/Redis  

app=Celery('任务名'，backend='xxx',broker='xxx')
'''

　　

4.Celery执行异步任务

4.1 创建py文件：celery_app_task.py

import celery
import time
# 指定broker（消息中间件），指定backend（结果存储）
broker = 'redis://127.0.0.1:6379/0'
backend = 'redis://127.0.0.1:6379/1'
# 实例化产生一个Celery对象 app=Celery('名字'，broker，backend)
app = celery.Celery('test',backend=backend,broker=broker)
# 加装饰器绑定任务，任务就是函数，在函数（add）上加装饰器app.task，表示该任务是被celery管理的，并且可以用celery执行
@app.task
def add(x, y):
    time.sleep(1)
    return x + y

　　　

4.2 创建py文件：add_task.py，提交任务

from celery_app_task import add
# 提交任务到消息队列中，只是把任务提交到消息队列中，并没有执行函数
result = add.delay(4, 5)
print(result.id)

4.3 创建py文件：run.py，执行任务，或者使用命令执行：celery worker -A celery_app_task -l info   

from celery_app_task import app
if __name__ == '__main__':
    app.worker_main()
    # app.worker_main(argv=['--loglevel=info')

　　

【注】执行命令：
　　linux：celery worker -A celery_app_task -l info   
　　windows：celery worker -A celery_app_task -l info -P eventlet

　　

4.4 创建py文件：result.py，查看任务执行结果

from celery.result import AsyncResult
from celery_app_task import app

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

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('任务已经开始被执行')

　　

4.5 总结

'''
-1 指定broker（消息中间件），指定backend（结果存储）
-2 实例化产生一个Celery对象 app=Celery('名字'，broker，backend)
-3 加装饰器绑定任务，在函数（add）上加装饰器app.task
-4 其他程序提交任务,先导入add，add.delay(参，参数)，会将该函数提交到消息中间件，但是并不会执行，有个返回值，
　　直接print会打印出任务的id，以后用id去查询任务是否执行完成
-5 启动worker去执行任务：  celery worker -A celery_task_s1 -l info   
　　　　　　　　windows下： celery worker -A celery_task_s1 -l info -P eventlet
-6 查看结果：根据id去查询
　　async = AsyncResult(id="a5ea035f-0cc3-44ba-b334-f5d7c7ce681d", app=app)
　　if async.successful():
　　　　#取出它return的值
　　　　result = async.get()
　　　　print(result)
'''

'''
文件执行顺序：　　
　　执行 add_task.py，添加任务，并获取任务ID
　　执行 run.py ，或者执行命令：celery worker -A celery_app_task -l info
　　执行 result.py,检查任务状态并获取结果
'''

　　注意:

此时如果python版本大于3.6可能会报错  参考博客

五、多任务结构

5.1 任务结构

pro_cel
├── celery_task # celery相关文件夹
│ ├── celery.py # celery连接和配置相关文件,必须叫这个名字
│ └── tasks1.py # 所有任务函数
│    └── tasks2.py # 所有任务函数
├── check_result.py # 检查结果
└── send_task.py # 触发任务

　　

5.2 celery.py

from celery import Celery

app = Celery('celery_demo',
             broker='redis://127.0.0.1:6379/1',
             backend='redis://127.0.0.1:6379/2',
             # 包含以下两个任务文件，去相应的py文件中找任务，对多个任务做分类
             include=['celery_task.tasks1',
                      'celery_task.tasks2'
                      ])

# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False

　　

5.3 task1.py

from celery import Celery

app = Celery('celery_demo',
             broker='redis://127.0.0.1:6379/1',
             backend='redis://127.0.0.1:6379/2',
             # 包含以下两个任务文件，去相应的py文件中找任务，对多个任务做分类
             include=['celery_task.tasks1',
                      'celery_task.tasks2'
                      ])

# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False

　　

5.4 task2.py

import time
from celery_task.celery import app
@app.task
def test_celery2(res):
    time.sleep(5)
    return "test_celery2任务结果:%s" % res

　　

5.5 send_task.py

from celery_task.tasks1 import test_celery1
from celery_task.tasks2 import test_celery2

# 立即告知celery去执行test_celery任务，并传入一个参数
result = test_celery1.delay('第一个的执行')
print(result.id)
result = test_celery2.delay('第二个的执行')
print(result.id)

　　

5.6 check_result.py

from celery.result import AsyncResult
from celery_task.celery import app

async = AsyncResult(id="08eb2778-24e1-44e4-a54b-56990b3519ef", app=app)

if async.successful():
    result = async.get()
    print(result)
    # result.forget() # 将结果删除，执行完成，结果不会自动删除
    # async.revoke(terminate=True)  # 无论现在是什么时候，都要终止
    # async.revoke(terminate=False) # 如果任务还没有开始执行呢，那么就可以终止。
elif async.failed():
    print('执行失败')
elif async.status == 'PENDING':
    print('任务等待中被执行')
elif async.status == 'RETRY':
    print('任务异常后正在重试')
elif async.status == 'STARTED':
    print('任务已经开始被执行')

　　

添加任务（执行send_task.py），

开启work：celery worker -A celery_task -l info -P eventlet，

检查任务执行结果（执行check_result.py）

六、Celery执行定时任务

6.1 设定时间让celery执行一个任务

# 比如几点几分几秒执行某个任务，添加任务的时候

# 方式一
from datetime import datetime
v1 = datetime(2019, 7, 12, 11, 13, 56)
print(v1)
v2 = datetime.utcfromtimestamp(v1.timestamp())
print(v2)
#取出要执行任务的时间对象，调用apply_async方法，args是参数，eta是执行的时间
result = celery_task_s1.add.apply_async(args=[1, 3], eta=v2)

# 方式二
from celery_app_task import add
from datetime import datetime

ctime = datetime.now()
# 默认用utc时间
utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
from datetime import timedelta
# 取10秒之后的时间对象
time_delay = timedelta(seconds=10)
task_time = utc_ctime + time_delay

# 取出要执行任务的时间对象，调用apply_async方法，args是参数，eta是执行时间
result = add.apply_async(args=[4, 3], eta=task_time)
print(result.id)

　　

6.2 类似于contab的定时任务（每天什么时候执行任务）

# 在多任务结构中celery.py修改如下：
from datetime import timedelta
from celery import Celery
from celery.schedules import crontab

cel = Celery('tasks', broker='redis://127.0.0.1:6379/1', backend='redis://127.0.0.1:6379/2', include=[
    'celery_task.tasks1',
    'celery_task.tasks2',
])
cel.conf.timezone = 'Asia/Shanghai'
cel.conf.enable_utc = False

cel.conf.beat_schedule = {
    # 名字随意命名
    'add-every-10-seconds': {
        # 执行tasks1下的test_celery函数
        'task': 'celery_task.tasks1.test_celery',
        # 每隔2秒执行一次
        # 'schedule': 1.0,
        # 'schedule': crontab(minute="*/1"),
        'schedule': timedelta(seconds=2),
        # 传递参数
        'args': ('test',)
    },
    # 'add-every-12-seconds': {
    #     'task': 'celery_task.tasks1.test_celery',
    #     每年4月11号，8点42分执行
    #     'schedule': crontab(minute=42, hour=8, day_of_month=11, month_of_year=4),
    #     'schedule': crontab(minute=42, hour=8, day_of_month=11, month_of_year=4),
    #     'args': (16, 16)
    # },
}

　　

启动一个beat：celery beat -A celery_task -l info
启动work执行：celery worker -A celery_task -l info -P eventlet

　　

七、django中使用celery

7.1 方式一（常用）

多任务结构直接拷过来，使用即可

在celery的任务函数中不能直接调用django的环境，需要手动添加

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "untitled15.settings")
import django
django.setup()

　　

7.2 django-celery（不常用）

7.2.1 安装包

celery==3.1.25
django-celery==3.1.20

　　

7.2.2 在项目目录下创建celeryconfig.py

import djcelery
djcelery.setup_loader()
CELERY_IMPORTS=(
    'app01.tasks',
)
#有些情况可以防止死锁
CELERYD_FORCE_EXECV=True
# 设置并发worker数量
CELERYD_CONCURRENCY=4
#允许重试
CELERY_ACKS_LATE=True
# 每个worker最多执行100个任务被销毁，可以防止内存泄漏
CELERYD_MAX_TASKS_PER_CHILD=100
# 超时时间
CELERYD_TASK_TIME_LIMIT=12*30

　　

7.2.3 在app01目录下创建tasks.py

from celery import task
@task
def add(a,b):
    with open('a.text', 'a', encoding='utf-8') as f:
        f.write('a')
    print(a+b)

　　

7.2.4 视图函数views.py

from django.shortcuts import render,HttpResponse
from app01.tasks import add
from datetime import datetime
def test(request):
    # result=add.delay(2,3)
    ctime = datetime.now()
    # 默认用utc时间
    utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
    from datetime import timedelta
    time_delay = timedelta(seconds=5)
    task_time = utc_ctime + time_delay
    result = add.apply_async(args=[4, 3], eta=task_time)
    print(result.id)
    return HttpResponse('ok')

　　

7.2.5 settings.py

INSTALLED_APPS = [
    ...
    'djcelery',
    'app01'
]

...

from djagocele import celeryconfig
BROKER_BACKEND='redis'
BOOKER_URL='redis://127.0.0.1:6379/1'
CELERY_RESULT_BACKEND='redis://127.0.0.1:6379/2'