首页 > 编程知识 正文

celery任务管理,celery 封装

时间:2023-05-03 11:57:10 阅读:198019 作者:4778

目录 一、Celery 对象解析二、创建异步任务的方法 task三、调用异步任务的三种方法四、获取任务结果和状态五、Celery 使用案例

一、Celery 对象解析

我们先来看一下 Celery 的初始化方法:

class Celery(object): def __init__(self, main=None, loader=None, backend=None, amqp=None, events=None, log=None, control=None, set_as_current=True, accept_magic_kwargs=False, tasks=None, broker=None, include=None, changes=None, config_source=None, fixups=None, task_cls=None, autofinalize=True, **kwargs):

常用的需要配置的参数:
这些参数都是 celery 实例化的配置,我们也可以不写,可以使用config_from_object方法加载配置;

main : 如果作为__main__运行,则为主模块的名称。用作自动生成的任务名称的前缀loader : 当前加载器实例。backend : 任务结果url;amqp : AMQP对象或类名,一般不管;log : 日志对象或类名;set_as_current : 将本实例设为全局当前应用tasks : 任务注册表。broker : 使用的默认代理的URL,任务队列;include : 每个worker应该导入的模块列表,以实例创建的模块的目录作为起始路径;
二、创建异步任务的方法 task

任何被 task 修饰的方法都会被创建一个 Task 对象,变成一个可序列化并发送到远程服务器的任务;它有多种修饰方式:

方式一:使用默认的参数

@celery.taskdef function_name(): pass

方式二:指定相关参数

@celery.task(bind=True, name='name')def function_name(): pass# task方法参数name : 可以显式指定任务的名字;默认是模块的命名空间中本函数的名字。serializer : 指定本任务的序列化的方法;bind : 一个bool值,设置是否绑定一个task的实例,如果绑定,task实例会作为参数传递到任务方法中,可以访问task实例的所有的属性,即前面反序列化中那些属性base : 定义任务的基类,可以以此来定义回调函数,默认是Task类,我们也可以定义自己的Task类default_retry_delay : 设置该任务重试的延迟时间,当任务执行失败后,会自动重试,单位是秒,默认3分钟;autoretry_for : 设置在特定异常时重试任务,默认False即不重试;retry_backoff : 默认False,设置重试时的延迟时间间隔策略;retry_backoff_max : 设置最大延迟重试时间,默认10分钟,如果失败则不再重试;retry_jitter : 默认True,即引入抖动,避免重试任务集中执行;# 当bind=True时,add函数第一个参数是self,指的是task实例@task(bind=True) # 第一个参数是self,使用self.request访问相关的属性def add(self, x, y): try: logger.info(self.request.id) except: self.retry() # 当任务失败则进行重试,也可以通过max_retries属性来指定最大重试次数

方式三:自定义Task基类

import celeryclass MyTask(celery.Task): # 任务失败时执行 def on_failure(self, exc, task_id, args, kwargs, einfo): print('{0!r} failed: {1!r}'.format(task_id, exc)) # 任务成功时执行 def on_success(self, retval, task_id, args, kwargs): pass # 任务重试时执行 def on_retry(self, exc, task_id, args, kwargs, einfo): pass@task(base=MyTask)def add(x, y): raise KeyError()# 方法相关的参数exc : 失败时的错误的类型;task_id : 任务的id;args : 任务函数的参数;kwargs : 键值对参数;einfo : 失败或重试时的异常详细信息;retval : 任务成功执行的返回值;

Task的常用属性

Task.name : 任务名称;Task.request : 当前任务的信息;Task.max_retries : 设置重试的最大次数Task.throws : 预期错误类的可选元组,不应被视为实际错误,而是结果失败;Task.rate_limit : 设置此任务类型的速率限制Task.time_limit : 此任务的硬限时(以秒为单位)。Task.ignore_result : 不存储任务状态。默认False;Task.store_errors_even_if_ignored : 如果True,即使任务配置为忽略结果,也会存储错误。Task.serializer : 标识要使用的默认序列化方法的字符串。Task.compression : 标识要使用的默认压缩方案的字符串。默认为task_compression设置。Task.backend : 指定该任务的结果存储后端用于此任务。Task.acks_late : 如果设置True为此任务的消息将在任务执行后确认 ,而不是在执行任务之前(默认行为),即默认任务执行之前就会发送确认;Task.track_started : 如果True任务在工作人员执行任务时将其状态报告为“已启动”。默认是False;
三、调用异步任务的三种方法

调用异步任务的三个方法分别是:

# 方法一:这是apply_async方法的别名,但接受的参数较为简单;task.delay()# 方法二:可以接受复杂的参数task.apply_async(args=[arg1, arg2], kwargs={key:value, key:value})# 方法三:可以发送未被注册的异步任务,即没有被celery.task装饰的任务;send_task()

