本文详细介绍了 Celery 这一强大的异步任务队列系统，以及如何在 Django 框架中应用它来实现定时任务和异步处理，从而提高运维开发（DevOps）的效率和应用性能。

下面我们先认识一下 Celery 吧！

c5b25c70dce2d63ad0a9cdc5365ad0a0

Celery - 分布式任务队列

Celery 是一个简单、灵活、可靠的分布式系统，用于处理大量消息，同时为操作提供维护该系统所需的工具。 它是一个专注于实时处理的任务队列，同时也支持任务调度。

image

Celery 通过消息进行通信，通常使用经纪人 Borker 和 workers 之间进行调解。要启动一个任务，客户端会在队列中放入一条消息，然后经纪人将消息传递给工人。

一个 Celery 系统可以由多个 worker 和 broker 组成，从而实现高可用性和横向扩展。

Celery 是用 Python 编写的，但协议可以用任何语言实现。除了 Python 之外，还有 Node.js 的 Node-celery，PHP 客户端，golang 的 gocelery 和 Rust 的 rusty-celery。

10a75d0cfae886067dff7e2d7678b9d2

Celery 简介

• Part1: 什么是消息队列 任务队列是用来作为跨线程或机器分配工作的机制。
  任务队列的输入是称为任务的工作单元。
  专用工作进程不断监视任务队列以执行新工作。
  Celery 通过消息进行通信，通常使用代理来协调客户端和工作器。要启动任务，客户端会将消息添加到队列，然后代理会将该消息传递给工作器。Celery 系统可以由多个工作者和代理组成，从而实现高可用性和水平扩展。
  Celery 是用 Python 编写的，但该协议可以用任何语言实现。除了 Python，还有适用于 Node.js 的node-celery和node-celery-ts，以及PHP 客户端。语言互操作性也可以通过公开 HTTP 端点并拥有请求它的任务（webhook）来实现。
• Part2: 我需要准备什么 Celery需要消息传输来发送和接收消息。
  RabbitMQ 和 Redis 代理传输功能齐全，但也支持大量其他实验性解决方案，包括使用 SQLite 进行本地开发。
  Celery可以在单台机器上运行，也可以在多台机器上运行，甚至跨数据中心运行。
• Part3: Celery 初体验
• 选择 Broker （MQ/Redis/其他 Broker）
• 安装 Celery(pip install celery)
• 应用:(创建 Celery 实例或者简称 app,它是 Celery 中执行所有操作的入口点,例如创建任务和管理 Worker)
• from celery import Celery
  app = Celery('tasks', broker='pyamqp://guest@localhost//') @app.task
  def add(x, y):
  return x + y
• 运行 Celery 的 Worker: (celery -A django_template_v1 worker -Q default -c 20 -l info -n default,生产环境中可以将 Worker 最为守护进程在后台运行,可以参考使用Supervisord)
• 调用任务:(要调用我们的任务，您可以使用该 delay()方法,这是方法的一个便捷的快捷方式apply_async() ，可以更好地控制任务的执行),该任务会交给 Worker 处理,可以在控制台输出验证,任务调用会返回一个AsyncResult 实例。这可用于检查任务的状态、等待任务完成或获取其返回值（或者如果任务失败，则获取异常和回溯）
• 保存任务结果:如果您想要跟踪任务的状态，Celery 需要将状态存储或发送到某个地方。有几种内置结果后端可供选择：SQLAlchemy / Django ORM、 MongoDB、Memcached、Redis、RPC（RabbitMQ /AMQP）以及 – 或者您可以定义自己的后端 (CELERY_RESULT_BACKEND = "django-db" )
• 监控任务执行: Flower 是一款用于监控和管理 Celery 集群的开源 Web 应用程序。它提供有关 Celery 工作器和任务状态的实时信息。
• 配置项:
• CELERY_ONCE_URL = "{}1".format(REDIS_URL)
  CELERY_RESULT_BACKEND = "django-db" # 使用django数据库
  CELERY_BROKER_URL = "{}8".format(REDIS_URL)
  CELERY_BEAT_SCHEDULER = "django_celery_beat.schedulers:DatabaseScheduler"
  # CELERY_ACCEPT_CONTENT = ["application/json"]
  CELERY_ACCEPT_CONTENT = ["json", "application/text"]
  CELERY_TASK_SERIALIZER = "json"
  CELERY_RESULT_SERIALIZER = "json"
  CELERYD_CONCURRENCY = 40 # celery worker的并发数
  CELERY_MAX_TASKS_PER_CHILD = 20
  CELERYD_MAX_TASKS_PER_CHILD = 20
  # 这个表示每个工作的进程／线程／绿程 在执行 n 次任务后，主动销毁，之后会起一个新的。主要解决一些资源释放的问题。
  CELERY_RESULT_EXPIRES = 7200 # celery任务执行结果的超时时间，
  CELERY_TASK_RESULT_EXPIRES = 7200 # 这个表示保存任务结果的时长，这个时间会被设置到 redis 里面（假设 backend 是 redis ），如果抓取数据量大的话，是可以缩短保存的时间，# 节省 backend 的资源（ redis 主要是内存）消耗，默认是 24 小时（ 86400 ），单位是秒。
  CELERY_TASK_TIME_LIMIT = 7200 # 单个任务的运行时间不超过此值，否则会被SIGKILL 信号杀死
  CELERY_ENABLE_UTC = False
  CELERY_WORKER_CONCURRENCY = 40 # celery worker的并发数
  DJANGO_CELERY_BEAT_TZ_AWARE = True
  CELERY_TIMEZONE = "Asia/Shanghai" # celery 时区问题
  CELERY_TASK_TRACK_STARTED = True
