Celery知识点总结（celery详解）

Celery知识点总结（celery详解）

本文内将主要介绍Celery的相关知识点，其中会涉及到其架构原理、重要功能讲解、相关配置以及使用技巧等。本文适合的阅读人群为在使用编写项目的主语言为Python且需要快速实现异步架构的开发者。笔者也会将自己的理解在文中进行阐述，这也算是在和大家交流心得的一个过程。若文中有错误的理解和概念，请大家及时纠正；吸纳大家的建议，对于我来说也是很重要的学习过程之一。

1. 架构

Celery是基于Python开发的一个分布式任务队列框架，支持使用任务队列的方式在分布的机器/进程/线程上执行任务调度。

其中，Celery的任务队列支持使用多种介质来实现，例如Redis, RabbitMQ和Amazon SQS等。通过使用其自带的任务调度器还可快速实现定时任务推送的功能。

Tips: Celery本质是一个任务队列框架，而非消息队列。虽然Celery和消息队列相同，都能够为项目提供异步处理数据的功能；但两者本质上适用的使用场景是不同的。消息队列除了能够提供异步处理数据的功能外，其还可以做到削峰填谷以及工程解耦。而Celery主要还是为项目提供异步处理任务的功能和易于扩展的分布式任务处理框架。笔者提醒大家一定要注意这一点认知，以便能够更好的、正确的使用Celery。

2. 任务路由

Celery中的任务路由是由Exchange和Queue这两个概念来实现的。

Celery可以在一个Broker上面开辟多个Queue，每个Queue绑定指定类型的任务；对应的Worker通过指定的队列获取任务；Exchange用来决定哪些任务发送到哪些Queue中。Celery结合了上述的这些功能从而实现了任务路由。本章节将会对于这两个概念展开介绍。

Tips: 这里需要注意一点的是，Celery中传递的是任务(Task)，而非消息。任务(Task)中会描述该任务所使用的方法(代码)，方法的入参等等。即任务发布方在发布任务之前，是需要清楚Task的调用细节的(至少要知道方法名以及入参规则)。这区别于消息队列的一点，消息队列中传递的是消息(可以理解为传送的方法的入参，而并没有指定处理的逻辑)。、

2.1 Exchange

发布的任务首先会经过Exchange。Exchange会根据设定条件将任务路由到不同的Queue中。不同类型的Exchange有着不同的路由规则。Exchange的类型有如下几类：

Direct Exchange直接交换，即指定一个任务被哪个队列接收。哪一个Queue绑定了任务中带有的Routing key,那么Exchange就会将任务直接转发给这个Queue。 Topic ExchangeTopic Exchange会使用通配符来匹配Routing key。Routing key的格式需要为使用“.”(点)符号来分割，通配符支持使用"#"(用于匹配0或1个字符)和"*"(用于匹配多个字符)。每一个Queue都会绑定一个通配公式，当任务中带有的Routing key满足通配公式时，则Exchange会将任务发送到该Queue中。这种类型的Exchange可以满足按需路由任务的需求。

在一个架构中，可以包含多个Exchange，每一个Exchange管理着不同的Queue与Worker(即真正执行Task逻辑的对象)。

2.2 Queue

当确定好Exchange的类型后，还需要创建所需的Queue，并将Queue绑定到相应的Exchange上。因此，在定义Queue的同时，还需要定义Queue与Exchange的绑定关系。定义Queue的配置如下所示:

CELERY_QUEUES = { Queue('queue01',Exchange('ex01',type='topic'), routing_key='*.task.*'), Queue('queue02', Exchange('ex01',type='topic'), routing_key='*.*.email'), Queue('queue03', Exchange('ex01',type='topic'), routing_key='*.add') }

其中，Exchange('ex01',type='topic')表示定义了一个名为ex01的Exchange，其类型为Topic Exchange。其次，Queue('queue01',Exchange('ex01',type='topic'), routing_key='.task.')表示定义了一个名叫queue01的队列，该列队绑定在了Exchange ex01上，并且只有当Routing key满足通配公式.task.时，ex01才会将任务转发到queue01中。

Tips: Exchange的定义是包含在Queue的定义当中的。

为了系统可靠性，还可以配置默认的Queue，Exchange以及Routing key；以保证任务必定会被处理。其相关的配置项如下:

default_exchange = Exchange('default', type='topic') app.conf.task_default_queue = 'Lost' app.conf.task_default_exchange = 'default' app.conf.task_default_routing_key = 'lost'

上述给出的示例，使用的是编码形式的配置方式。Celery同时还支持使用配置文件的方式进行配置，这里笔者也给出该种方式的Exchange与Queue的定义细节：

# Queues and Route Config task_default_exchange = "tsmp_topic_exchange" task_default_exchange_type = "topic" task_default_queue = "Lost" task_default_routing_key = "lost" task_queues = { "Alert": { "exchange": "tsmp_topic_exchange", "exchange_type": "topic", "routing_key": "alert" }, "Email": { "exchange": "tsmp_topic_exchange", "exchange_type": "topic", "routing_key": "email" } }

3. 任务

在Celery中，由于传递的是任务(Task)；因此构造Task便成为了Celery的另一大重点。本章节中会介绍Task的一些相关概念以及编写技巧。

3.1 任务注册

当Task方法实现后，需要使用app.task()装饰器来装饰Task方法。app.task()装饰器负责在应用任务注册表中注册你的任务。当任务被发送时，没有实际函数的代码被发送，只是将要执行任务对应的方法名称以及参数发送到Broker。当Worker收到消息时，它会在任务注册表中查找任务名称并使用指定参数执行相关方法。因此，任务发布者和Worker都应该有完整的Task源码。如果代码需要更新，则两者的源码都应同步更新。

Tips: 如果使用celerybeat来调度Task,则celerybeat中保存的Task信息一定要使用task的全路径名称，即包括包名。通常可根据已经注册上的任务名来模仿填写。 当Worker启动后，可以使用命令：celery -A inspect registered来查看当前已经注册了的任务。

3.2 自定义Task类

在Celery内部，是将Task看作为一个对象。即Celery实际上对Task这一概念进行了抽象。当多数任务都有着类似的内部逻辑时，或任务中逻辑过于复杂时；此时可以考虑将Task的内容封装到一个Task类中。之后，将自定义Task类作为app.task()装饰器中base的参数值后，就可以在Task的相应方法中调用自定义Task类中的方法了。例如:

from celery.utils.log import get_task_logger from cdm_celery import celery_config from cdm_celery.celery import app from cdm_celery.task.base.dns import CloudDomainServiceConsumer from factory.route53 import Route53Factory logger = get_task_logger(__name__) @app.task(base=CloudDomainServiceConsumer, bind=True, retry_backoff=True) def create_route53_record(self, type='', domain='', records=[]): factory = Route53Factory( access_key=celery_config.route53['access_key'], secret_access_key=celery_config.route53['secret_access_key'], domain=domain ) self.set_factory(factory) self.set_service_name('Route53') self.set_action_type('添加DNS记录') task_result = self.create_record(type, domain, records) # 调用自定义Task类方法 return task_result

除了将自身的业务需求封装在自定义Task类中，还可以重写一些celery.Task类本身的方法逻辑。例如通过重写
on_success、on_failure以及on_retry这3个方法可以实现Task的回调接口。例如：

from celery import Task class CloudDomainServiceConsumer(Task): def on_success(self, retval, task_id, args, kwargs): """任务执行成功回调方法 """ logger.info(log_msg_to_json_str( msg='执行成功', data={'args': args, 'kwargs': kwargs}, celery_task_id=task_id )) self.alert(data=retval) # 发送通知邮件 def on_failure(self, exc, task_id, args, kwargs, einfo): """任务执行失败回调方法 """ logger.error(log_msg_to_json_str( msg='执行失败;exc_type:{0};exc_details:{1};celery_info:{2}'.format( type(exc), exc, einfo, ), data={'args': args, 'kwargs': kwargs}, celery_task_id=task_id )) # 发送通知邮件 retval = { 'action_type': self.get_action_type(), 'service': self.get_service_name(), 'status': '添加失败', 'query_params': str(exc) } self.alert(data=retval) def on_retry(self, exc, task_id, args, kwargs, einfo): """任务重试回调方法 """ logger.warning(log_msg_to_json_str( msg='重试;exc_type:{0};exc_details:{1};celery_info:{2}'.format( type(exc), exc, einfo, ), data={'args': args, 'kwargs': kwargs}, celery_task_id=task_id ))

