Celery怎么创建

引言

在现代的Web应用开发中，异步任务处理是一个非常重要的功能。无论是发送邮件、处理图像、还是执行复杂的计算任务，异步任务处理都能显著提高应用的性能和用户体验。Celery是一个强大的分布式任务队列系统，广泛应用于Python项目中。本文将详细介绍如何创建和使用Celery，帮助开发者快速上手这一工具。

什么是Celery？

Celery是一个基于分布式消息传递的异步任务队列/作业队列。它专注于实时操作，但也支持任务调度。Celery是用Python编写的，但可以通过协议与其他语言进行通信。它支持多种消息代理，如RabbitMQ、Redis等，并且可以与Django、Flask等Web框架无缝集成。

安装Celery

在开始使用Celery之前，首先需要安装它。可以通过pip来安装Celery：

pip install celery

此外，还需要安装一个消息代理。本文以RabbitMQ为例，安装RabbitMQ可以通过以下命令：

sudo apt-get install rabbitmq-server

安装完成后，启动RabbitMQ服务：

sudo service rabbitmq-server start

创建Celery应用

1. 初始化Celery应用

首先，创建一个Python文件（例如celery_app.py），并在其中初始化Celery应用：

from celery import Celery

# 创建Celery应用实例
app = Celery('my_app', broker='amqp://guest@localhost//')

# 配置Celery
app.conf.update(
    result_backend='rpc://',
    task_serializer='json',
    accept_content=['json'],  # 只接受JSON格式的消息
    result_serializer='json',
    timezone='Asia/Shanghai',
    enable_utc=True,
)

在上面的代码中，我们创建了一个名为my_app的Celery应用实例，并指定了消息代理为RabbitMQ（amqp://guest@localhost//）。此外，我们还配置了一些基本的Celery设置，如任务序列化方式、时区等。

2. 定义任务

接下来，我们需要定义一些任务。任务是通过@app.task装饰器来定义的。例如，定义一个简单的加法任务：

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

3. 启动Celery Worker

要执行任务，需要启动Celery Worker。在终端中运行以下命令：

celery -A celery_app worker --loglevel=info

这将启动一个Celery Worker，它会从消息代理中获取任务并执行。

4. 调用任务

在另一个Python脚本或交互式Python环境中，可以调用定义的任务：

from celery_app import add

# 异步调用任务
result = add.delay(4, 6)

# 获取任务结果
print(result.get())

delay方法是apply_async的快捷方式，它会将任务放入队列中并立即返回一个AsyncResult对象。通过result.get()可以获取任务的执行结果。

配置Celery

Celery提供了丰富的配置选项，可以通过app.conf.update方法来更新配置。以下是一些常用的配置项：

• broker_url: 指定消息代理的URL。
• result_backend: 指定任务结果的存储后端。
• task_serializer: 指定任务的序列化方式。
• result_serializer: 指定任务结果的序列化方式。
• timezone: 指定时区。
• enable_utc: 是否启用UTC时间。

例如，配置Redis作为消息代理和结果后端：

app.conf.update(
    broker_url='redis://localhost:6379/0',
    result_backend='redis://localhost:6379/0',
    task_serializer='json',
    result_serializer='json',
    timezone='Asia/Shanghai',
    enable_utc=True,
)

任务调度

Celery不仅支持异步任务处理，还支持任务调度。可以通过celery beat来实现定时任务。

1. 配置定时任务

在celery_app.py中，添加定时任务配置：

from celery.schedules import crontab

app.conf.beat_schedule = {
    'add-every-30-seconds': {
        'task': 'celery_app.add',
        'schedule': 30.0,
        'args': (16, 16),
    },
    'add-every-minute': {
        'task': 'celery_app.add',
        'schedule': crontab(minute='*/1'),
        'args': (10, 10),
    },
}

在上面的配置中，我们定义了两个定时任务：一个是每30秒执行一次add任务，另一个是每分钟执行一次add任务。

2. 启动Celery Beat

要启动定时任务调度器，需要运行以下命令：

celery -A celery_app beat --loglevel=info

这将启动Celery Beat，它会根据配置的调度计划定期触发任务。

任务重试

在实际应用中，任务可能会因为各种原因失败。Celery提供了任务重试机制，可以在任务失败时自动重试。

1. 配置任务重试

在定义任务时，可以通过autoretry_for和retry_kwargs参数来配置任务重试：

@app.task(autoretry_for=(Exception,), retry_kwargs={'max_retries': 3, 'countdown': 5})
def add(x, y):
    return x + y

在上面的代码中，我们配置了任务在遇到任何异常时自动重试，最多重试3次，每次重试间隔5秒。

2. 手动重试

也可以在任务中手动调用retry方法来实现重试：

@app.task(bind=True)
def add(self, x, y):
    try:
        return x + y
    except Exception as exc:
        raise self.retry(exc=exc, countdown=5)

在上面的代码中，任务在遇到异常时会手动调用retry方法进行重试。

任务链

Celery支持将多个任务串联起来，形成一个任务链。任务链中的每个任务都会在前一个任务成功完成后执行。

1. 创建任务链

可以通过chain方法创建任务链：

from celery import chain

result = chain(add.s(4, 6), add.s(10)).apply_async()
print(result.get())

在上面的代码中，我们创建了一个任务链，首先执行add(4, 6)，然后将结果传递给add(10)。

2. 任务组

Celery还支持将多个任务并行执行，形成一个任务组。可以通过group方法创建任务组：

from celery import group

result = group(add.s(i, i) for i in range(10)).apply_async()
print(result.get())

在上面的代码中，我们创建了一个任务组，并行执行10个add任务。

任务结果

Celery提供了多种方式来存储任务结果，如Redis、RPC、数据库等。可以通过result_backend配置项来指定任务结果的存储后端。

1. 获取任务结果

可以通过AsyncResult对象来获取任务结果：

from celery.result import AsyncResult

result = add.delay(4, 6)
async_result = AsyncResult(result.id, app=app)
print(async_result.get())

2. 任务状态

可以通过AsyncResult对象获取任务的状态：

print(async_result.status)

任务状态可以是PENDING、STARTED、SUCCESS、FLURE等。

总结

本文详细介绍了如何创建和使用Celery，包括安装、初始化、定义任务、配置、任务调度、任务重试、任务链和任务结果等方面。Celery是一个功能强大的异步任务队列系统，适用于各种复杂的任务处理场景。通过本文的学习，开发者可以快速上手Celery，并在实际项目中应用它来提高应用的性能和用户体验。