• Part4: Celery 是什么

简单的:可制作最简单的APP

高可用的:当连接丢失或失败时，Worker 和 Clients 将自动重试，并且一些 Broker 通过Primary/Primary或Primary/Replica复制的方式支持 HA

快速地:单个 Celery 进程每分钟可以处理数百万个任务，往返延迟仅为亚毫秒（使用 RabbitMQ、librabbitmq 和优化设置）。

灵活的:Celery的几乎每个部分都可以扩展或单独使用，自定义池实现、序列化器、压缩方案、日志记录、调度程序、消费者、生产者、代理传输等等。

支持多种代理Broker: MQ\Redis\...

支持多种任务结果存储库: MQ\Redis\Memcache\SQLAlchemy\Django ORM\Mongo DB ...

支持并发

支持序列化

• Part5: Celery 有哪些特征监控
  调度
  工作流
  资源泄露保护
  时间和速率限制
  Worker组件
• Part6: 框架集成

框架名称 框架集成库 Pyramid https://pypi.org/project/pyramid_celery/ Pylons https://pypi.org/project/celery-pylons/ Flask not needed web2py https://pypi.org/project/web2py-celery/ Tornado https://pypi.org/project/tornado-celery/ Tryton https://pypi.org/project/celery_tryton/ Django https://docs.celeryq.dev/en/stable/django/first-steps-with-django.html#django-first-steps

image

• Part7: 安装和使用

pip install -U Celery
依赖包: Celery 还定义了一组包，可用于安装 Celery 和给定功能的依赖项。您可以在需求中或在pip 命令行中使用括号指定这些。可以通过用逗号分隔来指定多个包。

pip install "celery[librabbitmq]"

pip install "celery[librabbitmq,redis,auth,msgpack]"

有以下捆绑包可用：

序列化器
celery[auth]：
用于使用auth安全序列化器。

celery[msgpack]：
用于使用 msgpack 序列化器。

celery[yaml]：
用于使用 yaml 序列化器。

并发
celery[eventlet]：
使用https://pypi.org/project/eventlet/池。

celery[gevent]：
使用https://pypi.org/project/gevent/池。

传输和后端
celery[librabbitmq]：
用于使用 librabbitmq C 库。

celery[redis]：
使用 Redis 作为消息传输或结果后端。

celery[sqs]：
使用 Amazon SQS 作为消息传输（实验性）。

celery[tblib]：
使用该task_remote_tracebacks功能。

celery[memcache]：
使用 Memcached 作为结果后端（使用https://pypi.org/project/pylibmc/）

celery[pymemcache]：
使用 Memcached 作为结果后端（纯 Python 实现）。

celery[cassandra]：
使用 Apache Cassandra/Astra DB 作为带有 DataStax 驱动程序的结果后端。

celery[couchbase]：
使用 Couchbase 作为结果后端。

celery[arangodb]：
使用 ArangoDB 作为结果后端。

celery[elasticsearch]：
使用 Elasticsearch 作为结果后端。

celery[riak]：
使用 Riak 作为结果后端。

celery[dynamodb]：
使用 AWS DynamoDB 作为结果后端。

celery[zookeeper]：
使用 Zookeeper 作为消息传输。

celery[sqlalchemy]：
使用 SQLAlchemy 作为结果后端（支持）。

celery[pyro]：
用于使用 Pyro4 消息传输（实验性的）。

celery[slmq]：
用于使用 SoftLayer 消息队列传输（实验性）。

celery[consul]：
使用 Consul.io Key/Value 存储作为消息传输或结果后端（实验性的）。

celery[django]：
指定 Django 支持的最低版本。

您可能不应该在您的要求中使用它，它仅供参考。

celery[gcs]：
使用 Google Cloud Storage 作为结果后端（实验性）。

f99f16f4296737f92d89e3a1ebabc30b

Django 中如何使用 Celery 实现定时任务

7a4206fd44828d21cef73ffdc82e244d

• Part1: 安装 Django

Django==3.2.19
django-simple-captcha==0.5.17
django-timezone-field==4.1.2
djangorestframework==3.13.1
djangorestframework-simplejwt==5.2.0

• Part2:安装 redis

redis==4.5.4
django-redis==5.2.0

• Part3:配置 mysql-db

PyMySQL==1.0.2
pymongo==4.3.3

