cerely 的使用
Posted changwenjun-666
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了cerely 的使用相关的知识,希望对你有一定的参考价值。
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. 上述问题可以解决,在使用的时候添加一些代码即可
使用场景: 异步任务:将耗时操作任务提交给Celery去异步执行,比如发送短信/邮件、消息推送、音视频处理、页面的静态化等等 定时任务:定时执行某件事情,比如每天数据统计
3.Celery的安装配置 pip install celery 消息中间件:RabbitMQ/Redis app=Celery(‘任务名‘,backend=‘xxx‘,broker=‘xxx‘)
4.Celery执行异步任务 创建项目celery_demo: 创建py文件:celery_task.py import celery import time # broker=‘redis://127.0.0.1:6379/2‘ 不加密码 backend=‘redis://:12345@127.0.0.1:6379/1‘ broker=‘redis://:12345@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_task import add result = add.delay(4,5) print(result.id) # 此处的id会被后面的result.py利用到 创建py文件:run.py,执行任务,或者使用命令执行:celery worker -A celery_task -l info 注:windows下:celery worker -A celery_app_task -l info -P eventlet from celery_task import cel if __name__ == ‘__main__‘: cel.worker_main() # cel.worker_main(argv=[‘--loglevel=info‘]) 创建py文件:result.py,查看任务执行结果 from celery.result import AsyncResult from celery_app_task import cel async = AsyncResult(id="e919d97d-2938-4d0f-9265-fd8237dc2aa3", app=cel) #此处的id为任务生成时产生的! 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(‘任务已经开始被执行‘) 上述4个文件存在一个 先后的执行顺序: 执行 add_task.py,添加任务,并获取任务ID 执行 run.py ,或者执行命令:celery worker -A celery_app_task -l info -P eventlet 执行 result.py,检查任务状态并获取结果
5、多任务结构 pro_cel ├── celery_task# celery相关文件夹 │ ├── celery.py # celery连接和配置相关文件,必须叫这个名字 │ └── tasks1.py # 所有任务函数 │ └── tasks2.py # 所有任务函数 ├── task_result.py # 检查结果 └── submit_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‘ # 是否使用UTC cel.conf.enable_utc = False task1.py import time from celery_task.celery import cel @cel.task def test_celery(res): time.sleep(5) return "test_celery任务结果:%s"%res task2.py import time from celery_task.celery import cel @cel.task def test_celery2(res): time.sleep(5) return "test_celery2任务结果:%s"%res task_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(‘任务已经开始被执行‘) submit_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) 即: 先实例化celery对象 (celery.py) 再提交任务 (submit_task.py) 开启worker,可写脚本实现,也可在终端输入celery worker -A celery_task -l info -P eventlet 去 实现 检查任务结果 (task_result.py)
6、Celery执行定时任务 celery_app_task.py import time from celery import Celery broker = ‘redis://:12345@127.0.0.1:6379/1‘ backend = ‘redis://:12345@127.0.0.1:6379/2‘ app = Celery(‘test2‘,broker=broker,backend=backend) @app.task def adds(x, y): time.sleep(1) return x + y 设定时间让celery执行 submit_task.py from celery_app_task import adds 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) # apply_async是定时任务的方法,eta指的是utc时间,args是参数 # 方式二 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 # 即当前时间10秒后执行任务 # 使用apply_async并设定时间 result = adds.apply_async(args=[4, 3], eta=task_time) print(result.id)
7、定时任务高级版 contab 计划任务 多任务结构: time_clcok ├── celery_task# celery相关文件夹 │ ├── celery.py # celery连接和配置相关文件,必须叫这个名字 │ └── tasks1.py # 所有任务函数 │ └── tasks2.py # 所有任务函数 ├── check_result.py # 检查结果 └── 不需要触发函数 celery.py broker = ‘redis://:12345@127.0.0.1:6379/2‘ backend = ‘redis://:12345@127.0.0.1:6379/3‘ apps = Celery(‘test‘, broker=broker, backend=backend, include=[ ‘time_clock.task1‘, ‘time_clock.task2‘, ]) # include 下的 数据 是任务的文件名 # 配置文件 apps.conf.timezone = ‘Asia/Shanghai‘ apps.conf.enable_utc = False apps.conf.beat_schedule = ‘t1-every-10-seconds‘: # 执行task1下的t1函数 ‘task‘: ‘time_clock.task1.t1‘, ‘schedule‘: timedelta(seconds=30), ‘args‘: (1, 2) , ‘t1-every-12-seconds‘: ‘task‘:‘time_clock.task2.t2‘, # 每年的具体时间执行任务 ‘schedule‘:crontab(minute=5,hour=11,day_of_month=29,month_of_year=8), ‘args‘:(22,66) , task1.py from time_clock.celery import apps import time @apps.task # apps.task 不要忘记添加 def t1(x, y): time.sleep(1) return x+y task2.py from time_clock.celery import apps import time @apps.task def t2(x,y): time.sleep(1) return x+y check_result.py .......略 总结:缺少了触发文件,但是在celery.py里添加了conf.beat_schedule方法; 因此需要手动在终端执行一个命令:启动一个beat celery beat -A time_clock -l info 除了开启beat命令以后,还得在开启一个worker celery worker -A time_clock -l info -P eventlet
8、celery在django里的使用
一般而言,celery在django里配置的话,对版本要求很严,
因此一般是把脚本直接 拉入django根目录下去执行即可,
需要注意的是:在这个脚本中是不能直接使用orm操作的,是需要导入一些模块才可。
import os
import django
django.setup()
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "celery_test.settings")
在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‘
以上是关于cerely 的使用的主要内容,如果未能解决你的问题,请参考以下文章
ceres使用记录之autodiff_cost_function.h
如何使用 Ceres Solver 库拟合二阶多元多项式(3D 表面)