celery笔记三之task和task的调用

阿布的微信

本文首发于公众号：Hunter后端
原文链接：celery笔记三之task和task的调用

这一篇笔记介绍 task 和 task 的调用。
以下是本篇笔记目录：

• 基础的 task 定义方式
  
• 日志处理
  
• 任务重试
  
• 忽略任务运行结果
  
• task 的调用
  

1、基础的 task 定义方式

前面两篇笔记中介绍了最简单的定义方式，使用 @app.task 作为装饰器：

1. @app.task
2. def add(x, y):
3.     return x + y

复制代码

如果是在 Django 系统中使用 celery，需要定义一个延时任务或者周期定时任务，可以使用 @shared_task 来修饰

1. from celery import shared_task
3. @shared_task
4. def add(x, y):
5.     return x + y

复制代码

在 Django 系统中使用 celery 的方式会在接下来的几篇笔记中介绍道。
多个装饰器
如果是 celery 的任务和其他装饰器一起联用，记得将 celery 的装饰器放在最后使用，也就是列表的最前面：

1. @app.task
2. @decorator1
3. @decorator2
4. def add(x, y):
5.     return x + y

复制代码

task名称
每个 task 都有一个唯一的名称用来标识这个 task，如果我们在定义的时候不指定，系统会为我们默认一个名称，这些名称会在 celery 的 worker 启动的时候被系统扫描然后输出一个列表展示。
还是上一篇笔记中我们定义的两个 task，我们给其中一个指定 name：

1. #tasks1.py
2. from .celery import app
5. @app.task(name="tasks1.add")
6. def add(x, y):
7.     return x + y

复制代码

可以观察在 celery 的 worker 启动的时候，会有一个输出：

1. [tasks]
2.   . proj.tasks2.mul
3.   . tasks1.add

复制代码

可以看到这个地方，系统就会使用我们定义的 name 了。
2、日志处理

我们可以在启动 worker 的时候指定日志的输出，定义格式如下：

1. celery -A proj worker -l INFO --logfile=/Users/hunter/python/celery_log/celery.log

复制代码

在 task 中的定义可以使用 celery 中方法：

1. from celery.utils.log import get_task_logger
3. logger = get_task_logger(__name__)

复制代码

也可以直接使用 logging 模块：

1. import logging
3. logger1 = logging.getLogger(__name__)

复制代码

直接在 task 中输出：

1. @app.task(name="tasks1.add")
2. def add(x, y):
3.     logger.info("this is from logger")
4.     return x + y

复制代码

然后在 worker 启动时指定的日志文件就会有我们打印出的日志内容：

1. [2022-07-24 16:28:33,210: INFO/ForkPoolWorker-7] tasks1.add[4db4b0fc-c6ca-472a-8847-ae42e0a7959a]: this is from logger
2. [2022-07-24 16:28:33,224: INFO/ForkPoolWorker-7] Task tasks1.add[4db4b0fc-c6ca-472a-8847-ae42e0a7959a] succeeded in 0.016244667931459844s: 3

复制代码

3、任务重试

对于一个 task，我们可以对其设置 retry 参数来指定其在任务执行失败后会重试几次，以及隔多长时间重试。
比如对于下面的 div() 函数，我们来输入除数为 0 的情况查看重试的功能。
当然，这里我们是故意输入参数错误，在实际的项目中可能会是其他的原因造成任务失败，比如数据库连接失败等
任务重试的参数也都在 @app.task() 中定义：

1. # tasks1.py
3. @app.task(autoretry_for=(Exception, ),  default_retry_delay=10, retry_kwargs={'max_retries': 5})
4. def div(x, y):
5.     return x / y

复制代码

在这里，autoretry_for 表示的是某种报错情况下重试，我们定义的 Exception 表示任何错误都重试。
如果只是想在某种特定的 exception 情况下重试，将那种 exception 的值替换 Exception 即可。
default_retry_delay 表示重试间隔时长，默认值是 3 * 60s，即三分钟，是以秒为单位，这里我们设置的是 10s。
retry_kwargs 是一个 dict，其中有一个 max_retries 参数，表示的是最大重试次数，我们定为 5
然后可以尝试调用这个延时任务：

1. from proj.tasks1 import div
2. div.delay(1, 0)

复制代码

然后可以看到在日志文件会有如下输出：

1. [2022-07-24 16:59:35,653: INFO/ForkPoolWorker-7] Task proj.tasks1.div[1f65c410-1b2a-4127-9d83-a84b1ad9dd2c] retry: Retry in 10s: ZeroDivisionError('division by zero',)

复制代码

且每隔 10s 执行一次，一共执行 5 次，5次之后还是不成功则会报错。
retry_backoff 和 retry_backoff_max
还有一个 retry_backoff 和 retry_backoff_max 参数，这两个参数是用于这种情况：如果你的 task 依赖另一个 service 服务，比如会调用其他系统的 API，然后这两个参数可以用于避免请求过多的占用服务。
retry_backoff 参数可以设置成一个 布尔型数据，为 True 的话，自动重试的时间间隔会成倍的增长
第一次重试是 1 s后
第二次是 2s 后
第三次是 4s 后
第四次是 8s 后
...
如果 retry_backoff 参数是一个数字，比如是 3，那么后续的间隔时间则是 3 的倍数增长
第一次重试 3s 后
第二次是 6s 后
第三次是 12s 后
第四次是 24s 后
retry_backoff_max 是重试的最大的间隔时间，比如重试次数设置的很大，retry_backoff 的间隔时间重复达到了这个值之后就不再增大了。
这个值默认是 600s，也就是 10分钟。
我们看一下下面这个例子：