方法一:app.send_task
注意: send_task 在发送的时候是不会检查 tasks.add 函数是否存在的,即使为空也会发送成功,所以 celery 执行是可能找不到该函数报错;

# File_name:tasks.pyfrom celery import Celeryapp = Celery()def add(x, y): return x+yapp.send_task('tasks.add',args=[3,4]) # 参数基本和apply_async函数一样

方法二:Task.delay
delay 方法是 apply_async 方法的简化版,不支持执行选项,只能传递任务的参数。

from celery import Celeryapp = Celery()@app.taskdef add(x, y, z=0): return x + yadd.delay(30, 40, z=5)# 包括位置参数和关键字参数

方法三:Task.apply_async
apply_async 支持执行选项,它会覆盖全局的默认参数和定义该任务时指定的执行选项,本质上还是调用了 send_task 方法;

from celery import Celeryapp = Celery()@app.taskdef add(x, y, z=0): return x + y add.apply_async(args=[30,40], kwargs={'z':5})# 其他参数task_id : 为任务分配唯一id,默认是uuid;countdown : 设置该任务等待一段时间再执行,单位为s;eta : 定义任务的开始时间;eta=time.time()+10;expires : 设置任务时间,任务在过期时间后还没有执行则被丢弃;retry : 如果任务失败后, 是否重试;使用true或false,默认为trueshadow : 重新指定任务的名字str,覆盖其在日志中使用的任务名称;retry_policy : {},重试策略.如下:----max_retries : 最大重试次数, 默认为 3 次.----interval_start : 重试等待的时间间隔秒数, 默认为 0 , 表示直接重试不等待.----interval_step : 每次重试让重试间隔增加的秒数, 可以是数字或浮点数, 默认为 0.2----interval_max : 重试间隔最大的秒数, 即 通过 interval_step 增大到多少秒之后, 就不在增加了, 可以是数字或者浮点数, 默认为 0.2 .routing_key : 自定义路由键;queue : 指定发送到哪个队列;exchange : 指定发送到哪个交换机;priority : 任务队列的优先级,0到255之间,对于rabbitmq来说0是最高优先级;serializer :任务序列化方法;通常不设置;compression : 压缩方案,通常有zlib, bzip2headers : 为任务添加额外的消息;link : 任务成功执行后的回调方法;是一个signature对象;可以用作关联任务;link_error : 任务失败后的回调方法,是一个signature对象;# 其他参数参考用法如下:add.apply_async((2, 2), retry=True, retry_policy={ 'max_retries': 3, 'interval_start': 0, 'interval_step': 0.2, 'interval_max': 0.2,})

自定义发布者、交换机、路由键、队列、优先级、序列方案和压缩方法:

task.apply_async((2,2), compression='zlib', serialize='json', queue='priority.high', routing_key='web.add', priority=0, exchange='web_exchange')
四、获取任务结果和状态

由于 celery 发送的都是去其他进程执行的任务,如果需要在客户端监控任务的状态,有如下方法:

r = task.apply_async()r.ready() # 查看任务状态,返回布尔值, 任务执行完成, 返回 True, 否则返回 False.r.wait() # 会阻塞等待任务完成, 返回任务执行结果,很少使用;r.get(timeout=1) # 获取任务执行结果,可以设置等待时间,如果超时但任务未完成返回None;r.result # 任务执行结果,未完成返回None;r.state # PENDING, START, SUCCESS,任务当前的状态r.status # PENDING, START, SUCCESS,任务当前的状态r.successful # 任务成功返回truer.traceback # 如果任务抛出了一个异常,可以获取原始的回溯信息

但是一般业务中很少用到,因为获取任务执行的结果需要阻塞,celery使用场景一般是不关心结果的。

五、Celery 使用案例 # seting.py# 设置配置BROKER_URL = 'amqp://username:password@localhost:5672/yourvhost'CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'CELERY_TASK_SERIALIZER = 'msgpack'CELERY_RESULT_SERIALIZER = 'msgpack'CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24CELERY_ACCEPT_CONTENT = ["msgpack"]CELERY_DEFAULT_QUEUE = "default" CELERY_QUEUES = { "default": { # 这是上面指定的默认队列 "exchange": "default", "exchange_type": "direct", "routing_key": "default" }} # app.py --- 初始化celery对象 from celery import Celeryimport setingfrom task import test_one, test_twocelery = Celery(__name__, include=["task"]) # 设置需要导入的模块# 引入配置文件celery.config_from_object(seting)if __name__ == '__main__': test_one.apply_async((2,2), routing_key='default', priority=0, exchange='default') # task.py --- 定义需要执行的任务from app import celery@celery.taskdef test_one(x, y): return x + y@celery.task(name="one_name")def test_two(x, y): return x * y

版权声明:该文观点仅代表作者本人。处理文章:请发送邮件至 三1五14八八95#扣扣.com 举报,一经查实,本站将立刻删除。