本文内容纲要：

-前言 -Celery对象 -创建异步任务的方法task -Task的一般属性 -调用异步任务 -1.app.send_task -2.Task.delay -3.Task.apply_async -获取任务结果和状态 -使用celery -小结 -参考

前言

前面我们了解了celery的基本使用后，现在对其常用的对象和方法进行分析。

Celery对象

核心的对象就是Celery了，初始化方法：

classCelery(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): #常用的需要配置的参数 main:如果作为__main__运行，则为主模块的名称。用作自动生成的任务名称的前缀 loader:当前加载器实例。 backend:任务结果url； amqp:AMQP对象或类名，一般不管； log:日志对象或类名； set_as_current:将本实例设为全局当前应用 tasks:任务注册表。 broker:使用的默认代理的URL,任务队列； include:每个worker应该导入的模块列表，以实例创建的模块的目录作为起始路径；

这些参数都是celery实例化的配置，我们也可以不写，然后使用config_from_object方法加载配置；

创建异步任务的方法task

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

使用默认的参数

@celery.task deffunction_name(): pass

指定相关参数

@celery.task(bind=True,name=name) deffunction_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访问相关的属性 defadd(self,x,y): try: logger.info(self.request.id) except: self.retry()#当任务失败则进行重试

自定义Task基类

importcelery

classMyTask(celery.Task): #任务失败时执行 defon_failure(self,exc,task_id,args,kwargs,einfo): print({0!r}failed:{1!r}.format(task_id,exc)) #任务成功时执行 defon_success(self,retval,task_id,args,kwargs): pass #任务重试时执行 defon_retry(self,exc,task_id,args,kwargs,einfo): pass

@task(base=MyTask) defadd(x,y): raiseKeyError()

#方法相关的参数 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；

调用异步任务

调用异步任务有三个方法，如下：

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

1.app.send_task

#tasks.py fromceleryimportCelery app=Celery() defadd(x,y): returnx+y app.send_task(tasks.add,args=[3,4])#参数基本和apply_async函数一样 #但是send_task在发送的时候是不会检查tasks.add函数是否存在的，即使为空也会发送成功，所以celery执行是可能找不到该函数报错；

2.Task.delay

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

@app.task defadd(x,y,z=0): returnx+y add.delay(30,40,z=5)#包括位置参数和关键字参数

3.Task.apply_async

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

add.apply_async(args=[30,40],kwargs={z:5}) #其他参数 task_id:为任务分配唯一id，默认是uuid; countdown:设置该任务等待一段时间再执行，单位为s； eta:定义任务的开始时间；eta=time.time()+10; expires:设置任务时间，任务在过期时间后还没有执行则被丢弃； retry:如果任务失败后,是否重试;使用true或false，默认为true shadow：重新指定任务的名字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,bzip2 headers：为任务添加额外的消息； 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#任务成功返回true r.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*24 CELERY_ACCEPT_CONTENT=["msgpack"] CELERY_DEFAULT_QUEUE="default" CELERY_QUEUES={ "default":{#这是上面指定的默认队列 "exchange":"default", "exchange_type":"direct", "routing_key":"default" } } #app.py---初始化celery对象 fromceleryimportCelery importseting fromtaskimporttest_one,test_two celery=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---定义需要执行的任务 fromappimportcelery @celery.task deftest_one(x,y): returnx+y @celery.task(name="one_name") deftest_two(x,y): returnx*y

小结

分析了celery任务一些方法参数和相关源码，接下来我们去研究celery更复杂的用法。

参考

http://docs.celeryproject.org/en/latest/userguide/tasks.html#task-options http://docs.jinkan.org/docs/celery/getting-started/first-steps-with-celery.html http://www.pythondoc.com/flask-celery/first.html https://blog.csdn.net/kk123a/article/details/74549117 https://blog.csdn.net/preyta/article/details/54288870

本文内容总结：前言，Celery对象，创建异步任务的方法task，Task的一般属性，调用异步任务，1.app.send_task，2.Task.delay，3.Task.apply_async，获取任务结果和状态，使用celery，小结，参考，

原文链接：https://www.cnblogs.com/cwp-bg/p/10575688.html