Celery，一个实时处理的 Python 分布式系统

🕗 发布于 2024-07-10 04:10 Celery Python 库之旅 python

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

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

神奇的 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！

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

原文地址：https://blog.csdn.net/qq_44000141/article/details/140307508

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：《大语言模型的临床和外科应用：系统综述》
下一篇：GET请求和POST请求的区别

Spring Data JPA
这个接口的定义是在数据持久层（DAO 层，Repository 层），主要用于自定义一些查询方法（这是这个接口最主要的作用）。接口内部可以通过命名规则定义自定义查询方法，以便在业务逻辑层调用。（1.3
阅读更多2024-10-07
【JVM】深入解析 Java 虚拟机：内存区域、类加载与垃圾回收机制
本文介绍了 JVM 的内存区域划分、类加载过程及垃圾回收机制。内存区域包括程序计数器、堆、栈和元数据区，每个区域存储不同类型的数据。类加载过程涉及加载、验证、准备、解析和初始化五个步骤。垃圾回收机制主
阅读更多2024-10-07
鸿蒙开发（NEXT/API 12）【穿戴设备传感器获取】手机侧应用开发
手机侧应用可以通过Wear Engine获取穿戴设备上的传感器信息，并通过打开、关闭命令控制获取传感器数据。
阅读更多2024-10-07
c++剪枝
被剪枝支配的恐惧。QWQ
阅读更多2024-10-07
定时器实验(Proteus 与Keil uVision联合仿真）
一、T0工作在方式1，应使TMOD寄存器的M1、M0=01；应设置C/T*=0，为定时器模式；对T0的运行控制仅由TR0来控制，应使相应的GATE位为0。定时器T1不使用，各相关位均设为0。所以，TM
阅读更多2024-10-07
使用Python实现文本到语音转换（TTS）：打造高效易用的TTS应用
本文详细介绍了如何使用Python中的pyttsx3和gTTS库实现文本到语音转换（TTS）。我们讨论了两个库的安装、使用方法及各自的优缺点，展示了如何生成和播放语音，并通过代码示例构建了一个支持多语
阅读更多2024-10-07
亚马逊、ozon卖家：快速提升新品星级与评论数量的技巧
此外，自养号测评的技术简单易学，且可广泛应用于多个电商平台，如亚马逊、沃尔玛、eBay、Wish、Newegg、速卖通、阿里国际站、Shopee、美客多、敦煌网、Lazada、Temu、乐天、Ozon
阅读更多2024-10-07
《防MAC 地址欺骗攻击》
如果攻击者试图通过伪造多个 MAC 地址进行欺骗攻击，超过数量上限后，新的 MAC 地址将无法被学习，从而阻止攻击。将特定设备的 MAC 地址与交换机端口进行静态绑定，只有绑定的 MAC 地址对应的设
阅读更多2024-10-07
Parade Series - BASE64
Parade Series - BASE64
阅读更多2024-10-07
Stream流
在并行流中，流中的元素可以被分成多个数据块，每个块由不同的线程并行处理。并行流适合大数据量和 CPU 密集型的任务，它能够有效地利用多核 CPU 的计算能力。Stream就如同一个迭代器（Iterat
阅读更多2024-10-07