Celery简介

一. 什么是Clelery

1. 简介

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

2. Celery架构

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

• 消息中间件

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

• 任务执行单元

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

• 任务结果存储

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

3. 版本支持情况

# 1.新版本支持到3.5
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.

# 2.之后开发的版本不在支持2.7 
    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.

# 3.对于window支持的不够好
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. 消息队列

简介

博客地址： https://www.jianshu.com/p/689ce4205021

消息队列中间件是分布式系统中重要的组件，主要解决应用耦合、异步消息、流量削锋等问题。实现高性能、高可用、可伸缩和最终一致性架构。是大型分布式系统不可缺少的中间件。
目前在生产环境，使用较多的消息队列有ActiveMQ、RabbitMQ、ZeroMQ、Kafka、MetaMQ、RocketMQ等。

作用

• 异步的提交任务
• 对某些业务进行了解耦合（例如订单系统和库存系统）
• 流量削峰：秒杀活动，一般会因为流量过大，导致流量暴增，应用挂掉。为解决这个问题，一般需要在应用前端加入消息队列。

二. 使用场景

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

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

三. Celery的安装配置

pip install celery
消息中间件：RabbitMQ/Redis
app=Celery('任务名'，backend='xxx',broker='xxx')

四. Celery执行异步任务

1. 基本使用

1）任务的创建

import celery
import time
# broker='redis://127.0.0.1:6379/2' 不加密码
backend='redis://:123456@127.0.0.1:6379/1'
broker='redis://:123456@127.0.0.1:6379/2'
cel=celery.Celery('test',backend=backend,broker=broker)
@cel.task
def add(x,y):
    return x+y

2）任务的提交

from celery_app_task import add
result = add.delay(4,5)
print(result.id)

3）任务的执行

celery worker -A celery_app_task -l info
注：windows下：celery worker -A celery_app_task -l info -P eventlet

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

4）结果的查看

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

2. 多任务结构

1）架构图

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

2）celery.py文件的编写

from celery import Celery
cel = Celery('celery_task',
             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 # 是否使用UTC

3）任务的编写

# task1的文件
import time
from celery_task.celery import cel

@cel.task
def test_celery(res):
    time.sleep(5)
    return "test_celery任务结果:%s"%res


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

3）check_result.py结果检测

from celery.result import AsyncResult
from celery_task.celery import cel

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

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

# result.forget() # 将结果删除,执行完成，结果不会自动删除
# async.revoke(terminate=True)  # 无论现在是什么时候，都要终止
# async.revoke(terminate=False) # 如果任务还没有开始执行呢，那么就可以终止。

4) send_task.py任务发送到队列

from celery_task.tasks1 import test_celery
from celery_task.tasks2 import test_celery2

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

5）启动异步任务框架

添加任务（执行send_task.py），
开启work：celery worker -A celery_task -l info  -P  eventlet
检查任务执行结果（执行check_result.py）

五. Celery执行定时任务

1. 延迟任务

1）方式一

from celery_app_task import add
from datetime import datetime

v1 = datetime(2019, 2, 13, 18, 19, 56)   # 创建一个时间对象
v2 = datetime.utcfromtimestamp(v1.timestamp())
result = add.apply_async(args=[1, 3], eta=v1)  # 如果args没有参数，就写成空列表，不能不写

2）方式二

from datetime import timedelta, datetime

ctime = datetime.now()
utc_ctime = datetime.utcfromtimestamp(ctime.timestamp()) # 默认用utc时间
task_time = utc_ctime + timedelta(seconds=10)

# 使用apply_async并设定时间
result = add.apply_async(args=[4, 3], eta=task_time)

2. 定时任务

1）编写

from datetime import timedelta
from celery import Celery
from celery.schedules import crontab

cel = Celery('celery_task', 
             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 = {
    # 名字随意命名，每10秒发送一个任务到队列中
    'add-every-10-seconds': {
        # 必须要注册到include中
        '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)
    # },
}

2）启动

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

六. Django中使用Celery

1. 安装包

# 在windows下面固定的版本
celery==3.1.25
django-celery==3.1.20

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

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)

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

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'