Celery分布式任务队列

1.什么是Clelery

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

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

同时也支持任务调度

Celery架构

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

消息中间件

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

任务执行单元

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

任务结果存储

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

版本支持情况

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去异步执行,比如发送短信/邮件、消息推送、音视频处理等等

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

3.Celery的安装配置

pip install celery

消息中间件:RabbitMQ/Redis

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

4.Celery执行异步任务

基本使用

创建项目celerytest

创建py文件:celery_app_task.py

import celeryimport time# broker=‘redis://127.0.0.1:6379/2‘ 不加密码backend=‘redis://:[email protected]:6379/1‘broker=‘redis://:[email protected]:6379/2‘cel=celery.Celery(‘test‘,backend=backend,broker=broker)@cel.taskdef add(x,y):    return x+y??

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

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

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

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

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

创建py文件:result.py,查看任务执行结果

from celery.result import AsyncResultfrom 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连接和配置相关文件,必须叫这个名字    │   └── tasks1.py    #  所有任务函数    │   └── tasks2.py    #  所有任务函数    ├── check_result.py # 检查结果    └── send_task.py    # 触发任务

celery.py

from celery import Celery?cel = 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‘                      ])?# 时区cel.conf.timezone = ‘Asia/Shanghai‘# 是否使用UTCcel.conf.enable_utc = False

tasks1.py

import timefrom celery_task.celery import cel?@cel.taskdef test_celery(res):    time.sleep(5)    return "test_celery任务结果:%s"%res

tasks2.py

import timefrom celery_task.celery import cel@cel.taskdef test_celery2(res):    time.sleep(5)    return "test_celery2任务结果:%s"%res

check_result.py

from celery.result import AsyncResultfrom 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

from celery_task.tasks1 import test_celeryfrom celery_task.tasks2 import test_celery2?# 立即告知celery去执行test_celery任务,并传入一个参数result = test_celery.delay(‘第一个的执行‘)print(result.id)result = test_celery2.delay(‘第二个的执行‘)print(result.id)

添加任务(执行send_task.py),开启work:celery worker -A celery_task -l info -P eventlet,检查任务执行结果(执行check_result.py)

5.Celery执行定时任务

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

add_task.py

from celery_app_task import addfrom 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 timedeltatime_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)

类似于contab的定时任务

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

from datetime import timedeltafrom celery import Celeryfrom 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

6.Django中使用Celery

安装包

celery==3.1.25django-celery==3.1.20

在项目目录下创建celeryconfig.py

import djcelerydjcelery.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@taskdef 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,HttpResponsefrom app01.tasks import addfrom datetime import datetimedef 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 celeryconfigBROKER_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/zhangdajin/p/11128009.html

时间: 2024-10-28 19:24:44

Celery分布式任务队列的相关文章

Celery分布式任务队列快速入门

Celery介绍 Celery是基于Python开发的分布式任务队列.它支持使用任务队列的方式在分布的机器/进程/线程上执行任务调度. 如果业务场景中需要用到异步任务,就可以考虑使用celery, 举几个实例场景中可用的例子: 你想对100台机器执行一条批量命令,可能会花很长时间 ,但你不想让你的程序等着结果返回,而是给你返回 一个任务ID,你过一段时间只需要拿着这个任务id就可以拿到任务执行结果, 在任务执行ing进行时,你可以继续做其它的事情. 你想做一个定时任务,比如每天检测一下你们所有客

Celery分布式任务队列的认识和基本操作

一.简单认识 Celery是由Python开发.简单.灵活.可靠的分布式任务队列,其本质是生产者消费者模型,生产者发送任务到消息队列,消费者负责处理任务.Celery侧重于实时操作,但对调度支持也很好,其每天可以处理数以百万计的任务. 它的特点有: 简单:熟悉了它的流程后,配置使用简单: 高可用:任务执行失败或执行过程中发生连接中断,Celery会自动重新执行任务: 快速:一个单进程的Celery每分钟可处理上百万个任务: 灵活:Celery的各个组件都可以被扩展及自定制: 应用场景举例: 1.

Celery 分布式任务队列快速入门

本节内容 Celery介绍和基本使用 启用多个workers Celery 定时任务 与django结合 通过django配置celery periodic task 一.Celery介绍和基本使用 Celery 是一个 基于python开发的分布式异步消息任务队列,通过它可以轻松的实现任务的异步处理, 如果你的业务场景中需要用到异步任务,就可以考虑使用celery, 举几个实例场景中可用的例子: 你想对100台机器执行一条批量命令,可能会花很长时间 ,但你不想让你的程序等着结果返回,而是给你返

Celery 分布式任务队列入门

一.Celery介绍和基本使用 Celery 是一个 基于python开发的分布式异步消息任务队列,通过它可以轻松的实现任务的异步处理, 如果你的业务场景中需要用到异步任务,就可以考虑使用celery, 举几个实例场景中可用的例子: 你想对100台机器执行一条批量命令,可能会花很长时间 ,但你不想让你的程序等着结果返回,而是给你返回 一个任务ID,你过一段时间只需要拿着这个任务id就可以拿到任务执行结果, 在任务执行ing进行时,你可以继续做其它的事情. 你想做一个定时任务,比如每天检测一下你们

分布式任务队列celery用法详解

celery基础介绍:这个图我们可以看出,celery基本结构也就是三部分1 第一部分 broker也就是中间件消息队列,作用就是用来接收应用的请求这一部分常见玩法可以是rabbitmq和redis等2 第二部分 worker 也就是工作队列 也就是celery本身的任务队列服务,一般情况下大型的生产应用我们会结合supervisor来管理这么多的worker3 第三部分 result 存储,就是把执行的结果,状态等信息进行存储,常规用法我们可以用rabbitmq redis,mysql,mon

Python开发【模块】:Celery 分布式异步消息任务队列

Celery 前言: Celery 是一个 基于python开发的分布式异步消息任务队列,通过它可以轻松的实现任务的异步处理, 如果你的业务场景中需要用到异步任务,就可以考虑使用celery, 举几个实例场景中可用的例子: 你想对100台机器执行一条批量命令,可能会花很长时间 ,但你不想让你的程序等着结果返回,而是给你返回 一个任务ID,你过一段时间只需要拿着这个任务id就可以拿到任务执行结果, 在任务执行ing进行时,你可以继续做其它的事情. 你想做一个定时任务,比如每天检测一下你们所有客户的

分布式任务队列:Celery

一.Celery 安装与应用 关于 Celery 的内容,先依次看如下两篇文章,让你有个了解: http://blog.csdn.net/happyanger6/article/details/51404837 https://www.liaoxuefeng.com/article/00137760323922531a8582c08814fb09e9930cede45e3cc000 1.安装 Celery 和 Redis pip install celery # 这里我的 python 版本是

Celery异步任务队列/周期任务+ RabbitMQ + Django

一.Celery介绍和基本使用  Celery 是一个 基于python开发的分布式异步消息任务队列,通过它可以轻松的实现任务的异步处理, 如果你的业务场景中需要用到异步任务,就可以考虑使用celery, 举几个实例场景中可用的例子: 你想对100台机器执行一条批量命令,可能会花很长时间 ,但你不想让你的程序等着结果返回,而是给你返回 一个任务ID,你过一段时间只需要拿着这个任务id就可以拿到任务执行结果, 在任务执行ing进行时,你可以继续做其它的事情. 你想做一个定时任务,比如每天检测一下你

Django使用Celery异步任务队列

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