标签：异步 task 重试 celery 默认 Celery 任务 Task
目录
一、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.task def 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 celery
class 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.py
from celery import Celery
app = Celery()
def add(x, y):
return x+y
app.send_task('tasks.add',args=[3,4]) # 参数基本和apply_async函数一样
方法二：Task.delay
delay 方法是 apply_async 方法的简化版，不支持执行选项，只能传递任务的参数。
from celery import Celery app = Celery() @app.task def add(x, y, z=0): return x + y add.delay(30, 40, z=5) # 包括位置参数和关键字参数
方法三：Task.apply_async
apply_async 支持执行选项，它会覆盖全局的默认参数和定义该任务时指定的执行选项，本质上还是调用了 send_task 方法；
from celery import Celery
app = Celery()
@app.task
def 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，默认为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对象 from celery import Celery import seting from task import test_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 --- 定义需要执行的任务 from app import celery @celery.task def test_one(x, y): return x + y @celery.task(name="one_name") def test_two(x, y): return x * y
标签：异步,task,重试,celery,默认,Celery,任务,Task 来源： https://blog.51cto.com/u_15187242/2746788
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。