Celery分布式异步任务框架

执行celery： 

python manage.py celery -A TSDRM work -l info
python manage.py celery -A TSDRM flower -l info

测试项目中的redis是否连接：

 

一、什么是Celery

Celery是一个简单、灵活且可靠的，处理大量消息的分布式系统。专注于实时处理的异步任务队列，同时也支持定时任务

 

二、Celery架构

1、Celery的架构由三部分组成：

　　消息中间件（message broker）：

　　任务执行单元（worker）：

　　任务执行结果存储（task result store）：

user是提交任务的人，user把任务提交到消息中间件broker中，比如用户提交个3+5的任务，这个程序只负责提交，另一个程序负责执行。

workers负责执行代码，在消息中间件取一个一个的任务去执行，执行的结果需要存储在task result store中，用户user只需要去存储结果的task result store中看一下就知道这个任务有没有执行。

2、组成部分介绍

消息中间件：Celery本身不提供消息服务，但是可以方便的和第三方提供的消息中间件集成。包括，RabbitMQ, Redis等等

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

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

3、版本支持情况

Celery version 4.0 runs on
        Python ❨2.7, 3.4, 3.5❩
        PyPy ❨5.4, 5.5❩
    This is the last version to support Python 2.7, and from the next version (Celery 5.x) Python 3.5 or newer is required.

    If you’re running an older version of Python, you need to be running an older version of Celery:

        Python 2.6: Celery series 3.1 or earlier.
        Python 2.5: Celery series 3.0 or earlier.
        Python 2.4 was Celery series 2.2 or earlier.

    Celery is a project with minimal funding, so we don’t support Microsoft Windows.
　　 Please don’t open any issues related to that platform.

4、场景

异步任务：将耗时操作任务提交给Celery去异步执行，比如发送短信/邮件、消息推送、音视频处理，统计某个商品近半年的销量生成折线图等等

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

场景1：用户提交一个任务，把视频转成MP4格式。传上去之后转格式需要耗费很长时间。所以就可以把任务提交到消息中间件里，只返回一个消息告诉用户，视频正在转，你该干嘛就干嘛去。开启workes执行传视频的任务，视频转完之后，把结果放在结果存储store中。然后用户再发请求去结果存储中看一下视频有没有转好。

场景2：注册发邮件。一个用户一旦注册了之后。把发邮件的任务提交到消息中间件里。异步调用发邮件发邮件。所以当用户注册成功之后，后台有邮件正在发。

场景3：秒杀商品，把秒杀的任务提交到任务队列中，起个worker去执行。

一点击抢购，显示你正在排队中。正常的逻辑是，点击抢购，会到数据库里查看是否还有此商品，有的话，数据库商品就减一，生成一个订单返回给用户，用户进行支付。但是，你有没有想过，秒杀的时候，一瞬间来了好多好多人，这时候就会考虑到程序运行效率问题，解决方法是会在秒杀商品的时候提前把数据放在redis中，不在直接操作数据库而是redis。如果程序还不行的时候，就会对任务做缓冲，来个人把任务提交上去，来个人把任务提交上去，后面开启多个workers来执行任务，这就相当于把整个任务做成异步的，任务提交完成立马返回，你正在排队。

 场景4：利用Celery可以做页面静态化。首页经常被访问，每次都要查数据库非常耗时，可以把首页做成静态页面，数据是写死的，谁过来访问，就能看到首页。

 

三、Celery执行异步任务

安装依赖

pip3 install celery
pip3 install redis

1、基本使用

①创建celeryDemo项目

②创建任务：新建py文件：celery_task_s1.py

from celery import Celery
# 指定broker(消息中间件)，指定backend(结果储存)
backend = 'redis://127.0.0.1:6379/0'
broker = 'redis://127.0.0.1:6379/1'   # 0,1指定的是redis中的库

# 实例化产生一个Celery对象，一定要指定一个名字
app = Celery('test', broker=broker, backend=backend)

# 指定任务，就是一个函数
# 需要用一个装饰器绑定任务，表示该任务是被celery管理的，并且可以用celery执行的
@app.task
def add(x, y):
    import time
    time.sleep(3)
    return x+y

③提交任务到消息中间件中，获取任务id：新建py文件：add_task.py

# 用于提交任务的py文件
import celery_task_s1
# res = celery_task_s1.add(3, 7)  # 正常同步执行任务
# print(res)  # 打印结果为 10

# 提交任务到消息中间件中
res = celery_task_s1.add.delay(3, 7)  # 只是把任务提交到消息中间件中，并没有执行
print(res)   # 6e6464e8-3b0a-49f9-ad92-824725c2a573 任务的id号

③任务提交完，启动worker执行任务

注意：windows下启动需要安装依赖

pip3 install eventlet

命令执行：celery worker -A celery_task_s1 -l info

windows下：celery worker -A celery_task_s1 -l info -P eventlet

④获取任务执行结果：新建py文件：celery_result.py

from celery.result import AsyncResult
from celery_task_s1 import app

async = AsyncResult(id="6e6464e8-3b0a-49f9-ad92-824725c2a573", app=app)   # id是worker执行任务后拿到的id

if async.successful():
    # 取出它return的值
    result = async.get()
    print(result)     # 执行成功，打印的结果为10
    # result.forget() # 将结果删除
elif async.failed():
    print('执行失败')
elif async.status == 'PENDING':
    print('任务等待中被执行')
elif async.status == 'RETRY':
    print('任务异常后正在重试')
elif async.status == 'STARTED':
    print('任务已经开始被执行')

2、多任务结构

