Celery任务队列高级用法

2019-06-24

对于普通的任务来说可能无法满足我们的任务需求，所以还需要了解一些进阶用法，Celery 提供了诸多调度方式，例如任务编排、根据任务状态执行不同的操作、重试机制等。

Celery任务队列高级用法 _ Escape - 图1

任务队列Celery高级用法

1. 相关组件介绍

了解 Celery 相关的生态环境

Celery 通过消息机制进行通信，通常使用中间人 Broker 为客户端和 Worker 调节。启动一个任务，客户端向消息队列发送一条消息，然后中间人 Broker 将消息传递给一个 Worker，最后由 Worker 进行执行分配的任务。可以有多个 Broker 和 Worker，用来提高 Celery 的高可用性以及横向扩展能力。Celery 是用 Python 编写的，但协议可以用任何语言实现，如 Node.js 语言的 node-celery 项目。

  • 支持的中间人

    • RabbitMQ
    • Redis
    • Amazon SQS

Celery任务队列高级用法 _ Escape - 图2

任务队列Celery高级用法

  • 支持的结果存储

    • AMQP、Redis
    • Memcached
    • SQLAlchemy、Django ORM
    • Apache Cassandra、Elasticsearch

  • 支持的并发

    • prefork (multiprocessing)
    • Eventlet、gevent
    • solo (single threaded)

  • 支持的序列化

    • pickle、json、yaml、msgpack
    • zlib、bzip2 compression
    • Cryptographic message signing

Celery任务队列高级用法 _ Escape - 图3

任务队列Celery高级用法

  • 支持的框架集成

    • Flask - 不需要
    • Tornado - tornado-celery
    • Pyramid - pyramid_celery
    • web2py - web2py-celery

2. 简单使用方式

主要的用途就是，帮助你快速上手和使用！

  • [简单使用] 我们将所有的内容都保存到一个文件中，而针对于大型项目而已，可能就需要创建独立的模块和配置文件了。

    • 安装环境依赖和启动基础服务
    • 编写 app 程序
    • 启动 Worker 服务并调用任务

bash

  1. # 使用docker运行MQ服务
  2. $ docker run -d -p 5462:5462 rabbitmq
  4. # 使用docker运行Redis服务
  5. $ docker run -d -p 6379:6379 redis
  7. # 使用pip安装依赖
  8. $ pip install celery

python

  1. # 最简单的Demo示例(tasks.py)
  2. from celery import Celery
  4. app = Celery('hello', broker='amqp://guest@localhost//')
  6. @app.task
  7. def add(x, y):
  8.     return x + y

bash

  1. # 启动celery进程
  2. $ celery worker -A hello --loglevel=info
  4. # 调用任务
  5. # 需要我们创建的实例任务并通过delay()进行调用
  6. $ python
  7. >>> from hello.tasks import add
  8. >>> add.delay(4, 4)

  • [后端存储] 如果需要保存任务状态，Celery 需要在某处存储任务的状态信息，其中内置了一些后端存储方案。通过配置文件中的 backend 参数，来指定后端存储方案。在已经配置后端存储的情况下，重新执行任务调用，可以保存对应的消息状态。

    • ORM(SQLAlchemy/Django)
    • Memcached
    • Redis
    • RPC(RabbitMQ/AMQP)
    • 自定义的后端结果存储中间件

bash

  1. # RPC
  2. app = Celery('hello', backend='rpc://', broker='pyamqp://')
  4. # Redis
  5. app = Celery('hello', backend='redis://localhost', broker='pyamqp://')

bash

  1. # 调用任务
  2. # 需要我们创建的实例任务并通过delay()进行调用
  3. $ python
  4. >>> from hello.tasks import add
  5. >>> result = add.delay(4, 4)
  6. >>> result.ready()  # 检测是否已经处理完毕
  7. False
  8. >>> result.get(timeout=1)  # 设置超时时间
  9. 8
  10. >>> result.get(propagate=False)  # 是否再次引发异常
  11. >>> result.traceback  # 任务出现异常进行回溯
  • [相关配置] 大多数情况下，使用默认的配置就可以满足，当然也可以根据我们的实际需求按需配置。针对大型的项目，建议使用专用配置模块进行配置，将所有的配置项集中化配置。

