分布式任务队列Celery入门与进阶
一、简介
celery是由python开发、简单、灵活、可靠的分布式任务队列,其本质是生产者消费者模型,生产者发送任务到消息队列,消费者负责处理任务。celery侧重于实时操作,但对调度支持也很好,其每天可以处理数以百万计的任务。特点:
- 简单:熟悉celery的工作流程后,配置使用简单
- 高可用:当任务执行失败或执行过程中发生连接中断,celery会自动尝试重新执行任务
- 快速:一个单进程的celery每分钟可处理上百万个任务
- 灵活:几乎celery的各个组件都可以被扩展及自定制
应用场景举例:
1.web应用:当用户在网站进行某个操作需要很长时间完成时,我们可以将这种操作交给celery执行,直接返回给用户,等到celery执行完成以后通知用户,大大提好网站的并发以及用户的体验感。
2.任务场景:比如在运维场景下需要批量在几百台机器执行某些命令或者任务,此时celery可以轻松搞定。
3.定时任务:向定时导数据报表、定时发送通知类似场景,虽然linux的计划任务可以帮我实现,但是非常不利于管理,而celery可以提供管理接口和丰富的api。
二、架构&工作原理
celery由以下三部分构成:消息中间件(broker)、任务执行单元worker、结果存储(backend),如下图:
工作原理:
- 任务模块task包含异步任务和定时任务。其中,异步任务通常在业务逻辑中被触发并发往消息队列,而定时任务由celery beat进程周期性地将任务发往消息队列;
- 任务执行单元worker实时监视消息队列获取队列中的任务执行;
- woker执行完任务后将结果保存在backend中;
消息中间件broker
消息中间件broker官方提供了很多备选方案,支持rabbitmq、redis、amazon sqs、mongodb、memcached 等,官方推荐rabbitmq。
任务执行单元worker
worker是任务执行单元,负责从消息队列中取出任务执行,它可以启动一个或者多个,也可以启动在不同的机器节点,这就是其实现分布式的核心。
结果存储backend
backend结果存储官方也提供了诸多的存储方式支持:rabbitmq、 redis、memcached,sqlalchemy, django orm、apache cassandra、elasticsearch。
三、安装使用
这里我使用的redis作为消息中间件,redis安装可以参考https://www.cnblogs.com/wdliu/p/9360286.html。
celery安装:
pip3 install celery
简单使用
目录结构:
project/ ├── __init__.py ├── config.py └── tasks.py
各目录文件说明:
__init__.py:初始化celery以及加载配置文件
#!/usr/bin/env python3 # -*- coding:utf-8 -*- # author:wd from celery import celery app = celery('project') # 创建 celery 实例 app.config_from_object('project.config') # 加载配置模块
config.py: celery相关配置文件,更多配置参考:http://docs.celeryproject.org/en/latest/userguide/configuration.html
#!/usr/bin/env python3 # -*- coding:utf-8 -*- # author:wd broker_url = 'redis://10.1.210.69:6379/0' # broker配置,使用redis作为消息中间件 celery_result_backend = 'redis://10.1.210.69:6379/0' # backend配置,这里使用redis celery_result_serializer = 'json' # 结果序列化方案 celery_task_result_expires = 60 * 60 * 24 # 任务过期时间 celery_timezone='asia/shanghai' # 时区配置 celery_imports = ( # 指定导入的任务模块,可以指定多个 'project.tasks', )
tasks.py :任务定义文件
#!/usr/bin/env python3 # -*- coding:utf-8 -*- # author:wd from project import app @app.task def show_name(name): return name
启动worker:
celery worker -a project -l debug
各个参数含义:
worker: 代表第启动的角色是work当然还有beat等其他角色;
-a :项目路径,这里我的目录是project
-l:启动的日志级别,更多参数使用celery --help查看
查看日志输出,会发现我们定义的任务,以及相关配置:
虽然启动了worker,但是我们还需要通过delay或apply_async来将任务添加到worker中,这里我们通过交互式方法添加任务,并返回asyncresult对象,通过asyncresult对象获取结果:
asyncresult除了get方法用于常用获取结果方法外还提以下常用方法或属性:
- state: 返回任务状态;
- task_id: 返回任务id;
- result: 返回任务结果,同get()方法;
- ready(): 判断任务是否以及有结果,有结果为true,否则false;
- info(): 获取任务信息,默认为结果;
- wait(t): 等待t秒后获取结果,若任务执行完毕,则不等待直接获取结果,若任务在执行中,则wait期间一直阻塞,直到超时报错;
- successfu(): 判断任务是否成功,成功为true,否则为false;
四、进阶使用
对于普通的任务来说可能满足不了我们的任务需求,所以还需要了解一些进阶用法,celery提供了诸多调度方式,例如任务编排、根据任务状态执行不同的操作、重试机制等,以下会对常用高阶用法进行讲述。
定时任务&计划任务
celery的提供的定时任务主要靠schedules来完成,通过beat组件周期性将任务发送给woker执行。在示例中,新建文件period_task.py,并添加任务到配置文件中:
period_task.py:
#!/usr/bin/env python3 # -*- coding:utf-8 -*- # author:wd from project import app from celery.schedules import crontab @app.on_after_configure.connect def setup_periodic_tasks(sender, **kwargs): sender.add_periodic_task(10.0, add.s(1,3), name='1+3=') # 每10秒执行add sender.add_periodic_task( crontab(hour=16, minute=56, day_of_week=1), #每周一下午四点五十六执行sayhai sayhi.s('wd'),name='say_hi' ) @app.task def add(x,y): print(x+y) return x+y @app.task def sayhi(name): return 'hello %s' % name
config.py
#!/usr/bin/env python3 # -*- coding:utf-8 -*- # author:wd broker_url = 'redis://10.1.210.69:6379/0' # broker配置,使用redis作为消息中间件 celery_result_backend = 'redis://10.1.210.69:6379/0' # backend配置,这里使用redis celery_result_serializer = 'json' # 结果序列化方案 celery_task_result_expires = 60 * 60 * 24 # 任务过期时间 celery_timezone='asia/shanghai' # 时区配置 celery_imports = ( # 指定导入的任务模块,可以指定多个 'project.tasks', 'project.period_task', #定时任务 )
启动worker和beat:
celery worker -a project -l debug #启动work celery beat -a project.period_task -l debug #启动beat,注意此时对应的文件路径
我们可以观察worker日志:
还可以通过配置文件方式指定定时和计划任务,此时的配置文件如下:
#!/usr/bin/env python3 # -*- coding:utf-8 -*- # author:wd from project import app from celery.schedules import crontab broker_url = 'redis://10.1.210.69:6379/0' # broker配置,使用redis作为消息中间件 celery_result_backend = 'redis://10.1.210.69:6379/0' # backend配置,这里使用redis celery_result_serializer = 'json' # 结果序列化方案 celery_task_result_expires = 60 * 60 * 24 # 任务过期时间 celery_timezone='asia/shanghai' # 时区配置 celery_imports = ( # 指定导入的任务模块,可以指定多个 'project.tasks', 'project.period_task', ) app.conf.beat_schedule = { 'period_add_task': { # 计划任务 'task': 'project.period_task.add', #任务路径 'schedule': crontab(hour=18, minute=16, day_of_week=1), 'args': (3, 4), }, 'add-every-30-seconds': { # 每10秒执行 'task': 'project.period_task.sayhi', #任务路径 'schedule': 10.0, 'args': ('wd',) }, }
此时的period_task.py只需要注册到woker中就行了,如下:
#!/usr/bin/env python3 # -*- coding:utf-8 -*- # author:wd from project import app @app.task def add(x,y): print(x+y) return x+y @app.task def sayhi(name): return 'hello %s' % name
同样启动worker和beat结果和第一种方式一样。更多详细的内容请参考:http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html#crontab-schedules
任务绑定
celery可通过任务绑定到实例获取到任务的上下文,这样我们可以在任务运行时候获取到任务的状态,记录相关日志等。
修改任务中的period_task.py,如下:
#!/usr/bin/env python3 # -*- coding:utf-8 -*- # author:wd from project import app from celery.utils.log import get_task_logger
logger = get_task_logger(__name__) @app.task(bind=true) # 绑定任务 def add(self,x,y): logger.info(self.request.__dict__) #打印日志 try: a=[] a[10]==1 except exception as e: raise self.retry(exc=e, countdown=5, max_retries=3) # 出错每5秒尝试一次,总共尝试3次 return x+y
在以上代码中,通过bind参数将任务绑定,self指任务的上下文,通过self获取任务状态,同时在任务出错时进行任务重试,我们观察日志:
内置钩子函数
celery在执行任务时候,提供了钩子方法用于在任务执行完成时候进行对应的操作,在task源码中提供了很多状态钩子函数如:on_success(成功后执行)、on_failure(失败时候执行)、on_retry(任务重试时候执行)、after_return(任务返回时候执行),在进行使用是我们只需要重写这些方法,完成相应的操作即可。
在以下示例中,我们继续修改period_task.py,分别定义三个任务来演示任务失败、重试、任务成功后执行的操作:
#!/usr/bin/env python3 # -*- coding:utf-8 -*- # author:wd from project import app from celery.utils.log import get_task_logger from celery import task logger = get_task_logger(__name__) class demotask(task): def on_success(self, retval, task_id, args, kwargs): # 任务成功执行 logger.info('task id:{} , arg:{} , successful !'.format(task_id,args)) def on_failure(self, exc, task_id, args, kwargs, einfo): #任务失败执行 logger.info('task id:{} , arg:{} , failed ! erros : {}' .format(task_id,args,exc)) def on_retry(self, exc, task_id, args, kwargs, einfo): #任务重试执行 logger.info('task id:{} , arg:{} , retry ! einfo: {}'.format(task_id, args, exc)) @app.task(base=demotask,bind=true) def add(self,x,y): try: a=[] a[10]==1 except exception as e: raise self.retry(exc=e, countdown=5, max_retries=1) # 出错每5秒尝试一次,总共尝试1次 return x+y @app.task(base=demotask) def sayhi(name): a=[] a[10]==1 return 'hi {}'.format(name) @app.task(base=demotask) def sum(a,b): return 'a+b={} '.format(a+b)
此时的配置文件config.py:
#!/usr/bin/env python3 # -*- coding:utf-8 -*- # author:wd from project import app from celery.schedules import crontab broker_url = 'redis://10.1.210.69:6379/0' # broker配置,使用redis作为消息中间件 celery_result_backend = 'redis://10.1.210.69:6379/0' # backend配置,这里使用redis celery_result_serializer = 'json' # 结果序列化方案 celery_task_result_expires = 60 * 60 * 24 # 任务过期时间 celery_timezone='asia/shanghai' # 时区配置 celery_imports = ( # 指定导入的任务模块,可以指定多个 'project.tasks', 'project.period_task', ) app.conf.beat_schedule = { 'add': { # 每10秒执行 'task': 'project.period_task.add', #任务路径 'schedule': 10.0, 'args': (10,12), }, 'sayhi': { # 每10秒执行 'task': 'project.period_task.sayhi', #任务路径 'schedule': 10.0, 'args': ('wd',), }, 'sum': { # 每10秒执行 'task': 'project.period_task.sum', #任务路径 'schedule': 10.0, 'args': (1,3), }, }
然后重启worker和beat,查看日志:
任务编排
在很多情况下,一个任务需要由多个子任务或者一个任务需要很多步骤才能完成,celery同样也能实现这样的任务,完成这类型的任务通过以下模块完成:
-
group: 并行调度任务
-
chain: 链式任务调度
-
chord: 类似group,但分header和body2个部分,header可以是一个group任务,执行完成后调用body的任务
-
map: 映射调度,通过输入多个入参来多次调度同一个任务
-
starmap: 类似map,入参类似*args
-
chunks: 将任务按照一定数量进行分组
修改tasks.py:
#!/usr/bin/env python3 # -*- coding:utf-8 -*- # author:wd from project import app @app.task def add(x,y): return x+y @app.task def mul(x,y): return x*y @app.task def sum(data_list): res=0 for i in data_list: res+=i return res
group: 组任务,组内每个任务并行执行
和project同级目录新建consumer.py如下:
from celery import group from project.tasks import add,mul,sum res = group(add.s(1,2),add.s(1,2))() # 任务 [1+2,1+2] while true: if res.ready(): print('res:{}'.format(res.get())) break
结果:
chain:链式任务
链式任务中,默认上一个任务的返回结果作为参数传递给子任务
from celery import chain from project.tasks import add,mul,sum res = chain(add.s(1,2),add.s(3),mul.s(3))() # 任务((1+2)+3)*3 while true: if res.ready(): print('res:{}'.format(res.get())) break #结果 #res:18
还可以使用|表示链式任务,上面任务也可以表示为:
res = (add.s(1,2) | add.s(3) | (mul.s(3)))() res.get()
chord:任务分割,分为header和body两部分,hearder任务执行完在执行body,其中hearder返回结果作为参数传递给body
from celery import chord from project.tasks import add,mul,sum res = chord(header=[add.s(1,2),mul.s(3,4)],body=sum.s())() # 任务(1+2)+(3*4) while true: if res.ready(): print('res:{}'.format(res.get())) break #结果: #res:15
chunks:任务分组,按照任务的个数分组
from project.tasks import add,mul,sum res = add.chunks(zip(range(5),range(5)),4)() # 4 代表每组的任务的个数 while true: if res.ready(): print('res:{}'.format(res.get())) break
结果:
delay &apply_async
对于delay和apply_async都可以用来进行任务的调度,本质上是delay对apply_async进行了再一次封装(或者可以说是快捷方式),两者都返回asyncresult对象,以下是两个方法源码。
def delay(self, *args, **kwargs): """star argument version of :meth:`apply_async`. does not support the extra options enabled by :meth:`apply_async`. arguments: *args (any): positional arguments passed on to the task. **kwargs (any): keyword arguments passed on to the task. returns: celery.result.asyncresult: future promise. """ return self.apply_async(args, kwargs)
def apply_async(self, args=none, kwargs=none, task_id=none, producer=none, link=none, link_error=none, shadow=none, **options): """apply tasks asynchronously by sending a message. arguments: args (tuple): the positional arguments to pass on to the task. kwargs (dict): the keyword arguments to pass on to the task. countdown (float): number of seconds into the future that the task should execute. defaults to immediate execution. eta (~datetime.datetime): absolute time and date of when the task should be executed. may not be specified if `countdown` is also supplied. expires (float, ~datetime.datetime): datetime or seconds in the future for the task should expire. the task won't be executed after the expiration time. shadow (str): override task name used in logs/monitoring. default is retrieved from :meth:`shadow_name`. connection (kombu.connection): re-use existing broker connection instead of acquiring one from the connection pool. retry (bool): if enabled sending of the task message will be retried in the event of connection loss or failure. default is taken from the :setting:`task_publish_retry` setting. note that you need to handle the producer/connection manually for this to work. retry_policy (mapping): override the retry policy used. see the :setting:`task_publish_retry_policy` setting. queue (str, kombu.queue): the queue to route the task to. this must be a key present in :setting:`task_queues`, or :setting:`task_create_missing_queues` must be enabled. see :ref:`guide-routing` for more information. exchange (str, kombu.exchange): named custom exchange to send the task to. usually not used in combination with the ``queue`` argument. routing_key (str): custom routing key used to route the task to a worker server. if in combination with a ``queue`` argument only used to specify custom routing keys to topic exchanges. priority (int): the task priority, a number between 0 and 9. defaults to the :attr:`priority` attribute. serializer (str): serialization method to use. can be `pickle`, `json`, `yaml`, `msgpack` or any custom serialization method that's been registered with :mod:`kombu.serialization.registry`. defaults to the :attr:`serializer` attribute. compression (str): optional compression method to use. can be one of ``zlib``, ``bzip2``, or any custom compression methods registered with :func:`kombu.compression.register`. defaults to the :setting:`task_compression` setting. link (signature): a single, or a list of tasks signatures to apply if the task returns successfully. link_error (signature): a single, or a list of task signatures to apply if an error occurs while executing the task. producer (kombu.producer): custom producer to use when publishing the task. add_to_parent (bool): if set to true (default) and the task is applied while executing another task, then the result will be appended to the parent tasks ``request.children`` attribute. trailing can also be disabled by default using the :attr:`trail` attribute publisher (kombu.producer): deprecated alias to ``producer``. headers (dict): message headers to be included in the message. returns: celery.result.asyncresult: promise of future evaluation. raises: typeerror: if not enough arguments are passed, or too many arguments are passed. note that signature checks may be disabled by specifying ``@task(typing=false)``. kombu.exceptions.operationalerror: if a connection to the transport cannot be made, or if the connection is lost. note: also supports all keyword arguments supported by :meth:`kombu.producer.publish`. """ if self.typing: try: check_arguments = self.__header__ except attributeerror: # pragma: no cover pass else: check_arguments(*(args or ()), **(kwargs or {})) app = self._get_app() if app.conf.task_always_eager: with denied_join_result(): return self.apply(args, kwargs, task_id=task_id or uuid(), link=link, link_error=link_error, **options) if self.__v2_compat__: shadow = shadow or self.shadow_name(self(), args, kwargs, options) else: shadow = shadow or self.shadow_name(args, kwargs, options) preopts = self._get_exec_options() options = dict(preopts, **options) if options else preopts options.setdefault('ignore_result', self.ignore_result) return app.send_task( self.name, args, kwargs, task_id=task_id, producer=producer, link=link, link_error=link_error, result_cls=self.asyncresult, shadow=shadow, task_type=self, **options )
对于其使用,apply_async支持常用参数:
- eta:指定任务执行时间,类型为datetime时间类型;
- countdown:倒计时,单位秒,浮点类型;
- expires:任务过期时间,如果任务在超过过期时间还未执行则回收任务,浮点类型获取datetime类型;
- retry:任务执行失败时候是否尝试,布尔类型。;
- serializer:序列化方案,支持pickle、json、yaml、msgpack;
- priority:任务优先级,有0~9优先级可设置,int类型;
- retry_policy:任务重试机制,其中包含几个重试参数,类型是dict如下:
max_retries:最大重试次数 interval_start:重试等待时间 interval_step:每次重试叠加时长,假设第一重试等待1s,第二次等待1+n秒 interval_max:最大等待时间 ####示例 add.apply_async((1, 3), retry=true, retry_policy={ 'max_retries': 1, 'interval_start': 0, 'interval_step': 0.8, 'interval_max': 5, })
更多参数参考:http://docs.celeryproject.org/en/latest/reference/celery.app.task.html#celery.app.task.task.apply_async
五、管理与监控
celery管理和监控功能是通过flower组件实现的,flower组件不仅仅提供监控功能,还提供http api可实现对woker和task的管理。
安装使用
pip3 install flower
启动
flower -a project --port=5555 # -a :项目目录 #--port 指定端口
访问http:ip:5555
api使用,例如获取woker信息:
curl http://127.0.0.1:5555/api/workers
结果:
更多api参考:https://flower.readthedocs.io/en/latest/api.html