Celery

1、什么是Celery

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

执行异步任务：

创建任务：tasks.py
把任务添加到队列中：add_task.py
开启work，执行任务
- 用命令：celery -A tasks worker -l info
- 在 Windows下：celery -A tasks worker -l info -P eventlet
查看任务结果：task_resut.py

多任务结构：

重点：执行work的时候：celery -A tasks worker -l info -P eventlet

2、Celery架构

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

消息中间件

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

任务执行单元

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

任务结果存储

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

3、使用场景

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

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

4、Celery的安装配置

pip install celery

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

5、Celery执行异步任务

基本使用

创建项目：celerytest

创建py文件：task.py

`from celery import Celery

import time

# broker:消息中间人用redis

broker = 'redis://127.0.0.1:6397/1'

# 结果存储在redis中

backend = 'redis://127.0.0.1:6379/2'

# 第一个参数是别名，可以随便写

app = Celery('test', broker=broker, backend=backend)

@app.task

def add(x, y):

    time.sleep(3)

    return x + y

创建py文件：add_task.py,添加任务

import task

if __name__ == '__main__':

    # 之前这样写，直接就执行 函数

    task.add()

    # 现在把函数添加到执行队列中，参数写在delay中

    # result不是函数的执行结果，他是个对象

    result = task.add.delay(2, 3)

    # 这个任务唯一的id

    print(result.id)

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

from celery.result import AsyncResult

from task import app

async = AsyncResult(id='ac2a7e52-ef66-4caa-bffd-81414d869f85', 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('任务已经开始被执行')

执行add_task.py,添加任务，并获取任务ID

执行命令：celery worker -A celery_app_task -l info -P eventlet

执行result.py检查任务状态并获取结果

6、Celery执行定时任务

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

add_task.py

from celery_app_task import add

from datetime import datetime

# 方式一

# v1 = datetime(2019, 2, 13, 18, 19, 56)

# print(v1)

# v2 = datetime.utcfromtimestamp(v1.timestamp())

# print(v2)

# result = add.apply_async(args=[1, 3], eta=v2)

# print(result.id)

# 方式二

ctime = datetime.now()

# 默认用utc时间

utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())

from datetime import timedelta

time_delay = timedelta(seconds=10)

task_time = utc_ctime + time_delay

# 使用apply_async并设定时间

result = add.apply_async(args=[4, 3], eta=task_time)

print(result.id)

启动一个beat: celery beat -A celery_task -l info

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

7、Django中使用Celery

在项目目录下创建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

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

视图函数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')

settings.py

INSTALLED_APPS = [

    ...

    'djcelery',

    'app01'

]

...

from django_celery import celeryconfig

BROKER_BACKEND='redis'

BOOKER_URL='redis://127.0.0.1:6379/1'

CELERY_RESULT_BACKEND='redis://127.0.0.1:6379/2'

巴特西

Celery的基本使用