python

  1. # [配置项] 通过upate进行配置
  2. # 程序启动时导入的模块列表，便于Worker执行相应的任务
  3. from celery import Celery
  5. app = Celery('hello',
  6.              backend='redis://localhost',
  7.              broker='pyamqp://',
  8.              include=['hello.tasks'])
  10. app.conf.update(
  11.     task_serializer='json',  # 设置任务输出格式
  12.     accept_content=['json'],  # 忽略其他内容
  13.     result_serializer='json',  # 设置结果输出格式
  14.     timezone='Europe/Oslo',  # 时区
  15.     enable_utc=True,  # 启用UTC时钟
  16.     task_routes={  # 设置任务专用队列
  17.         'hello.tasks.add': 'low-priority',
  18.     },
  19.     task_annotations = {  # 任务进行限速
  20.         'hello.tasks.add': {'rate_limit': '10/m'}
  21.     }
  22. )
  24. if __name__ == '__main__':
  25.     app.start()
  27. # $ celery worker -A hello -Q hello
  28. # >>> from hello.tasks import add
  29. # >>> add.apply_async((2, 2), queue='hello')

python

  1. # [配置模块] 加载配置模块
  2. app.config_from_object('celery_config')
  4. # celery_config.py
  5. broker_url = 'pyamqp://'
  6. result_backend = 'rpc://'
  8. task_serializer = 'json'
  9. result_serializer = 'json'
  10. accept_content = ['json']
  11. timezone = 'Europe/Oslo'
  12. enable_utc = True
  14. task_routes = {
  15.     'hello.tasks.add': 'low-priority',
  16. }
  18. task_annotations = {
  19.     'hello.tasks.add': {'rate_limit': '10/m'}
  20. }

bash

  1. # inspect
  2. $ celery -A proj inspect --help
  4. # control
  5. $ celery -A proj control --help
  7. # event
  8. $ celery -A proj events --help
  10. # status
  11. $ celery -A proj status --help

bash

3. 进阶 - 定时任务

bash

  1. # Celery的提供的定时任务主要靠schedules来完成
  2. # 通过beat组件周期性的将任务发送给woker进行执行

python

  1. # period_task.py
  2. # 新建period_task.py文件并添加任务到配置文件中
  4. from learn_celery import app
  5. from celery.schedules import crontab
  7. @app.task
  8. def add(x, y):
  9.     print(x+y)
  10.     return x+y
  12. @app.task
  13. def say_hello(name):
  14.     return f'hello {name}'
  16. @app.on_after_configure.connect
  17. def setup_periodic_tasks(sender, **kwargs):
  18.     # 每10秒执行add任务
  19.     sender.add_periodic_task(10.0, add.s(1,3), name='1+3=')
  20.     sender.add_periodic_task(
  21.         # 每周一下午四点五十六执行say_hello任务
  22.         crontab(hour=16, minute=56, day_of_week=1),
  23.         say_hello.s('escape'), name='say_hello'
  24.     )

python

  1. # config.py
  3. BROKER_URL = 'redis://:escape@127.0.0.1:6379/0' # Broker配置
  4. CELERY_RESULT_BACKEND = 'redis://:escape@127.0.0.1:6379/0' # BACKEND配置
  5. CELERY_RESULT_SERIALIZER = 'json' # 结果序列化方案
  6. CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24 # 任务过期时间
  7. CELERY_TIMEZONE='Asia/Shanghai' # 时区配置
  8. CELERY_IMPORTS = (              # 指定导入的任务模块
  9.     'learn_celery.tasks',       # 普通任务
  10.     'learn_celery.period_task', # 定时任务
  11. )

bash

  1. # 启动worker和beat之后观察worker日志
  2. # 会发现定时任务会自动往队列中添加，之后被对应worker消费掉
  4. # 启动worker
  5. $ celery worker -A learn_celery.tasks -l debug
  7. # 启动beat
  8. $ celery beat -A learn_celery.period_task -l debug

