如何使用`Celery`或`RQ`实现`异步`任务队列，并进行`分布式`调度。

好的，我们现在开始讨论如何使用 Celery 和 RQ 实现异步任务队列，并进行分布式调度。

异步任务队列与分布式调度的概念

在深入研究 Celery 和 RQ 之前，我们先明确异步任务队列和分布式调度的概念。

• 异步任务队列： 传统的同步任务处理方式，客户端发起请求后必须等待服务器完成处理并返回结果。异步任务队列则将耗时的任务放入队列中，由独立的worker进程在后台执行，客户端无需等待，提高了响应速度和系统吞吐量。

• 分布式调度： 将任务分配到不同的服务器或节点上执行，从而提高整体的处理能力和可靠性。分布式调度需要考虑任务的划分、分配、监控和容错等问题。

Celery：强大的分布式任务队列

Celery 是一个功能强大、灵活且可靠的异步任务队列/分布式任务队列。它支持多种消息中间件（如 RabbitMQ、Redis）和结果存储后端，适用于各种规模的应用程序。

1. Celery 的核心组件

• Celery Client： 负责将任务添加到任务队列中。通常在 Web 应用或其他服务中使用。

• Celery Worker： 负责从任务队列中获取任务并执行。可以部署在多台服务器上，实现分布式处理。

• Broker (消息中间件)： 用于存储任务消息的队列。Celery 支持 RabbitMQ、Redis 等多种 Broker。

• Result Backend (结果存储后端)： 用于存储任务执行结果。Celery 支持 Redis、数据库等多种 Backend。

2. Celery 的安装与配置

首先，我们需要安装 Celery 和选择的消息中间件和结果存储后端。这里以 RabbitMQ 作为 Broker，Redis 作为 Result Backend 为例。

pip install celery[redis]  # Celery with Redis support
pip install kombu  # Celery依赖，会自动安装

接下来，创建一个 Celery 配置文件 celeryconfig.py：

# celeryconfig.py
broker_url = 'amqp://guest:guest@localhost//'  # RabbitMQ 连接地址
result_backend = 'redis://localhost:6379/0'  # Redis 连接地址
task_serializer = 'json'
result_serializer = 'json'
accept_content = ['json']
timezone = 'Asia/Shanghai'  # 设置时区
enable_utc = True

3. 定义 Celery 任务

创建一个 tasks.py 文件，定义 Celery 任务：

# tasks.py
from celery import Celery
import time

app = Celery('my_tasks', broker='amqp://guest:guest@localhost//', backend='redis://localhost:6379/0')
app.config_from_object('celeryconfig')

@app.task
def add(x, y):
    """
    一个简单的加法任务
    """
    print(f"Executing add task with x={x}, y={y}")
    time.sleep(5)  # 模拟耗时操作
    return x + y

@app.task
def multiply(x, y):
    """
    一个简单的乘法任务
    """
    print(f"Executing multiply task with x={x}, y={y}")
    time.sleep(3)  # 模拟耗时操作
    return x * y

4. 启动 Celery Worker

在命令行中启动 Celery Worker：

celery -A tasks worker -l info

-A tasks 指定 Celery 应用所在的模块。 -l info 设置日志级别为 INFO。

5. 调用 Celery 任务

在 Python 代码中调用 Celery 任务：

# main.py
from tasks import add, multiply

# 异步调用 add 任务
result_add = add.delay(4, 5)
print(f"Add task ID: {result_add.id}")

# 异步调用 multiply 任务
result_multiply = multiply.delay(2, 3)
print(f"Multiply task ID: {result_multiply.id}")

# 获取 add 任务的结果 (阻塞式)
print(f"Add task result: {result_add.get()}")

# 获取 multiply 任务的结果 (阻塞式)
print(f"Multiply task result: {result_multiply.get()}")

6. 分布式调度

Celery 本身就支持分布式调度，只需要启动多个 Celery Worker，它们会自动从 Broker 中获取任务并执行。可以将 Worker 部署在不同的服务器上，从而实现任务的分布式处理。

# 在 Server 1 上启动 Worker
celery -A tasks worker -l info -n worker1@server1

# 在 Server 2 上启动 Worker
celery -A tasks worker -l info -n worker2@server2

-n worker1@server1 和 -n worker2@server2 用于指定 Worker 的名称，方便管理和监控。

7. Celery 的高级特性

• Task Routing： 可以根据任务的类型将任务路由到不同的 Worker 队列。

• Concurrency： 可以设置每个 Worker 的并发数，控制 Worker 的负载。

• Rate Limiting： 可以限制任务的执行频率，防止 Worker 被压垮。

• Retry： 可以配置任务失败时的重试策略。

• Periodic Tasks (Celery Beat)： 可以定期执行任务。

RQ：简单易用的任务队列