Tips: 自定义的Task类必须继承celery.Task类或其子类。

3.3 幂等性

Task的幂等性是需要开发者自己实现的，Celery是无法检测Task逻辑是否是幂等的。为了保证异步数据一致性，往往都需要我们将Task实现为支持幂等性的。

3.4 Task函数

3.5 日志

Celery使用的也是Python的标准logging模块；因此具体的配置与使用配置logging模块大致相同。不同只是在配置入口时，需要使用相应的装饰器来获取Celery的logger对象。这里介绍一些关于自定义Celery日志的方法。

关闭celery默认日志使用celery.signals.setup_logging。 import celery

@celery.signals.setup_logging.connectdef on_setup_logging(**kwargs):pass

2. 自定义celery自身日志(非task日志) 使用装饰器@celery.signals.after_setup_logger.connect装饰自定义日志格式方法。 3. 自定义Task日志 使用装饰器@celery.signals.after_setup_task_logger.connect装饰自定义日志格式方法。 ```python import celery @celery.signals.after_setup_task_logger.connect def on_after_setup_task_logger(logger, *args, **kwargs): logger.setLevel(logging.INFO) set_logger_file_handler(os.path.join(BASE_DIR, 'log/celery/task/info.log'), custom_logger=logger, level=logging.INFO) set_logger_file_handler(os.path.join(BASE_DIR, 'log/celery/task/error.log'), custom_logger=logger, level=logging.ERROR) set_logger_file_handler(os.path.join(BASE_DIR, 'log/celery/task/warning.log'), custom_logger=logger, level=logging.WARNING)

对于Celery日志的自定义，笔者有一些实践经验想分享给大家。由于Celery是一个异步任务框架，因此正常情况下会有多个Worker来并行处理任务。由于Python的logging模块对于多进程打印日志支持的不好，因此如果多个Worker往同一批日志文件中输出日志(仅针对单台服务器上启动多个Worker的情况)，则会导致日志输出混乱。此时好的做法是为每一个Worker建立起独立的日志文件，例如可以使用Worker名称或进程号来命名等。命名时可以使用统一的命名前缀或后缀，这样也方便与后期的日志采集工作。

3.6 任务重试

任务重试有两种定义方式：

手动定义需要自行在Task的逻辑中使用try except来抓取错误，并调用Task的retry()方法来进行任务重试。同时，可以为task设置一个最大重试次数max_retries;并且还可通过default_retry_delay来定义重试的时间间隔(延迟时间的单位是秒)。注意: retry()方法返回的是一个错误,因为需要向上层抛出(raise)出去。 自动重试使用autoretry_for参数即可，表示当出现任何错误的时候都将重试。还可以通过Task.retry_backoff来设置延迟重试，同时还可以设置延迟因子Task.retry_jitter来控制延迟的间隔。Task.max_retries用来设置重试的最大次数。例如： @app.task(autoretry_for=(Exception,)) def task(): pass

3.7 实时获取Task状态

对于获取Task状态，这里提供一个思路供大家参考： 1.将装饰器方法中bind参数定义为True。2.编写处理回调消息的逻辑方法3.启动任务4.调用任务的对象的get方法，其中on_message参数指定为自定义的消息回调处理方法;get方法的返回值即为当前任务的状态。可以配合while+sleep来不间断的判断任务的执行状态。

3.8 相关配置

在编写完相应的Task后，还需要配置如下一些配置才可以进行使用：

broker要指定代理者(任务队列)；用来承载需要处理的任务。 backend (仅Worker使用)Worker执行完相应Task后做产生的result将会保存在backend中。以便与后续的分析和任务状态确认。例如: # Result Backend Config BASE_DIR = Path(__file__).resolve().parent.parent result_backend = ''.join(['file://', os.path.join(BASE_DIR, 'cdm_celery/result')]) 任务注册在调用Task的时候需要使用app对象的task属性来调用相应的任务，而task属性中就是使用任务注册表中来查找哪些任务可以被调用。Worker接收到执行任务信息后，也先会使用app对象来查询要执行的task对应的源代码是哪个，查询的依据就是任务注册表。因此Task在使用前需要进行任务注册，即自定义Task的信息都需要填写在待注册列表中，以便Celery在启动的时候将这些Task都注册到任务注册表中。例如: imports = ( 'cdm_celery.task.route53.tasks', 'cdm_celery.task.email.tasks', 'cdm_celery.task.dnspod.tasks', 'cdm_celery.task.qywx.tasks' )

3.9 任务调用链

一般，我们都会将所需的业务逻辑或计算任务封装到Task中。当一个任务过于繁杂时，可能需要将其拆分为多个子任务。往往，这些子任务还存在一定的执行顺序。面对这个需求时，可以通过Task调用Task的方式，来将这些子任务串连起来，达到顺序执行的效果。这里笔者提供一个曾经实现过的思路，供大家学习和思考。通过结合上述的任务调用链和反射概念，可以实现动态配置任务调用链的功能。即在无需更改原代码的情况下可以快速在链中增加，修改或删除任务。

3.9.1 反射

通过反射操作，可以实现动态加载Task的功能，这为后续维护任务增加了很大的便利性。其实现的思路如下:

首先，在一些全局的配置中建立反射任务队列 alert_actions = [ 'cdm_celery.task.email.tasks.send_email', 'cdm_celery.task.qywx.tasks.qywx_alert' ] 其次，利用Python的相关模块实现反射逻辑 from importlib import import_module

def modules_extract(module_names=None):if not isinstance(module_names, Iterable):raise ValueError("需要提供包含了模块全路径名称字符串的可迭代对象")

for m in module_names: obj = CloudDomainServiceConsumer.module_extract(m) yield obj

### 3.9.2 Task调用Task 在相关的逻辑中加载反射任务队列和调用反射逻辑。同时，`通过使用Celery Task signature的概念来调用相关Task`即可。 ```python def alert(self, data=None): for alert_task in CloudDomainServiceConsumer.modules_extract(alert_actions): alert_task.signature( (), data, exchange=app.conf.task_queues['Alert']['exchange'], routing_key=app.conf.task_queues['Alert']['routing_key'] ).apply_async()