bash

  1. # config.py
  2. # 还可以通过配置文件方式指定定时和计划任务
  4. from learn_celery import app
  5. from celery.schedules import crontab
  7. BROKER_URL = 'redis://:escape@127.0.0.1:6379/0' # Broker配置
  8. CELERY_RESULT_BACKEND = 'redis://:escape@127.0.0.1:6379/0' # BACKEND配置
  9. CELERY_RESULT_SERIALIZER = 'json' # 结果序列化方案
  10. CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24 # 任务过期时间
  11. CELERY_TIMEZONE='Asia/Shanghai' # 时区配置
  12. CELERY_IMPORTS = (              # 指定导入的任务模块
  13.     'learn_celery.tasks',       # 普通任务
  14.     'learn_celery.period_task', # 定时任务
  15. )
  17. app.conf.beat_schedule = {
  18.     'period_add_task': { # 添加计划任务
  19.         'task': 'learn_celery.period_task.add', # 任务路径
  20.         'schedule': crontab(hour=18, minute=16, day_of_week=1), # 执行周期
  21.         'args': (3, 4), # 传入参数
  22.     },
  23.     'say_hello': { # 添加计划任务
  24.         'task': 'learn_celery.period_task.say_hello', # 任务路径
  25.         'schedule': 10.0, # 执行周期
  26.         'args': ('escape',) # 传入参数
  27.     },
  28. }

bash

  1. # period_task.py
  2. # 此时的period_task.py只需要注册到woker中就行了
  4. from project import app
  6. @app.task
  7. def add(x,y):
  8.     print(x+y)
  9.     return x+y
  11. @app.task
  12. def say_hello(name):
  13.     return f'hello {name}'

4. 进阶 - 任务绑定

bash

  1. # Celery可通过任务绑定到实例获取到任务的上下文
  2. # 这样我们可以在任务运行时候获取到任务的状态，记录相关日志等

python

  1. # period_task.py
  2. # 通过bind参数将任务绑定，self指任务的上下文
  3. # 通过self获取任务状态，同时在任务出错时进行任务重试
  5. from learn_celery import app
  6. from celery.utils.log import get_task_logger
  8. logger = get_task_logger(__name__)
  10. @app.task(bind=True) # 绑定任务
  11. def add(self, x, y):
  12.     logger.info(self.request.__dict__) # 打印日志
  13.     try:
  14.         a=[]
  15.         a[10] == 1
  16.     except Exception as e:
  17.         # 出错每5秒尝试一次，总共尝试3次
  18.         raise self.retry(exc=e, countdown=5, max_retries=3)
  19.     return x+y

Celery任务队列高级用法 _ Escape - 图4

任务队列Celery快速入门

5. 进阶 - 内置钩子函数

bash

  1. # Celery在执行任务时候，提供了钩子方法用于在任务执行完成时候进行对应的操作
  2. # 在Task源码中提供了很多状态钩子函数如:
  3. #   on_success(成功后执行)
  4. #   on_failure(失败时候执行)
  5. #   on_retry(任务重试时候执行)
  6. #   after_return(任务返回时候执行)
  7. # 在进程中使用是我们只需要重写这些方法，完成相应的操作即可

python

  1. # period_task.py
  2. # 在以下示例中，我们继续修改period_task.py文件
  3. # 分别定义三个任务来演示任务失败、重试、任务成功后执行的操作
  5. from learn_celery import app
  6. from celery.utils.log import get_task_logger
  7. from celery import Task
  9. logger = get_task_logger(__name__)
  11. class demotask(Task):
  12.     # 任务成功执行
  13.     def on_success(self, retval, task_id, *args, **kwargs):
  14.         logger.info(f'task id:{task_id}, arg:{args}, successful!')
  16.     # 任务失败执行
  17.     def on_failure(self, exc, task_id, *args, **kwargs, einfo):
  18.         logger.info(f'task id:{task_id}, arg:{args}, failed! erros:{exc}')
  20.     # 任务重试执行
  21.     def on_retry(self, exc, task_id, *args, **kwargs, einfo):
  22.         logger.info(f'task id:{task_id}, arg:{args}, retry! einfo:{exc}')
  24. @app.task(base=demotask, bind=True)
  25. def add(self, x, y):
  26.     try:
  27.         a=[]
  28.         a[10] == 1
  29.     except Exception as e:
  30.         # 出错每5秒尝试一次，总共尝试1次
  31.         raise self.retry(exc=e, countdown=5, max_retries=1)
  32.     return x+y
  34. @app.task(base=demotask)
  35. def say_hello(name):
  36.     a=[]
  37.     a[10] == 1
  38.     return f'hello {name}'
  40. @app.task(base=demotask)
  41. def sum(a, b):
  42.     return f'a+b={a+b}'

