Celery的基本使用

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'