RQ (Redis Queue) 是一个基于 Redis 的简单、轻量级的 Python 任务队列库。它易于上手，适用于小型项目或需要快速实现异步任务队列的场景。

1. RQ 的核心组件

• RQ Worker： 负责从 Redis 队列中获取任务并执行。

• Queue： Redis 队列，用于存储任务。

• Job： 代表一个任务。

• Redis： 用作 Broker 和 Result Backend。

2. RQ 的安装与配置

pip install rq

RQ 默认使用 Redis 作为 Broker 和 Result Backend，因此需要先安装 Redis 并启动 Redis 服务器。

3. 定义 RQ 任务

创建一个 tasks_rq.py 文件，定义 RQ 任务：

# tasks_rq.py
import time

def count_words_at_url(url):
    """
    从 URL 读取内容并统计单词数量
    """
    import requests
    resp = requests.get(url)
    return len(resp.text.split())

def process_data(data):
    """
    模拟数据处理任务
    """
    print(f"Processing data: {data}")
    time.sleep(2) # 模拟耗时操作
    return f"Processed: {data}"

4. 启动 RQ Worker

rq worker

默认情况下，RQ Worker 会监听名为 default 的队列。可以使用 -q 参数指定要监听的队列。

5. 调用 RQ 任务

# main_rq.py
import redis
from rq import Queue

# 连接 Redis
redis_connection = redis.Redis(host='localhost', port=6379, db=0)

# 创建队列
q = Queue(connection=redis_connection)

from tasks_rq import count_words_at_url, process_data

# 将 count_words_at_url 任务添加到队列
job = q.enqueue(count_words_at_url, 'http://nvie.com/')
print(f"Job ID: {job.id}")

# 将 process_data 任务添加到队列
job2 = q.enqueue(process_data, "Sample Data")
print(f"Job ID: {job2.id}")

# 获取任务结果 (需要阻塞等待)
print(f"Job Result: {job.result}")
print(f"Job Result: {job2.result}")

6. 分布式调度

与 Celery 类似，RQ 也支持分布式调度。可以启动多个 RQ Worker，它们会自动从 Redis 队列中获取任务并执行。

# 在 Server 1 上启动 Worker，监听 default 队列
rq worker

# 在 Server 2 上启动 Worker，监听 high 优先级队列
rq worker -q high

7. RQ 的高级特性

• Priority Queues： 可以创建多个优先级不同的队列，将任务分配到不同的队列中。

• Job Dependencies： 可以设置任务之间的依赖关系，确保任务按照指定的顺序执行。

• Failed Job Registry： 可以查看失败的任务，并进行重试或处理。

• Scheduler： 可以定时执行任务。

Celery vs RQ：选择哪个？

特性	Celery	RQ
复杂度	较高	较低
功能丰富度	很高	较低
消息中间件支持	支持多种 Broker (RabbitMQ, Redis, 等)	只支持 Redis
适用场景	大型项目，需要复杂的任务调度和管理功能	小型项目，需要快速实现异步任务队列
分布式调度	强大，支持各种路由和并发控制	简单，通过启动多个 Worker 实现
学习曲线	较陡峭	平缓
监控与管理	功能强大，提供 Flower 等监控工具	相对简单，可以通过 Redis 命令和 API 进行监控

任务调度策略

无论使用 Celery 还是 RQ，都需要考虑任务调度策略，以确保任务能够高效、可靠地执行。

• 优先级调度： 根据任务的重要性设置优先级，让高优先级的任务优先执行。
• 公平调度： 避免某些 Worker 一直处于繁忙状态，而其他 Worker 处于空闲状态。
• 资源感知调度： 根据 Worker 的资源情况（如 CPU、内存）将任务分配到合适的 Worker 上。
• 故障转移： 当 Worker 发生故障时，将任务重新分配到其他 Worker 上。
• 监控与报警： 监控任务的执行状态，及时发现和处理问题。

一些建议

• 选择合适的消息中间件： RabbitMQ 适用于对消息可靠性要求较高的场景，Redis 适用于对性能要求较高的场景。
• 合理配置 Worker 数量： 根据服务器的性能和任务的负载情况，合理配置 Worker 的数量。
• 监控任务队列的状态： 及时发现和处理队列中的异常情况。
• 使用任务路由和优先级队列： 根据任务的类型和重要性，合理分配任务。
• 处理任务执行失败的情况： 设置重试机制或将失败的任务记录下来，进行人工处理。

结束语：技术选型与应用场景

Celery 和 RQ 都是优秀的异步任务队列库，选择哪个取决于项目的具体需求和规模。Celery 功能强大，适用于大型项目；RQ 简单易用，适用于小型项目。希望以上内容能够帮助你理解如何使用 Celery 和 RQ 实现异步任务队列，并进行分布式调度。