python

  1. # config.py
  3. from learn_celery import app
  4. from celery.schedules import crontab
  6. BROKER_URL = 'redis://:escape@127.0.0.1:6379/0' # Broker配置
  7. CELERY_RESULT_BACKEND = 'redis://:escape@127.0.0.1:6379/0' # BACKEND配置
  8. CELERY_RESULT_SERIALIZER = 'json' # 结果序列化方案
  9. CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24 # 任务过期时间
  10. CELERY_TIMEZONE='Asia/Shanghai' # 时区配置
  11. CELERY_IMPORTS = (              # 指定导入的任务模块
  12.     'learn_celery.tasks',       # 普通任务
  13.     'learn_celery.period_task', # 定时任务
  14. )
  16. app.conf.beat_schedule = {
  17.     'add': { # 每10秒执行
  18.         'task': 'learn_celery.period_task.add',
  19.         'schedule': 10.0,
  20.         'args': (10, 12),
  21.     },
  22.     'say_hello': { # 每10秒执行
  23.         'task': 'learn_celery.period_task.say_hello',
  24.         'schedule': 10.0,
  25.         'args': ('escape',),
  26.     },
  27.     'sum': { # 每10秒执行
  28.         'task': 'learn_celery.period_task.sum',
  29.         'schedule': 10.0,
  30.         'args': (1, 3),
  31.     },
  32. }

Celery任务队列高级用法 _ Escape - 图5

任务队列Celery快速入门

6. 进阶 - 任务编排

python

  1. # 在很多情况下，一个任务需要由多个子任务或者一个任务需要很多步骤才能完成
  2. # Celery同样也能实现这样的任务，完成这类型的任务通过以下模块完成
  3. #   group: 并行调度任务
  4. #   chain: 链式任务调度
  5. #   chord: 类似group但分header和body2个部分;header可以是一个group任务，执行完成后调用body的任务
  6. #   map: 映射调度，通过输入多个入参来多次调度同一个任务
  7. #   starmap: 类似map，入参类似＊args
  8. #   chunks: 将任务按照一定数量进行分组

python

  1. # tasks.py
  3. from learn_celery import app
  5. @app.task
  6. def add(x, y):
  7.     return x+y
  9. @app.task
  10. def mul(x, y):
  11.     return x*y
  13. @app.task
  14. def sum(data_list):
  15.     res=0
  16.     for i in data_list:
  17.         res+=i
  18.     return res

python

  1. # consumer.py
  2. # group: 组任务
  3. # 组内每个任务并行执行
  5. from celery import group
  6. from learn_celery.tasks import add
  8. # 任务 [1+2, 1+2]
  9. res = group(add.s(1,2), add.s(1,2))()
  10. while True:
  11.     if res.ready():
  12.         print(f'res:{res.get()}')
  13.         break

Celery任务队列高级用法 _ Escape - 图6

任务队列Celery快速入门

python

  1. # consumer.py
  2. # chain: 链式任务
  3. # 链式任务中，默认上一个任务的返回结果作为参数传递给子任务
  5. from celery import group
  6. from learn_celery.tasks import add, mul, sum
  8. # 任务((1+2)+3)*3
  9. res = chain(add.s(1,2), add.s(3), mul.s(3))()
  10. while True:
  11.     if res.ready():
  12.         print(f'res:{res.get()}')
  13.         break

python

  1. # consumer.py
  2. # chord：任务分割
  3. # 分为header和body两部分
  4. # hearder任务执行完在执行body，其中hearder返回结果作为参数传递给body
  6. from celery import group
  7. from learn_celery.tasks import add, mul, sum
  9. # 任务(1+2)+(3*4)
  10. res = chord(header=[add.s(1,2), mul.s(3,4)], body=sum.s())()
  11. while True:
  12.     if res.ready():
  13.         print('res:{}'.format(res.get()))
  14.         break

python

  1. # consumer.py
  2. # chunks：任务分组
  3. # 按照任务的个数分组
  5. from learn_celery.tasks import add, mul, sum
  7. # 4代表每组的任务的个数
  8. res = add.chunks(zip(range(5), range(5)), 4)()
  9. while True:
  10.     if res.ready():
  11.         print('res:{}'.format(res.get()))
  12.         break

