celery异步框架

一、什么是celery

Celery是一个简单、灵活且可靠的，处理大量消息的分布式系统

专注于实时处理的异步任务队列

同时也支持任务调度

二、Celery架构

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

2.1 消息中间件

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

2.2 任务执行单元

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

2.3 任务结果存储

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

2.4 版本支持情况

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.

三、使用场景

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

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

四、Celery的安装配置

pip install celery

消息中间件：RabbitMQ/Redis

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

五、Celery执行异步任务

5.1 创建py文件：celery_app_task.py

import celery
import time

# 指定broker（消息中间件），指定backend（结果存储）
broker = ‘redis://127.0.0.1:6379/0‘
backend = ‘redis://127.0.0.1:6379/1‘

# 实例化产生一个Celery对象 app=Celery(‘名字‘，broker，backend)
app = celery.Celery(‘test‘,backend=backend,broker=broker)

# 加装饰器绑定任务，任务就是函数，在函数（add）上加装饰器app.task，表示该任务是被celery管理的，并且可以用celery执行
@app.task
def add(x, y):
    time.sleep(1)
    return x + y

5.2 创建py文件：add_task.py，提交任务

from celery_app_task import add
# 提交任务到消息队列中，只是把任务提交到消息队列中，并没有执行函数
result = add.delay(4, 5)
print(result.id)

5.3 创建py文件：run.py，执行任务，或者使用命令执行：celery worker -A celery_app_task -l info

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

【注】执行命令：

　　linux：celery worker -A celery_app_task -l info

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

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

from celery.result import AsyncResult
from celery_app_task import app

async = AsyncResult(id="e919d97d-2938-4d0f-9265-fd8237dc2aa3", 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(‘任务已经开始被执行‘)

5.5 总结

-1 指定broker（消息中间件），指定backend（结果存储）
-2 实例化产生一个Celery对象 app=Celery(‘名字‘，broker，backend)
-3 加装饰器绑定任务，在函数（add）上加装饰器app.task
-4 其他程序提交任务,先导入add，add.delay(参，参数)，会将该函数提交到消息中间件，但是并不会执行，有个返回值，　　直接print会打印出任务的id，以后用id去查询任务是否执行完成
-5 启动worker去执行任务：  celery worker -A celery_task_s1 -l info   　　　　　　　　windows下： celery worker -A celery_task_s1 -l info -P eventlet
-6 查看结果：根据id去查询
　　async = AsyncResult(id="a5ea035f-0cc3-44ba-b334-f5d7c7ce681d", app=app)
　　if async.successful():
　　　　#取出它return的值
　　　　result = async.get()
　　　　print(result)

文件执行顺序：　　

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

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

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

六、多任务结构

6.1 任务结构

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

6.2 celery.py

from celery import Celery

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

# 时区
app.conf.timezone = ‘Asia/Shanghai‘
# 是否使用UTC
app.conf.enable_utc = False

6.3 task1.py

from celery import Celery

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

# 时区
app.conf.timezone = ‘Asia/Shanghai‘
# 是否使用UTC
app.conf.enable_utc = False

6.4 task2.py

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

6.5 send_task.py

from celery_task.tasks1 import test_celery1
from celery_task.tasks2 import test_celery2

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

6.6 check_result.py

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

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

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）

七、Celery执行定时任务

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

# 比如几点几分几秒执行某个任务，添加任务的时候

# 方式一
from datetime import datetime
v1 = datetime(2019, 7, 12, 11, 13, 56)
print(v1)
v2 = datetime.utcfromtimestamp(v1.timestamp())
print(v2)
#取出要执行任务的时间对象，调用apply_async方法，args是参数，eta是执行的时间
result = celery_task_s1.add.apply_async(args=[1, 3], eta=v2)

# 方式二
from celery_app_task import add
from datetime import datetime

ctime = datetime.now()
# 默认用utc时间
utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
from datetime import timedelta
# 取10秒之后的时间对象
time_delay = timedelta(seconds=10)
task_time = utc_ctime + time_delay

# 取出要执行任务的时间对象，调用apply_async方法，args是参数，eta是执行时间
result = add.apply_async(args=[4, 3], eta=task_time)
print(result.id)

7.2 类似于contab的定时任务（每天什么时候执行任务）

在多任务结构中celery.py修改如下：

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

cel = Celery(‘tasks‘, 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 = {
    # 名字随意命名
    ‘add-every-10-seconds‘: {
        # 执行tasks1下的test_celery函数
        ‘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)
    # },
}

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

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

八、django中使用celery

8.1 方式一（常用）

多任务结构直接拷过来，使用即可

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

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "untitled15.settings")
import django
django.setup()

8.2 django-celery（不常用）

8.2.1 安装包

celery==3.1.25
django-celery==3.1.20

8.2.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

8.2.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)

