什么是celery
Celery是一个简单、灵活且可靠的，处理大量消息的分布式系统
专注于实时处理的异步任务队列
同时也支持任务调度
celery架构
celery的架构由三部分组成，消息中间件（message broker），任务执行单元（worker）和任务执行结果存储（task result store）组成。
Celery本身不提供消息服务，但是可以方便的和第三方提供的消息中间件集成。包括，RabbitMQ, Redis等等
任务执行单元
任务结果存储
Task result store用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括AMQP, redis等
celery能做什么
异步任务
定时任务
周期任务
使用场景
耗时操作（所有的耗时操作都可以）将耗时操作任务提交给Celery去异步执行，比如发送短信/邮件、消息推送、音视频处理等等
定时任务（每天定时推送微信服务号消息，）定时执行某件事情，比如每天数据统计
使用
pip install celery
celery执行异步任务：
项目结构
celery的简单实例
celery_task_s1.py代码
from celery import Celery #不加密码 broker = 'redis://127.0.0.1:6379/0' #指定broker backend = 'redis://127.0.0.1:6379/1' #指定存储结果 # 加密码 # broker = 'redis://:123456@127.0.0.1:6379/2' # backend = 'redis://:123456@127.0.0.1:6379/1' # 一定要指定一个名字 app = Celery('test',broker=broker,backend=backend) #任务其实就是个函数 # 需要用一个装饰器装饰，表示该任务是被celery管理的，并且可以用celery执行的 @app.task def add(x,y) -> int: import time time.sleep(2) return x+y
add_task.py
#用于提交任务的py文件 import celery_task_s1 #正常同步执行任务 会睡两秒在执行 # ret=celery_task_s1.add(3,4) # print(ret) #提交任务到消息队列中 #只是把任务提交到消息队列中，并没有执行 # ret=celery_task_s1.add.delay(3,4) # print(ret) #a5ea035f-0cc3-44ba-b334-f5d7c7ce681d :任务的id号 #任务提交完成后，需要启动worker，可以用命令启动： # celery worker -A celery_task_s1 -l info #windows上：celery worker -A celery_task_s1 -l info -P eventlet win上首次使用worker需要安装eventlet模块
celery_result.py
from celery_task_s1 import app from celery.result import AsyncResult async = AsyncResult(id='30314784-04bd-4ec5-acab-5cb22181cfcb',app=app) if async.successful(): #取出它return的值 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('任务已经开始被执行')
写一个py文件：celery_task
- 指定brokey(消息中间件），指定backend(结果存储）
- 实例化产生一个Celery对象，app=Celery('起一个名字'，broker,backend)
- 加装饰器绑定任务，在函数（add）上加装饰器app.task
- 其他程序提交任务，先导入add, add.delay(参数，参数），会将该函数提交到消息中间件，但是并不会执行，有个返回值，直接print会打印出任务的id,以后用id去查询任务是否执行完成
- 启动worker去执行任务(worker可以先启动)：
- celery worker -A celery_task_s1 -l info (-l是艾欧）
- Windows下：celery worker -A celery_task_s1 -l info -P eventlet (需要安装eventlet模块)
- 还有一种作为自执行文件启动，可以写在往里面放值的文件下面这样运行放值的模块的时候worker也启动了 但是win上好像不支持
-
from celery_app_task import cel if __name__ == '__main__': cel.worker_main() # cel.worker_main(argv=['--loglevel=info')
- 查看结果：根据id去查询
async = AsyncResult(id="a5ea035f-0cc3-44ba-b334-f5d7c7ce681d", app=app) if async.successful(): #取出它return的值 result = async.get() print(result)
多任务结构
pro_cel ├── celery_task# celery相关文件夹 │ ├── celery.py # celery连接和配置相关文件,必须叫这个名字 │ └── order_task.py # 所有任务函数 │ └── user_task.py # 所有任务函数 ├── celery_result.py # 检查结果 └── add_task.py # 触发任务
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', ] ) #时区 # app.conf.timezone = 'Asia/Shanghai' #是否使用UTC # app.conf.enable_utc = False
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
add_task.py
from celery_task.order_task import order_add from celery_task.user_task import user_add # ret=order_add.delay(5,6) ret=user_add.delay(10,60) print(ret)
celery_result.py
from celery.result import AsyncResult from celery_task.celery import app from add_task import ret async = AsyncResult(id='c951c4f7-dcfc-46fc-b60e-ee67b61211ec',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('任务已经开始被执行')
启动worker: celery_task是包的名字（文件夹名字）
celery worker -A celery_task -l info -P eventlet #放值的时候手动启动add_task文件
celery执行定时任务
设定时间让celery执行一个任务
add_task.py 就是上面的执行异步任务的那个文件(往里面放值的时候使用定时方式）
#用于提交任务的py文件 import celery_task_s1 #执行定时任务：3s钟以后执行add任务 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) # print(result.id) #这里需要注意的是result是个对象，你可以直接打印因为内部写了__str__ #第二种获取时间的方法 ctime = datetime.now() # 默认用utc时间 utc_ctime = datetime.utcfromtimestamp(ctime.timestamp()) from datetime import timedelta #取10s之后的时间对象 time_delay = timedelta(seconds=3) task_time = utc_ctime + time_delay result = celery_task_s1.add.apply_async(args=[4, 3], eta=task_time) print(result.id) #这里需要注意的是result是个对象，你可以直接打印因为内部写了__str__ #任务提交完成后，需要启动worker，可以用命令启动： # celery worker -A celery_task_s1 -l info #windows上：celery worker -A celery_task_s1 -l info -P eventlet
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', ] ) #时区问题 每天的什么时候去执行任务 #时区 # app.conf.timezone = 'Asia/Shanghai' #是否使用UTC # app.conf.enable_utc = False #beat_schedule 定义定时任务的 app.conf.beat_schedule = { #名字随意命令 'add-every-2-seconds':{ #执行tasks1下的test_celery函数 'task':'celery_task.order_task.order_add', #每隔2秒执行一次 # 'schedule':1.0, 几秒钟几分钟几小时后schedule对应的参数 # 'schedule':crontab(minute='*/1'), 'schedule':timedelta(seconds=10), #传递参数 'args':(5,6) }, # 'add-every-12-seconds':{ # 'task':'celery_task.order_task.user_add', # #每年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) # } }
创建worker的方式并没有发生变化，但是这里要注意的是，每间隔一定时间后需要生产出来任务给worker去执行，这里需要一个生产者beat
启动一个beat:（这个负责定时周期往里面放值）
celery beat -A celery_task -l info
启动worker执行（winds）
celery worker -A celery_task -l info -P eventlet
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
第一种 直接把多任务结构文件直接拷贝在项目目录下
注意在celery的任务函数中不能直接调用django的环境，需要手动添加
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "untitled15.settings") import django django.setup()
第二种 在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 #from 这个django项目 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/znicy/p/5626040.html