Celery任务队列高级用法 _ Escape - 图7

任务队列Celery快速入门

7. 异步调用原理

主要解释 delay 和 apply_async 的使用方法和区别

bash

  1. # 对于delay和apply_async都可以用来进行任务的调度
  2. # 本质上是delay对apply_async进行了再一次封装或者可以说是快捷方式
  3. # 两者都返回AsyncResult对象，以下是两个方法源码

python

  1. # --------------------------------------------
  2. # delay源码
  3. # --------------------------------------------
  4. def delay(self, *args, **kwargs):
  5.         """Star argument version of :meth:`apply_async`.
  7.         Does not support the extra options enabled by :meth:`apply_async`.
  9.         Arguments:
  10.             *args (Any): Positional arguments passed on to the task.
  11.             **kwargs (Any): Keyword arguments passed on to the task.
  12.         Returns:
  13.             celery.result.AsyncResult: Future promise.
  14.         """
  15.         return self.apply_async(args, kwargs)

python

  1. # --------------------------------------------
  2. # apply_async源码
  3. # --------------------------------------------
  4. def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,
  5.                     link=None, link_error=None, shadow=None, **options):
  6.         if self.typing:
  7.             try:
  8.                 check_arguments = self.__header__
  9.             except AttributeError:  # pragma: no cover
  10.                 pass
  11.             else:
  12.                 check_arguments(*(args or ()), **(kwargs or {}))
  14.         app = self._get_app()
  15.         if app.conf.task_always_eager:
  16.             with denied_join_result():
  17.                 return self.apply(args, kwargs, task_id=task_id or uuid(),
  18.                                   link=link, link_error=link_error, **options)
  20.         if self.__v2_compat__:
  21.             shadow = shadow or self.shadow_name(self(), args, kwargs, options)
  22.         else:
  23.             shadow = shadow or self.shadow_name(args, kwargs, options)
  25.         preopts = self._get_exec_options()
  26.         options = dict(preopts, **options) if options else preopts
  28.         options.setdefault('ignore_result', self.ignore_result)
  30.         return app.send_task(
  31.             self.name, args, kwargs, task_id=task_id, producer=producer,
  32.             link=link, link_error=link_error, result_cls=self.AsyncResult,
  33.             shadow=shadow, task_type=self,
  34.             **options
  35.         )

8. 使用注意事项

Celery 快速入门 > 分布式任务队列 Celery 入门与进阶

bash

  1. # -------------------------------------------------------------------
  2. # 1.在celery中worker启动时，如果是root用户则需要设置环境变量
  3. # -------------------------------------------------------------------
  4. $ export C_FORCE_ROOT='true'

bash

  1. # -------------------------------------------------------------------
  2. # 2.使用RabbitMQ或Redis作为Broker的话，生产环境永远不要使用关系型数据库
  3. # 3.Celery4.x开始不再支持Windows平台，如果需要在Windows开发请使用3.x的版本
  4. $ pip install celer[redis]==4.3.0
  5. $ pip install celery[librabbitmq]
  6. $ pip install celery[librabbitmq,redis,auth,msgpack]
  7. # -------------------------------------------------------------------

bash

  1. # 4.良好的配置操作
  2. # 禁用预取任务是为了防止Broker分配任务不均导致多次重复执行
  3. # 设置执行多少次任务之后进行销毁是为了防止卡死的出现和发生
  4. worker_prefetch_multiplier=0   # 禁用预取任务
  5. worker_max_tasks_per_child=50  # 每个worker执行50次任务即销毁

python

  1. # -------------------------------------------------------------------
  2. # 5.不要使用复杂对象作为任务函数的参数
  3. # -------------------------------------------------------------------
  5. # Good
  6. @app.task
  7. def my_task(user_id):
  8.     user = User.objects.get(id=user_id)
  9.     print(user.name)
  10.     ...
  12. # Bad
  13. @app.task
  14. def my_task(user):
  15.     print(user.name)
  16.     ...

9. 相关参考链接

授人玫瑰，手有余香！

文章作者: Escape

文章链接: https://www.escapelife.site/posts/6.3417e+60.html

版权声明: 本博客所有文章除特别声明外，均采用CC BY 4.0 许可协议。转载请注明来源 Escape !

Python Celery