python之celery使用详解(二)
本文内容纲要:
-前言
-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