1. # tasks1.py
3. @app.task(autoretry_for=(Exception, ), retry_backoff=2, retry_backoff_max=40, retry_kwargs={'max_retries': 8})
4. def div(x, y):
5.     return x / y

复制代码

关于重试的机制，理论上应该是按照我们前面列出来的重试时间间隔进行重试，但是如果我们这样直接运行 div.delay()，得出的间隔时间是不定的，是在 0 到 最大值之间得出的一个随机值。
这样产生的原因是因为还有一个 retry_jitter 参数，这个参数默认是 True，所以时间间隔会是一个随机值。
如果需要任务延时的间隔值是按照 retry_backoff 和 retry_backoff_max 两个设定值来运行，那么则需要将 retry_jitter 值设为 False。

1. # tasks1.py
3. @app.task(autoretry_for=(Exception, ), retry_backoff=2, retry_backoff_max=40, retry_jitter=False, retry_kwargs={'max_retries': 8})
4. def div(x, y):
5.     return x / y

复制代码

然后运行 div 的延时任务，就可以看到延时任务按照规律的间隔时间重试了，以下是日志：

1. [2022-07-24 19:00:38,588: INFO/ForkPoolWorker-7] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] retry: Retry in 2s: ZeroDivisionError('division by zero',)
2. [2022-07-24 19:00:40,662: INFO/MainProcess] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] received
3. [2022-07-24 19:00:40,664: INFO/ForkPoolWorker-7] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] retry: Retry in 4s: ZeroDivisionError('division by zero',)
4. [2022-07-24 19:00:44,744: INFO/MainProcess] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] received
5. [2022-07-24 19:00:44,746: INFO/ForkPoolWorker-7] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] retry: Retry in 8s: ZeroDivisionError('division by zero',)
6. [2022-07-24 19:00:52,870: INFO/MainProcess] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] received
7. [2022-07-24 19:00:52,872: INFO/ForkPoolWorker-7] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] retry: Retry in 16s: ZeroDivisionError('division by zero',)
8. [2022-07-24 19:01:09,338: INFO/MainProcess] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] received
9. [2022-07-24 19:01:09,340: INFO/ForkPoolWorker-7] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] retry: Retry in 32s: ZeroDivisionError('division by zero',)
10. [2022-07-24 19:01:41,843: INFO/MainProcess] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] received
11. [2022-07-24 19:01:41,845: INFO/ForkPoolWorker-7] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] retry: Retry in 40s: ZeroDivisionError('division by zero',)
12. [2022-07-24 19:02:21,923: INFO/MainProcess] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] received
13. [2022-07-24 19:02:21,925: INFO/ForkPoolWorker-7] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] retry: Retry in 40s: ZeroDivisionError('division by zero',)
14. [2022-07-24 19:03:02,001: INFO/MainProcess] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] received
15. [2022-07-24 19:03:02,003: INFO/ForkPoolWorker-7] Task proj.tasks1.div[7e689dcf-8069-4f17-8815-fe58f9800fc0] retry: Retry in 40s: ZeroDivisionError('division by zero',)

复制代码

因为我们设置的重试间隔时间最大为 40s，所以这个地方延时间隔时间到了 40 之后，就不再往上继续增长了。
4、忽略任务运行结果

有时候延时任务的结果我们并不想保存，但是我们配置了 result_backend 参数，这个时候我们有三种方式不保存运行结果。
1.ignore_result=True 不保存任务运行的结果

1. @app.task(ignore_result=True)
2. def add(x, y):
3.     return x + y

复制代码

2.app.conf 配置
也可以通过 app.conf 的配置来禁用结果的保存：

1. app.conf.update(
2.     task_ignore_result=True
3. )

复制代码

3.执行单个任务的时候禁用

1. from proj.tasks1 import add
2. add.apply_async((1, 2), ignore_result=True)

复制代码

apply_async() 函数的作用相当于是带参数的 delay()，或者 delay() 是简化版的 apply_async()，这个我们下面会介绍。
5、task 的调用

前面简单两个简单的调用方法，一个是 apply_async()，一个是 delay()。
简单来说就是 delay() 是不带参数执行的 apply_async()。
以下用 add() 函数为例介绍一下他们的用法：
delay()
纯粹的延时任务，只能如下操作：

1. add.delay(1, 2)

复制代码

apply_async()
带参数的用法，add() 函数的参数用 () 包起来：

1. add.apply_async((1, 2))

复制代码

也可以带其他参数，比如上面介绍的不保存运行结果：

1. add.apply_async((1, 2), ignore_result=True)

复制代码

这个函数还可以指定延时的时间：
countdown参数
现在开始 10s 后开始运行：

1. add.apply_async((1, 2), countdown=10)

复制代码

eta参数
也可以用 eta 参数来指定 10s 后运行：

1. from datetime import datetime, timedelta
3. now = datetime.now()
4. add.apply_async((1, 2), eta=now + timedelta(seconds=10))

复制代码

expires参数
这个是用来设置过期的参数：

1. add.apply_async((1, 2), countdown=60, expires=120)

复制代码

上面的参数表示，距现在60秒后开始执行，两分钟后过期
如果想获取更多后端相关文章，可扫码关注阅读：

来源:https://www.cnblogs.com/hunterxiong/p/17478782.html
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！