8.2.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‘)

8.2.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‘

原文地址：https://www.cnblogs.com/wangyong123/p/11782965.html

时间： 2024-08-02 07:13:19

celery异步框架的相关文章

celery异步执行任务框架

Celery 官方 Celery 官网:http://www.celeryproject.org/ Celery 官方文档英文版:http://docs.celeryproject.org/en/latest/index.html Celery 官方文档中文版:http://docs.jinkan.org/docs/celery/ Celery异步任务框架 """ 1)可以不依赖任何服务器,通过自身命令,启动服务(内部支持socket) 2)celery服务为为其他项目服务提

Django使用Celery异步任务队列

1  Celery简介 Celery是异步任务队列,可以独立于主进程运行,在主进程退出后,也不影响队列中的任务执行. 任务执行异常退出,重新启动后,会继续执行队列中的其他任务,同时可以缓存停止期间接收的工作任务,这个功能依赖于消息队列(MQ.Redis). 1.1  Celery原理 Celery的架构由三部分组成,消息中间件(message broker),任务执行单元(worker)和任务执行结果存储(task result store)组成. 消息中间件:Celery本身不提供消息服务,但

平台项目 ~ celery 异步之异步处理功能

一 简介:今天来聊聊celery两大功能之一的异步处理 二  标准流程: 1 建立 config 文件           class Config:             ENABLE_UTC = False             CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/5'             BROKER_URL = 'redis://127.0.0.1:6379/6'            注意 1 所有参数都必须大写,并且

基于SEDA的异步框架设计与实现

基于SEDA的异步框架设计与实现 二.为什么使用SEDA 目前,面对并发环境,主流互联网服务器编程模型有两种:多线程模型以及事件驱动模型.但是这两个模型都不足以解决这个问题.我们来首先看一下这两种编程模型. 1.多线程并发模型 多线程并发模型是目前最普遍的服务器编程模型,该模型的架构如下图所示:        该模型针对每一个请求,会为其创建并分配一个线程.该线程负责这个请求的处理.该模型的优点:执行粒度是整个完整的处理流程.处理逻辑清晰,容易开发.但与此同时缺点也很明显:如果处理过程中某一步骤

无废话Android之内容观察者ContentObserver、获取和保存系统的联系人信息、网络图片查看器、网络html查看器、使用异步框架Android-Async-Http（4）

1.内容观察者ContentObserver 如果ContentProvider的访问者需要知道ContentProvider中的数据发生了变化,可以在ContentProvider 发生数据变化时调用getContentResolver().notifyChange(uri, null)来通知注册在此URI上的访问者,例子如下: private static final Uri URI = Uri.parse("content://person.db"); public class

jQuery异步框架探究1:jQuery._Deferred方法

jQuery异步框架应用于jQuery数据缓存模块.jQuery ajax模块.jQuery事件绑定模块等多个模块,是jQuery的基础功能之一.实际上是jQuery实现的一个异步处理框架,从本质上讲与java aio没有区别,所以需要从更抽象层面的"异步处理"的视角分析解读该模块.这个部分与dom功能关系不大,是独立部分,可以看作是jQuery工具系列之一. 与异步框架相关的方法定义于jQuery类的静态方法中.只有三个方法,但是功能和应用及其强大!本篇详细讲解第一个方法jQuery

几个常用的异步框架和网络访问框架区分对比

Part1: 由于在我们的程序中,不允许一些耗时的任务在主线程中出现,主要是为了防止阻塞主线程而导致的 Anr(Application not Responding),一些耗时任务主要包括: 网络访问,缓慢的磁盘操作,比较耗时的算法 当我们的主线程在一定时间里对某一事件的处理超过一定时间后会主线程会崩溃报ANR, 通常的解决方案:采用子线程技术来将耗时任务与主线程进行脱离 1.handler机制 只需要将UI更新参数在子线程中使用sendMessage发送到定义好的Handler里的handle

python—Celery异步分布式

Celery异步分布式 Celery是一个python开发的异步分布式任务调度模块 Celery本身并不提供消息服务,使用第三方服务,也就是borker来传递任务,目前支持rebbimq,redis, 数据库等 使用redis连接url的格式为: redis://:[email protected]:port/db_number 例如: BROKER_URL = 'redis://localhost:6379/0' 1)huang.py from celery import Celery bro

Celery异步任务

在实际开发过程中,会遇到很多耗时操作,这时如果不采取措施,程序会进入到阻塞状态,直到耗时任务完成,为了保证整个项目的流畅性,通常会对这些耗时任务进行异步操作,具体步骤如下: 1.创建celery_tasks用于保存celery异步任务 2.在celery_tasks目录下创建config.py文件,用于保存celery的配置信息 broker_url = "redis://127.0.0.1/14" 3.在celery_tasks目录下创建main.py文件,用于作为celery的启动