• Part4:安装 django 的 redis 和 celery 依赖

celery==5.2.7
django-celery-beat==2.2.0
django-celery-results==2.0.1
celery-once==3.0.1

• Part5:celery 配置(celery.py 和 settings.py)

celery.py

# -*- coding: utf-8 -*-

from __future__ import absolute_import, unicode_literals
import os
from celery_once import QueueOnce
from django.apps import apps
# set the default Django settings module for the 'celery' program.
from django.utils import timezone
from kombu import Queue, Exchange
from celery import Celery, platforms
from django.conf import settings

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'django_template_v1.settings')

app = Celery('product_hub')

app.conf.ONCE = {
'backend': 'celery_once.backends.Redis',
'settings': {
  'url': settings.CELERY_ONCE_URL,
  'default_timeout': 60 * 60
}
}

app.now = timezone.now
app.config_from_object('django.conf:settings', namespace='CELERY')

default_exchange = Exchange('default', type='direct')
config_exchange = Exchange('config', type='direct')

app.conf.task_time_limit = 86400
app.conf.worker_prefetch_multiplier = 10
app.conf.worker_max_tasks_per_child = 100
app.conf.task_default_queue = 'default'
app.conf.task_default_exchange = 'default'
app.conf.task_default_routing_key = 'default'
app.conf.task_default_exchange_type = 'direct'
app.conf.task_queues = (
  Queue('default', default_exchange, routing_key='default'),
  Queue('config', config_exchange, routing_key='config'),
)
platforms.C_FORCE_ROOT = True
app.autodiscover_tasks(lambda: [n.name for n in apps.get_app_configs()])


class ProductHubTask(QueueOnce):

  # def run(self, *args, **kwargs):
  #     pass

  max_retries = 3
  # autoretry_for = (Exception, KeyError, RuntimeError)
  retry_kwargs = {'max_retries': 1}
  retry_backoff = False

  def on_failure(self, exc, task_id, args, kwargs, einfo):
      print(str(einfo))
      return super(ProductHubTask, self).on_failure(exc, task_id, args, kwargs, einfo)

  def on_retry(self, exc, task_id, args, kwargs, einfo):
      # print('task retry, reason: {0}'.format(exc))
      print(str(einfo))
      return super(ProductHubTask, self).on_failure(exc, task_id, args, kwargs, einfo)

settings.py

CELERY_ONCE_URL = "{}1".format(REDIS_URL)
CELERY_RESULT_BACKEND = "django-db"  # 使用django数据库
CELERY_BROKER_URL = "{}8".format(REDIS_URL)
CELERY_BEAT_SCHEDULER = "django_celery_beat.schedulers:DatabaseScheduler"
# CELERY_ACCEPT_CONTENT = ["application/json"]
CELERY_ACCEPT_CONTENT = ["json", "application/text"]
CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json"
CELERYD_CONCURRENCY = 40  # celery worker的并发数
CELERY_MAX_TASKS_PER_CHILD = 20
CELERYD_MAX_TASKS_PER_CHILD = 20
# 这个表示每个工作的进程／线程／绿程 在执行 n 次任务后，主动销毁，之后会起一个新的。主要解决一些资源释放的问题。
CELERY_RESULT_EXPIRES = 7200  # celery任务执行结果的超时时间，
CELERY_TASK_RESULT_EXPIRES = 7200 # 这个表示保存任务结果的时长，这个时间会被设置到 redis 里面（假设 backend 是 redis ），如果抓取数据量大的话，是可以缩短保存的时间，# 节省 backend 的资源（ redis 主要是内存）消耗，默认是 24 小时（ 86400 ），单位是秒。
CELERY_TASK_TIME_LIMIT = 7200 # 单个任务的运行时间不超过此值，否则会被SIGKILL 信号杀死
CELERY_ENABLE_UTC = False
CELERY_WORKER_CONCURRENCY = 40  # celery worker的并发数
DJANGO_CELERY_BEAT_TZ_AWARE = True
CELERY_TIMEZONE = "Asia/Shanghai"  # celery 时区问题
CELERY_TASK_TRACK_STARTED = True

------------INSTALLED_APPS-------------------
"django_celery_beat",
"django_celery_results.apps.CeleryResultConfig",

• Part6:创建 task 任务 app 下新建 tasks.py 文件 # import json
  import logging
  # from datetime import datetime, date
  from celery import shared_task
  rom django_template_v1.celery import ProductHubTask
  logger = logging.getLogger("celery")
  @shared_task(base=ProductHubTask)
  def sync_redis_cart_to_mysql():
  """同步redis的数据到mysql做持久化保存"""
  pass
  
  
• Part7:启动 celery 的 worker

celery -A django_template_v1 worker -Q default -c 20  -l info -n default

• Part8:异步调用或定时调用

image

image

上述就是在 Django 中如何基础的使用 Celery 来实现异步和定时任务

a670552f577dc1222e3a3c36f4506bb3

最后

个人工作中的见解,可以参考来配置试试看,若文章有错误的地方欢迎指正。

image