celery_mul_Demo
    ├── celery_task   　　   # celery相关文件夹
    │   ├── celery.py  　　  # celery连接和配置相关文件,必须叫这个名字
    │   └── order_task.py   #  所有任务函数
    │   └── user_task.py    #  所有任务函数
    ├── add_task.py 　　     # 提交任务
    └── celery_result.py    # 查看结果

①创建任务：

新建py文件：celery.py(必须叫这个名字)　　

# 这个文件必须叫celery，生成celery对象

from celery import Celery

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

新建py文件：order_task.py

from celery_task.celery import app

@app.task
def order_add(x, y):
    import time
    time.sleep(3)
    return x+y

新建py文件：user_task.py

from celery_task.celery import app

@app.task
def user_add(x, y):
    import time
    time.sleep(3)
    return x+y

②提交任务：新建py文件：add_task.py

from celery_task.order_task import order_add
from celery_task.user_task import user_add

# res = order_add.delay(10, 20)   # 打印结果为 26a025a6-f910-4e71-8173-c119c665df84
res = user_add.delay(100, 200)    # 打印结果为 c67e5176-3972-49c2-9e13-13f7d6eea4af
print(res)

③任务提交完，启动worker执行任务

windows下：celery worker -A celery_task -l info -P eventlet

等待提交任务

提交任务order_task

提交任务user_task

④获取任务执行结果：新建py文件：celery_result.py

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

async = AsyncResult(id="c67e5176-3972-49c2-9e13-13f7d6eea4af", app=app)   # id是worker执行user_task任务后拿到的id

if async.successful():
    # 取出它return的值
    result = async.get()
    print(result)     # 执行成功，打印的结果为300
    # result.forget() # 将结果删除
elif async.failed():
    print('执行失败')
elif async.status == 'PENDING':
    print('任务等待中被执行')
elif async.status == 'RETRY':
    print('任务异常后正在重试')
elif async.status == 'STARTED':
    print('任务已经开始被执行')

四、celery执行定时任务

1、指定的几时几点几分执行任务

①创建celeryDemo项目

②创建任务：新建py文件：celery_task_s1.py

from celery import Celery
# 指定broker(消息中间件)，指定backend(结果储存)
backend = 'redis://127.0.0.1:6379/0'
broker = 'redis://127.0.0.1:6379/1'   # 0,1指定的是redis中的库

# 实例化产生一个Celery对象，一定要指定一个名字
app = Celery('test', broker=broker, backend=backend)

# 指定任务，就是一个函数
# 需要用一个装饰器绑定任务，表示该任务是被celery管理的，并且可以用celery执行的
@app.task
def add(x, y):
    import time
    time.sleep(3)
    return x+y

③提交任务到消息中间件中，获取任务id：新建py文件：add_task.py

# 用于提交任务的py文件
import celery_task_s1

# 执行定时任务:指定时间执行add_task任务
from datetime import datetime
v1 = datetime(2019, 7, 12, 20, 59, 56)   # 指定的几时几点几分执行任务
print(v1)   # 2019-07-12 20:59:56
v2 = datetime.utcfromtimestamp(v1.timestamp())
print(v2)  # 2019-07-12 12:59:56
# 取出要执行任务的时间对象，调用apply_async方法，args是参数，eta是执行的时间
result = celery_task_s1.add.apply_async(args=[1, 3], eta=v2)
print(result.id)   # e28be60b-1546-4b92-987f-dfbf357e623f

③任务提交完，启动worker执行任务

windows下：celery worker -A celery_task_s1 -l info -P eventlet

指定的时间就会执行任务得到结果

2、每天的什么时候执行任务，每天到这个时间点都会执行，循环执行的

①创建项目celery_mul_Demo

②创建任务：新建py文件：celery.py(必须叫这个名字)　　

# 这个文件必须叫celery，生成celery对象


from celery import Celery

broker = 'redis://127.0.0.1:6379/0'
backend = 'redis://127.0.0.1:6379/1'
app = Celery('test', broker=broker, backend=backend,
             include=['celery_task.order_task',
                      ]
             )

from datetime import timedelta

app.conf.beat_schedule = {
    # 名字随意命名
    'add-every-10-seconds': {
        # 执行tasks1下的test_celery函数
        'task': 'celery_task.order_task.order_add',
        # 每隔3秒执行一次
        # 'schedule': 1.0,
        # 'schedule': crontab(minute="*/1"),
        'schedule': timedelta(seconds=3),
        # 传递参数
        'args': (5, 6)
    },
    # 每年什么时间执行什么
    # 'add-every-12-seconds': {
    #     'task': 'celery_task.order_task.order_add',
    #     # 每年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)
    # },
}

新建py文件：order_task.py

from celery_task.celery import app

@app.task
def order_add(x, y):
    import time
    time.sleep(3)
    return x+y

③这里必须启动一个beat：celery beat -A celery_task -l info

再启动work执行：celery worker -A celery_task -l info -P eventlet

五、在django中使用celery

①创建项目：django_celery_Demo

②把多任务结构写的文件夹celery_task直接拷贝到项目中

③views.py

from celery_task.user_task import user_add

def index(request):
    res = user_add.delay(3, 7)
    return HttpResponse(res.id)

def check_result(request):
    res = request.GET.get('id')
    from celery.result import AsyncResult
    from celery_task.celery import app
    async = AsyncResult(id=res, app=app)  # id是worker执行任务后拿到的id
    if async.successful():
        # 取出它return的值
        res = async.get()
        print(res)
        return HttpResponse('ok')

④url.py

url(r'^index/', views.index),
url(r'^check/', views.check_result),

⑤在命令行中先启动worker：celery worker -A celery_task -l info -P eventlet

⑥启动项目

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

import os
if __name__ == '__main__':
    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "untitled15.settings")
    import django
    django.setup()

    from app01 import models

    books = models.Book.objects.all()
    print(books)