4. 配置

对于Celery的使用配置，笔者在这里想提醒大家注意一点: 4.0版本以后启用了新的小写的配置项名称。配置文件中不能同时出现老版本的CELERY开头大写和新的小写配置项名，建议统一使用新的小写配置项名。

5. 小彩蛋

这里之所以说是彩蛋，是因为这个章节所介绍的不是Celery的相关知识点。只是笔者在日常工作和学习中偶然碰到的与Celery相关的一个知识点，觉得很有意思，想给大家分享一下而已。我们经常会在一些业务或异步任务中添加重试机制的相关逻辑，业界使用最多的重试机制实现方式为指数退避算法。这里先对指数退避算法做一个简单的介绍:

在以太网中，该算法通常用于冲突后的调度重传。根据时隙和重传尝试次数来决定延迟重传。在 c 次碰撞后（比如请求失败），会选择 0 和 $2^c-1$ 之间的随机值作为时隙的数量。 对于第 1 次碰撞来说，每个发送者将会等待 0 或 1 个时隙进行发送。 而在第 2 次碰撞后，发送者将会等待 0 到 3（ 由 $2^2-1$ 计算得到）个时隙进行发送。 而在第 3 次碰撞后，发送者将会等待 0 到 7（ 由 $2^3-1$ 计算得到）个时隙进行发送。 以此类推…… 随着重传次数的增加，延迟的程度也会指数增长。说的通俗点，每次重试的时间间隔都是上一次的两倍。注意: 每次延长的是时间区间的最大值，并且每次的真正需要等待的时间是随机从时间区间中取出的。

恰巧，在Celery中有指数退避算法的相关实现(celery/utils/time.py中):

def get_exponential_backoff_interval( factor, retries, maximum, full_jitter=False ): """Calculate the exponential backoff wait time.""" # Will be zero if factor equals 0 countdown = factor * (2 ** retries) # Full jitter according to # https://awsarchitectureblog.com/2015/03/backoff.html if full_jitter: countdown = random.randrange(countdown + 1) # Adjust according to maximum wait time and account for negative values. return max(0, min(maximum, countdown))

大家可以模仿上述的代码来实现自需的指数退避算法。

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。