大家好！我是爱摸鱼的小鸿，关注我，收看每期的编程干货。

一个简单的库，也许能够开启我们的智慧之门，
一个普通的方法，也许能在危急时刻挽救我们于水深火热，
一个新颖的思维方式，也许能激发我们无尽的创造力，
一个独特的技巧，也许能成为我们的隐形盾牌……

神奇的 Python 库之旅，第 9 章

一、什么是 Celery？

Celery 是一个强大的工具，它能够帮助我们管理和调度复杂的任务。无论是处理异步任务、计划任务，还是分布式任务，Celery 都能轻松胜任。在这篇文章中，我们将深入探讨 Celery 的魅力，通过多个代码示例，带你全面了解这个神器。

Celery 是一个简单、灵活且可靠的分布式系统，用于处理大量消息，并提供了维护这些任务的工具。它的使用场景非常广泛，比如电子邮件发送、视频处理、数据分析、Web 爬虫等。

Github 项目地址：

https://github.com/celery/celery

…

二、为什么选择 Celery？

选择 Celery 的理由有很多，这里列出几个主要的优势：

• 异步任务处理：能让你在不阻塞主进程的情况下处理任务；
• 定时任务：支持类似 cron 的定时任务调度；
• 分布式执行：支持将任务分发到多个机器上执行，提升系统性能；
• 高可用性：可以与消息队列（如 RabbitMQ, Redis）结合，实现高可用性和可靠性。

在开始之前，我们需要安装 Celery 和一个消息代理（这里我们选择 Redis）：

pip install celery redis

…

三、Celery 编程示例

快速开始：一个简单的任务
让我们从一个简单的例子开始，创建一个任务来演示 Celery 的基本用法。

# tasks.py
from celery import Celery

# 创建 Celery 实例
app = Celery('tasks', broker='redis://localhost:6379/0')

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

在这个例子中，我们定义了一个名为 add 的任务，它接收两个参数并返回它们的和。接下来，我们可以启动一个 Celery worker 来处理这个任务：

celery -A tasks worker --loglevel=info

启动 Celery worker 后，我们可以在 Python 交互式环境中调用这个任务：

>>> from tasks import add
>>> result = add.delay(4, 6)
>>> result.get(timeout=10)
10

深入挖掘：任务链和组
Celery 的强大之处在于它能够处理复杂的工作流，比如任务链和任务组。

任务链
任务链允许我们将多个任务串联起来，前一个任务的输出作为下一个任务的输入：

from celery import chain

# 定义任务
@app.task
def multiply(x, y):
    return x * y

@app.task
def add_and_multiply(x, y, z):
    return chain(add.s(x, y), multiply.s(z))()

# 调用任务链
result = add_and_multiply(2, 3, 4)
print(result.get())  # 输出 20，因为 (2 + 3) * 4 = 20

任务组
任务组允许我们并行执行一组任务，并在所有任务完成后获得结果：

from celery import group

# 定义任务组
@app.task
def sum_list(numbers):
    return sum(numbers)

@app.task
def process_groups():
    return group(sum_list.s([1, 2, 3]), sum_list.s([4, 5, 6]), sum_list.s([7, 8, 9]))().get()

# 调用任务组
result = process_groups()
print(result)  # 输出 [6, 15, 24]

…

定时任务
Celery 还支持定时任务，这类似于 Unix 系统的 cron 作业：

from celery import Celery
from celery.schedules import crontab

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

@app.task
def scheduled_task():
    print("This task runs every 10 seconds")

app.conf.beat_schedule = {
    'run-every-10-seconds': {
        'task': 'periodic_tasks.scheduled_task',
        'schedule': 10.0,
    },
}

# 启动 Celery beat 进程
celery -A periodic_tasks beat --loglevel=info

…

错误处理与重试机制
在实际应用中，任务失败是不可避免的。Celery 提供了优雅的错误处理和重试机制：

@app.task(bind=True, max_retries=3)
def unreliable_task(self):
    try:
        # 可能失败的操作
        risky_operation()
    except Exception as exc:
        # 捕获异常并重试
        raise self.retry(exc=exc, countdown=60)

在这个例子中，unreliable_task 如果失败，会在 60 秒后重试，总共重试 3 次。


使用 Celery 在 Web 应用
Celery 常用于 Web 应用中来处理后台任务。下面是一个简单的 Flask 应用，展示了如何集成 Celery：

from flask import Flask, request, jsonify
from celery import Celery

app = Flask(__name__)
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

def make_celery(app):
    celery = Celery(app.import_name, broker=app.config['CELERY_BROKER_URL'])
    celery.conf.update(app.config)
    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)
    celery.Task = ContextTask
    return celery

celery = make_celery(app)

@celery.task
def long_task():
    import time
    time.sleep(10)
    return 'Task completed!'

@app.route('/start-task', methods=['POST'])
def start_task():
    task = long_task.apply_async()
    return jsonify({}), 202, {'Location': url_for('get_status', task_id=task.id)}

@app.route('/status/<task_id>')
def get_status(task_id):
    task = long_task.AsyncResult(task_id)
    response = {
        'state': task.state,
        'result': task.result,
    }
    return jsonify(response)

if __name__ == '__main__':
    app.run(debug=True)

在这个示例中，我们定义了一个 Flask 应用，并将 Celery 集成到其中。long_task 是一个模拟长时间运行的任务，客户端可以通过 start-task 路由启动任务，并通过 status/<task_id> 路由查询任务状态。

更多功能、详细用法可参考官方文档：

https://docs.celeryq.dev/en/stable

…

四、总结

Celery 是一个强大的工具，它为我们处理异步任务、计划任务和分布式任务提供了极大的便利。在这篇文章中，我们通过多个代码示例，展示了 Celery 的基本用法、任务链和组、定时任务、错误处理和重试机制，以及如何在 Web 应用中集成 Celery。

如果你在日常工作中需要处理复杂的任务调度和分布式任务，那么 Celery 绝对是一个值得深入学习和使用的工具。希望这篇文章能够帮助你更好地理解和使用 Celery，让你的工作更加高效、顺畅。

试一试吧，Celery 会让你的任务调度变得简单又高效！

…

五、作者Info

Author：小鸿的摸鱼日常

Goal：让编程更有趣！ 专注于 Web 开发、爬虫，游戏开发，数据分析、自然语言处理，AI 等，期待你的关注，让我们一起成长、一起Coding！

版权说明：本文禁止抄袭、转载，侵权必究！