Celery 1
Celery是一个用Python开发的异步的分布式任务调度模块
Celery有以下优点:
- 简单:一但熟悉了celery的工作流程后,配置和使用还是比较简单的
- 高可用:当任务执行失败或执行过程中发生连接中断,celery 会自动尝试重新执行任务
- 快速:一个单进程的celery每分钟可处理上百万个任务
- 灵活: 几乎celery的各个组件都可以被扩展及自定制
应用:
创建tasks.py
from celery import Celery app = Celery('tasks', broker='amqp://guest@localhost//') @app.task def add(x, y): return x + y
Celery 的第一个参数是当前模块的名称,这个参数是必须的,这样的话名称可以自动生成。第二个参数是中间人关键字参数,指定你所使用的消息中间人的 URL,此处使用了 RabbitMQ,也是默认的选项
调用任务:
你可以用 delay()方法来调用任务
这是 apply_async() 方法的快捷方式,该方法允许你更好地控制任务执行
from tasks import add add.delay(4, 4)
保存结果:
如果你想要保持追踪任务的状态,Celery 需要在某个地方存储或发送这些状态。可以从内建的几个结果后端选择:SQLAlchemy/Django ORM、 Memcached 、 Redis 、 AMQP( RabbitMQ )或 MongoDB , 或者你可以自制。
下例中你将会使用 amqp 结果后端来发送状态消息。后端通过 Celery 的 backend 参数来指定。如果你选择使用配置模块,则通过 CELERY_RESULT_BACKEND 选项来设置:
app = Celery('tasks', backend='amqp', broker='amqp://')
配置:
Celery,如同家用电器一般,并不需要太多的操作。它有一个输入和一个输出, 你必须把输入连接到中间人上,如果想则把输出连接到结果后端上。但如果你仔细观察后盖,有一个盖子露出许多滑块、转盘和按钮:这就是配置。
默认配置对大多数使用案例已经足够好了,但有许多事情需要微调来让 Celery 如你所愿地工作。
配置可以直接在应用上设置,也可以使用一个独立的配置模块。
例如你可以通过修改 CELERY_TASK_SERIALIZER 选项来配置序列化任务载荷的默认的序列化方式:
app.conf.CELERY_TASK_SERIALIZER = 'json'
如果你一次性设置多个选项,你可以使用update:
app.conf.update( CELERY_TASK_SERIALIZER='json', CELERY_ACCEPT_CONTENT=['json'], # Ignore other content CELERY_RESULT_SERIALIZER='json', CELERY_TIMEZONE='Europe/Oslo', CELERY_ENABLE_UTC=True, )
对于大型项目,采用独立配置模块更为有效,事实上你会为硬编码周期任务间隔和任务路由选项感到沮丧,因为中心化保存配置更合适。尤其是对于库而言,这使得用户控制任务行为成为可能,你也可以想象系统管理员在遇到系统故障时对配置做出简单修改。
你可以调用 config_from_object() 来让 Celery 实例加载配置模块:
app.config_from_object('celeryconfig')
配置文件:celeryconfig.py
from kombu import Queue, Exchange BROKER_URL = 'redis://127.0.0.1:6379/7' CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/8' CELERY_IMPORTS = ( 'celery_app.task' ) CELERY_QUEUES = ( Queue('default', exchange=Exchange('default'), routing_key='default'), Queue('app_task1', exchange=Exchange('app_task1'), routing_key='app_task1'), Queue('app_task2', exchange=Exchange('app_task2'), routing_key='app_task2'), ) CELERY_ROUTES = { 'celery_app.task.task1': {'queue': 'app_task1', 'routing_key': 'app_task1'}, 'celery_app.task.task2': {'queue': 'app_task2', 'routing_key': 'app_task2'} }
定义routes用来决定不同的任务去哪一个queue
在启动worker时指定该worker执行哪一个queue中的任务
celery -A celery_app worker -l info -Q app_task1 -P eventlet celery -A celery_app worker -l info -Q app_task2 -P eventlet
RabbitMQ 是默认的中间人,所以除了需要你要使用的中间人实例的 URL 位置, 它并不需要任何额外的依赖或起始配置:
BROKER_URL = 'amqp://guest:guest@localhost:5672//'
如果想获取每个任务的执行结果,还需要配置一下把任务结果存在哪
app.conf.result_backend = 'redis://localhost:6379/0'
完整使用Celery流程
开始使用Celery啦
安装celery模块
pip install celery
创建一个celery application 用来定义你的任务列表
创建一个任务文件就叫tasks.py吧
from celery import Celery app = Celery('tasks', broker='redis://localhost', backend='redis://localhost') @app.task def add(x,y): print("running...",x,y) return x+y
启动Celery Worker来开始监听并执行任务
celery -A tasks worker --loglevel=info
调用任务
再打开一个终端, 进行命令行模式,调用任务
from tasks import add add.delay(4, 4)
看你的worker终端会显示收到 一个任务,此时你想看任务结果的话,需要在调用 任务时 赋值个变量
result = add.delay(4, 4)
结果
result.ready() # False or True
在项目中如何使用celery
可以把celery配置成一个应用
目录格式如下
proj/__init__.py /celery.py /tasks.py
proj/celery.py内容
from __future__ import absolute_import, unicode_literals from celery import Celery app = Celery('proj', broker='amqp://', backend='amqp://', include=['proj.tasks']) # Optional configuration, see the application user guide. app.conf.update( result_expires=3600, ) if __name__ == '__main__': app.start()
proj/tasks.py中的内容
from __future__ import absolute_import, unicode_literals from .celery import app @app.task def add(x, y): return x + y @app.task def mul(x, y): return x * y @app.task def xsum(numbers): return sum(numbers)
启动worker
celery -A proj worker -l info
Celery 定时任务
celery支持定时任务,设定好任务的执行时间,celery就会定时自动帮你执行, 这个定时任务模块叫celery beat
写一个脚本 叫periodic_task.py
from celery import Celery from celery.schedules import crontab app = Celery() @app.on_after_configure.connect def setup_periodic_tasks(sender, **kwargs): # Calls test('hello') every 10 seconds. sender.add_periodic_task(10.0, test.s('hello'), name='add every 10') # Calls test('world') every 30 seconds sender.add_periodic_task(30.0, test.s('world'), expires=10) # Executes every Monday morning at 7:30 a.m. sender.add_periodic_task( crontab(hour=7, minute=30, day_of_week=1), test.s('Happy Mondays!'), ) @app.task def test(arg): print(arg)
add_periodic_task 会添加一条定时任务
上面是通过调用函数添加定时任务,也可以像写配置文件 一样的形式添加, 下面是每30s执行的任务
app.conf.beat_schedule = { 'add-every-30-seconds': { 'task': 'tasks.add', 'schedule': 30.0, 'args': (16, 16) }, } app.conf.timezone = 'UTC'
任务添加好了,需要让celery单独启动一个进程来定时发起这些任务, 注意, 这里是发起任务,不是执行,这个进程只会不断的去检查你的任务计划, 每发现有任务需要执行了,就发起一个任务调用消息,交给celery worker去执行
启动任务调度器 celery beat
celery -A periodic_task beat
此时还差一步,就是还需要启动一个worker,负责执行celery beat发起的任务
启动celery worker来执行任务
celery -A periodic_task worker
此时观察worker的输出,是不是每隔一小会,就会执行一次定时任务呢!
celery -A periodic_task beat -s /home/celery/var/run/celerybeat-schedule
更复杂的定时配置
from celery.schedules import crontab app.conf.beat_schedule = { # Executes every Monday morning at 7:30 a.m. 'add-every-monday-morning': { 'task': 'tasks.add', 'schedule': crontab(hour=7, minute=30, day_of_week=1), 'args': (16, 16), }, }
最佳实践之与django结合
Django项目目录:
- proj/ - proj/__init__.py - proj/settings.py - proj/urls.py - manage.py
创建文件proj/proj/celery.py
from __future__ import absolute_import, unicode_literals import os from celery import Celery # set the default Django settings module for the 'celery' program. os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings') app = Celery('proj') # Using a string here means the worker don't have to serialize # the configuration object to child processes. # - namespace='CELERY' means all celery-related configuration keys # should have a `CELERY_` prefix. app.config_from_object('django.conf:settings', namespace='CELERY') # Load task modules from all registered Django app configs. app.autodiscover_tasks() @app.task(bind=True) def debug_task(self): print('Request: {0!r}'.format(self.request))
配置proj/proj/__init__.py:
from __future__ import absolute_import, unicode_literals # This will make sure the app is always imported when # Django starts so that shared_task will use this app. from .celery import app as celery_app __all__ = ['celery_app']
为celery设置环境变量
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')
创建Celery应用
app = Celery('proj')
配置settings
app.config_from_object('django.conf:settings', namespace='CELERY')
配置应用:
app.conf.update( # 配置broker, 这里我们用redis作为broker BROKER_URL='redis://:332572@127.0.0.1:6379/1', )
设置app自动加载任务:
app.autodiscover_tasks(settings.INSTALLED_APPS) # 从已经安装的app中查找任务
然后在具体的app里的tasks.py里写你的任务
# Create your tasks here from __future__ import absolute_import, unicode_literals from celery import shared_task @shared_task def add(x, y): return x + y @shared_task def mul(x, y): return x * y @shared_task def xsum(numbers): return sum(numbers)
在你的django views里调用celery task
from django.shortcuts import render,HttpResponse # Create your views here. from bernard import tasks def task_test(request): res = tasks.add.delay(228,24) print("start running task") print("async task res",res.get() ) return HttpResponse('res %s'%res.get())
注意,经测试,每添加或修改一个任务,celery beat都需要重启一次,要不然新的配置不会被celery beat进程读到
最新文章
- 通过反射获取DLL的类实现加载窗体
- HTTP协议用的TCP但是只建立单向连接
- [AJAX系列]XMLHttpRequest请求
- JDK的安装和配置
- 迅为iTOP-4412核心板调整电压
- java 导入包(误区)
- 图片放大镜插件 Cloud Zoom v3.1
- JVM启动过程——JVM之一
- 理解Java中的抽象
- jQuery.merge( first, second )返回: Array
- cobbler部署安装CentOS6.8
- 修改input属性placeholder的样式
- P1313 计算系数
- Maven私有仓库: 发布release版本报错:Return code is: 400, ReasonPhrase: Repository does not allow upd ating assets: maven-releases.
- 深入理解Linux内核-系统调用
- Nvidia显卡安装驱动
- 【BZOJ2331】[SCOI2011]地板 插头DP
- goim源码分析与二次开发-comet分析二
- mono for android之文件系统与应用程序首选项(转)
- 20155204 2016-2017-2 《Java程序设计》第9周学习总结