celery

时间 2019-11-19

标签 celery 繁體版

原文原文链接

前言：在正式学习以前，先简单的对celery作一个介绍！redis

1.什么是celery

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

2.celery架构

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

3.消息中间件

　　celery自己不提供消息服务，可是能够方便你的和第三方提供的消息中间件集成，包括，RabbitMQ, Redis等等架构

4.任务执行单元

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

5.任务结果存储

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

1、版本支持状况

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.

2、使用场景

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

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

页面静态化：耗时操做均可以用cerely分布式

3、Celery的安装配置

pip install celery

消息中间件：RabbitMQ/Redis函数

使用

使用步骤以下

　　写一个py文件：celery_task

　　　　1.指定broker(消息中间件)，指定backend（用于结果存储）

　　　　2.实例化产生一个celery对象　app=Celery('任务名'，backend='xxx',broker='xxx')

　　　　3.加装饰器绑定任务，在函数(add)上加装饰器app.task

　　　　4.其余程序提交任务，先导入add,add.delay（参数1，参数2），会将该函数提交到消息中间件，可是并不会执行，有个返回值，直接print，会获得你执行的一个id

　　　5 启动worker去执行任务

　　　　6.查看任务执行的结果，根据前面的id去检查，由于任务是异步，因此要从新写一个py文件

启动worker的命令

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

查看任务的执行结果

async = AsyncResult(id="a5ea035f-0cc3-44ba-b334-f5d7c7ce681d", app=app)
    if async.successful():
        #取出它return的值
        result = async.get()
        print(result)

4、Celery执行异步任务

基本使用

建立项目celerytest

建立py文件：celery_app_task.py

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

建立py文件：add_task.py，添加任务

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

建立py文件：run.py，执行任务

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

或者使用命令执行

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

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

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

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

执行 run.py ，或者执行命令：celery worker -A celery_app_task -l info

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

多任务结构

　　新建一个项目，项目结构以下

　　　　　　　　　　pro_cel
                    ├── celery_task# celery相关文件夹
                    │   ├── celery.py   # celery链接和配置相关文件,必须叫这个名字
                    │   └── order_task.py    #  全部任务函数
                    │   └── user_task.py    #  全部任务函数
                    ├── check_result.py # 检查结果
                    └── send_task.py    # 触发任务

注意：因为后面执行worker命令的缘由，因此文件夹里面的名字必须叫作celery

celery.py内容

#必须叫celery，生成celery对象

from celery import  Celery
from datetime import timedelta
from celery.schedules import crontab
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',
                    ]
           )

主要就是作的生成celery对象，有多个任务就去相应的任务下面执行具体的代码

order_task.py内容

from celery_task.celery import app
@app.task
def order_add(x,y):
    import time
    time.sleep(1)
    return x+y

user_task.py

from celery_task.celery import app
@app.task
def user_add(x,y):
    import time
    time.sleep(1)
    return x+y

添加任务send_task.py

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

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

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()
    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）

celer执行定时任务

　　设定时间让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)

天天的何时执行任务

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

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

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

django中使用celery

对于多任务结构，直接拷贝过来直接使用，注意，因为celery的任务函数不能直接调用django的环境，因此咱们须要手动添加以下几行代码，这样就能够正常使用了

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

django-celery

安装包

celery==3.1.25
django-celery==3.1.20

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

可是，这个要求比较高，有一个不符合就不会成功，因此咱们推荐使用第一个，就是直接将整个celery的项目文件拷贝过来使用，只须要本身作一个django环境的配置便可！