python 异步任务框架 Celery 入门

发布于 更新于
本文目录
  1. Celery
  2. 1. 简介
  3. 版本要求
  4. 安装
  5. 捆绑包
  6. 2. 简单使用
  7. 1. 选择一个broker
  8. 2. 编写一个celery应用
  9. 3.运行celery的worker服务
  10. 4. 调用任务
  11. 5. 保存结果
  12. 3. 在应用中使用celery
  13. 创建项目
  14. proj/celery.py
  15. proj/tasks.py
  16. 启动worker
  17. 调用任务
  18. 4. 在django中使用celery
  19. 使用@shared_task装饰器

技术文章Linux与数据库

Celery是使用python编写的分布式任务调度框架。

Celery

1. 简介

Celery是使用python编写的分布式任务调度框架。

它有几个主要的概念：

  • celery应用

    • 用户编写的代码脚本，用来定义要执行的任务，然后通过broker将任务发送到消息队列中

  • broker

    • 代理，通过消息队列在客户端和worker之间进行协调。

    • celery本身并不包含消息队列，它支持一下消息队列

      • RabbitMQ

      • Rdis

      • Amazon SQS

      • Zookeeper

    • 更多关于Broker见官方文档

  • backend

    • 数据库，用来存储任务返回的结果。

  • worker

    • 工人，用来执行broker分派的任务。

  • 任务

    • 任务，定义的需要执行的任务

版本要求

Celery5.1 要求：

  • python（3.6，3.7，3.8）

Celery 是一个资金最少的项目，所以我们不支持 Microsoft Windows。

更多更详细的版本要求见官方文档

安装

使用pip安装：

pip install -U Celery

捆绑包

Celery还定义了一组包，用于安装Celery和给定的依赖项。

可以在pip命令中实现中括号来指定这些依赖项。

pip install "celery[librabbitmq]"

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

具体支持的依赖包见官方文档

2. 简单使用

1. 选择一个broker

使用celery首先需要选择一个消息队列。安装任意你熟悉的前面提到的celery支持的消息队列。

2. 编写一个celery应用

首先我们需要编写一个celery应用，它用来创建任务和管理wokers，它要能够被其他的模块导入。

创建一个tasks.py文件：

from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379/0')

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

第一个参数tasks是当前模块的名称，它可以省略，建议以当前模块名为名称。

第二个关键字参数broker='redis://localhost:6379/0'指定我们使用redis作为消息队列，并指定连接地址。

3.运行celery的worker服务

cd到tasks.py所在目录，然后运行下面的命令来启动worker服务

celery -A tasks worker --loglevel=INFO

4. 调用任务

>>> from tasks import add
>>> add.delay(4,4)

通过调用任务的delay来执行对应的任务。celery会把执行命令发送到broker，broker再将消息发送给worker服务来执行，如果一切正常你将会在worker服务的日志中看到接收任务和执行任务的日志。

5. 保存结果

如果你想要跟踪任务的状态以及保存任务的返回结果，celery需要把它发送到某个地方。celery提供多种结果后端。

我们这里以reids为例，修改tasks.py中的代码，添加一个redis后端。

app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/1')

更多结果后端见官方文档。

重新启动worker服务，重新打开python解释器

>>> from tasks import add
>>> result = add.delay(4,4)

ready()方法返回任务是否执行完成：

>>> result.ready()
False

还可以等待结果完成，但很少使用这种方法，因为它将异步调用转换为同步调用

>>> result.get(timeout=1)
8

3. 在应用中使用celery

创建项目

项目结构：

proj/__init__.py
    /celery.py
    /tasks.py

proj/celery.py

from celery import Celery

app = Celery('proj',
             broker='redis://localhost:6379/0',
             backend='redis://localhost:6379/1',
             include=['proj.tasks'])

# 配置
app.conf.update(
    result_expires=3600, # 结果过期时间
)

在这个模块中我们创建了一个Celery模块。要在你的项目中使用celery只需要导入此实例。

proj/tasks.py

from .celery import app


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


@app.task
def mul(x, y):
    return x * y


@app.task
def xsum(numbers):
    return sum(numbers)

启动worker

celery -A proj worker -l INFO

调用任务

>>> from proj.tasks import add

>>> add.delay(2, 2)

4. 在django中使用celery

要在你的django项目中使用celery，首先需要定义一个Celery的实例。

如果你又django项目如下：

- proj/
  - manage.py
  - proj/
    - __init__.py
    - settings.py
    - urls.py

那么推荐的方法是创建一个新的proj/proj/celery.py模块来定义芹菜实例:

file:proj/proj/celery.py

import os

from celery import Celery

# 为`celery`设置默认的django设置模块
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')

app = Celery('proj')

# 设置配置来源
app.config_from_object('django.conf:settings', namespace='CELERY')

# 加载所有的已注册django应用中的任务
app.autodiscover_tasks()


@app.task(bind=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

然后你需要在你的proj/proj/__init__.py模块中导入这个应用程序。这样就可以保证Django启动时加载应用程序，以便于@shared_task装饰器的使用。

proj/proj/__init__.py:

from .celery import app as celery_app

__all__ = ('celery_app',)

请注意，此示例项目布局适用于较大的项目，对于简单的项目，可以使用包含定义应用程序和任务的单个模块。

接下来我们来解释一下celery.py中的代码，首先，我们设置celery命令行程序的环境变量DJANGO_SETTINGS_MODULE的默认值：

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

这一行的作用是加载当前django项目的环境设置，特别是当需要在异步任务中用到orm。它必须在创建应用程序实例之前。

app = Celery('proj')

我们还添加了Django设置模块作为Celery的配置源。这意味着我们不必使用多个配置文件，而是直接在Django的配置文件中配置Celery。

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

大写命名空间意味着所有Celery配置项必须以大写指定，并以CELERY_开头，因此例如broker_url设置变为CELERY_BROKER_URL。

例如，Django项目的配置文件可能包括:

settings.py

CELERY_TIMEZONE = "Asia/Shanghai"
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_TIME_LIMIT = 30*60

接下来，可重用应用程序的常见做法是在单独的tasks.py模块中定义所有任务，Celery有一种方法可以自动发现这些模块：

app.autodiscover_tasks()

使用上面的行，Celery 将按照tasks.py约定自动从所有已安装的应用程序中发现任务：

- app1/
    - tasks.py
    - models.py
- app2/
    - tasks.py
    - models.py

这样就不必手动将各个模块添加到CELERY_IMPORTS设置中。

使用@shared_task装饰器

我们编写的任务可能会存在于可重用的应用程序中，而可重用的应用程序不能依赖与项目本身，因此无法直接导入celery应用实例。

@shared_task装饰器可以让我们无需任何具体的celery实例创建任务：

demoapp/tasks.py

# Create your tasks here

from demoapp.models import Widget

from celery import shared_task


@shared_task
def add(x, y):
    return x + y


@shared_task
def mul(x, y):
    return x * y


@shared_task
def xsum(numbers):
    return sum(numbers)


@shared_task
def count_widgets():
    return Widget.objects.count()


@shared_task
def rename_widget(widget_id, name):
    w = Widget.objects.get(id=widget_id)
    w.name = name
    w.save()

来源：testingpai

继续阅读