[异步任务处理]Celery + RabbitMQ

发表于 2025-04-04 更新于 2025-06-28 分类于工具/tool 阅读次数：

本文字数： 15k 阅读时长 ≈ 28 分钟

Celery 是一个使用 Python 开发的分布式任务队列框架，广泛用于处理异步任务和定时任务。它通过消息中间件（Broker）将耗时或复杂的任务分发给一个或多个工作节点（Worker）进行异步执行，适用于高并发、任务解耦、批量数据处理、后台服务等场景。

Celery解析

Celery/RabbitMQ安装

# Install RabbitMQ
$ sudo apt-get install rabbitmq-server

# Install Celery
$ pip3 install celery -i https://pypi.tuna.tsinghua.edu.cn/simple

工作流程

任务提交：应用调用 task.delay(args) 方法，将任务发送到 Broker；
任务入队：Broker 接收任务并放入队列中等待处理；
任务消费：Worker 监听到队列中有新任务，取出并执行；
结果返回（可选）：若配置了 Result Backend，则任务执行结果会被写入指定存储；
结果查询（可选）：生产者可通过任务 ID 查询执行状态和结果。

注意：在 Celery 中，一个 Worker 实例通常会管理多个进程或线程，这意味着它可以同时执行多个任务。

关键组件

Celery 的核心架构由以下几个关键组件组成：

Producer（生产者）

也称为客户端（Client），是任务的发起方。通常是一个 Web 应用（如 Django 或 Flask），当有需要异步处理的操作时，它会向 Broker 发送一个任务消息。

Broker（消息中间件）

负责接收来自生产者的任务消息，并将其排队等待 Worker 处理。常用的 Broker 包括：

RabbitMQ：功能强大、稳定性高，适合生产环境。

Redis：性能优异，支持发布/订阅机制，也常用于缓存。

其他支持的包括 Amazon SQS、ZooKeeper、MongoDB 等。

Worker（消费者）

Worker 是运行在独立进程或机器上的任务执行单元。它们持续监听 Broker 中的任务队列，一旦发现新任务，就会取出并执行。一个系统中可以部署多个 Worker，实现横向扩展，提升任务处理能力。

Task（任务）

任务是 Celery 执行的基本单位，通常是一个带有 @app.task 装饰器的 Python 函数。它可以接受参数，返回结果，并支持重试、延迟、优先级等功能。

Result Backend（结果后端，可选）

如果需要获取任务执行的结果，可以通过配置 Result Backend 来存储任务状态和返回值。常用方案包括 Redis、RabbitMQ（RPC 模式）、数据库等。

QuickStart

Introduction to Celery

完整的 Celery 异步任务处理流程，包括：

定义 Celery 实例与异步任务
启动 Celery Worker 监听队列
调用异步任务并获取结果

定义 Celery 实例与异步任务

# celery_app.py
from celery import Celery

# 初始化 Celery 应用实例
# broker: 使用 RabbitMQ 作为消息代理
# result_backend: 可选，用于存储任务执行结果（这里使用 RPC 返回）
app = Celery(
    'tasks',
    broker='amqp://guest@localhost//',         # RabbitMQ 默认连接地址
    backend='rpc://'                            # 使用 RabbitMQ 的 RPC 方式返回任务结果
)

# 示例任务 1：无参数打印 Hello
@app.task
def print_hello():
    print("hello Celery")

# 示例任务 2：两个数字相乘
@app.task
def add_numbers(num1, num2):
    """
    接收两个数字，返回它们的乘积。
    """
    return num1 * num2

# 示例任务 3：接收多个数字列表，返回总和
@app.task
def add_numbers2(**kwargs):
    """
    接收关键字参数中的 numbers 列表，返回其总和。
    
    示例调用：
        add_numbers2.delay(numbers=[10, 20, 30])
    """
    assert 'numbers' in kwargs.keys(), "必须传入 key 为 'numbers' 的参数"
    return sum(kwargs['numbers'])

注意：@app.task必须放在函数定义的最上层装饰器位置

启动 Celery Worker 监听队列

1	$ celery -A celery_app worker --loglevel=info

-A celery_app：指定 Celery 实例所在的模块名（即 celery_app.py）
worker：启动一个 Worker 进程来消费任务
--loglevel=info：设置日志级别为 info

调用异步任务并获取结果

celery.app.task.Task.delay

# run_task.py
from celery_worker import print_hello, add_numbers, add_numbers2

# 测试 1: print_hello()
print("=== 测试 print_hello ===")
result_hello = print_hello.delay()
print("任务ID:", result_hello.id)
print("任务结果:", result_hello.get(timeout=10))  # 如果没有返回值，结果为 None

# 测试 2: add_numbers(num1, num2)
print("\n=== 测试 add_numbers ===")
numbers_add = [10, 20]
result_add = add_numbers.delay(*numbers_add)  # 解包传参
print("任务ID:", result_add.id)
print("任务结果:", result_add.get(timeout=10))

# 测试 3: add_numbers2(**kwargs)
print("\n=== 测试 add_numbers2 ===")
result_add2 = add_numbers2.delay(numbers=[5, 15, 30])  # 以字典形式传递参数
print("任务ID:", result_add2.id)
print("任务结果:", result_add2.get(timeout=10))

执行测试

启动Celery，然后运行测试程序，执行日志如下：

$ celery -A celery_app worker --loglevel=info

 -------------- celery@LAPTOP-S3BIPLGN v5.5.3 (immunity)
--- ***** -----
-- ******* ---- Linux-6.6.87.2-microsoft-standard-WSL2-x86_64-with-glibc2.35 2025-06-28 21:38:42
- *** --- * ---
- ** ---------- [config]
- ** ---------- .> app:         tasks:0x76c43b4f90d0
- ** ---------- .> transport:   amqp://guest:**@localhost:5672//
- ** ---------- .> results:     rpc://
- *** --- * --- .> concurrency: 22 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** -----
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery


[tasks]
  . celery_app.add_numbers
  . celery_app.add_numbers2
  . celery_app.print_hello

[2025-06-28 21:38:42,429: INFO/MainProcess] Connected to amqp://guest:**@127.0.0.1:5672//
[2025-06-28 21:38:42,432: INFO/MainProcess] mingle: searching for neighbors
[2025-06-28 21:38:43,455: INFO/MainProcess] mingle: all alone
[2025-06-28 21:38:43,468: INFO/MainProcess] celery@LAPTOP-S3BIPLGN ready.
[2025-06-28 21:39:14,508: INFO/MainProcess] Task celery_app.print_hello[cfa60a5a-ebb1-40d1-93e1-288f9000c2db] received
[2025-06-28 21:39:14,509: WARNING/ForkPoolWorker-16] hello Celery
[2025-06-28 21:39:14,516: INFO/ForkPoolWorker-16] Task celery_app.print_hello[cfa60a5a-ebb1-40d1-93e1-288f9000c2db] succeeded in 0.006867269999929704s: None
[2025-06-28 21:39:14,518: INFO/MainProcess] Task celery_app.add_numbers[da6053cb-90bf-4343-953e-43c29362795a] received
[2025-06-28 21:39:14,519: INFO/ForkPoolWorker-16] Task celery_app.add_numbers[da6053cb-90bf-4343-953e-43c29362795a] succeeded in 0.0003271770001447294s: 200
[2025-06-28 21:39:14,521: INFO/MainProcess] Task celery_app.add_numbers2[7e828bbb-3026-415c-bf1c-c15570be5176] received
[2025-06-28 21:39:14,522: INFO/ForkPoolWorker-16] Task celery_app.add_numbers2[7e828bbb-3026-415c-bf1c-c15570be5176] succeeded in 0.00023004600006970577s: 50

$ python3 run_task.py 
=== 测试 print_hello ===
任务ID: cfa60a5a-ebb1-40d1-93e1-288f9000c2db
任务结果: None

=== 测试 add_numbers ===
任务ID: da6053cb-90bf-4343-953e-43c29362795a
任务结果: 200

=== 测试 add_numbers2 ===
任务ID: 7e828bbb-3026-415c-bf1c-c15570be5176
任务结果: 50

高级设置

delay vs. apply_async

1 2	delay(args, kwargs) apply_async(args=None, kwargs=None, task_id=None, producer=None, link=None, link_error=None, shadow=None, *options)[source]

delay是apply_async的简易版本，比较apply_async会少一些输入选项。它们的目的都是为了发送消息给异步应用任务

def delay(self, *args, **kwargs):
    """Star argument version of :meth:`apply_async`.

    Does not support the extra options enabled by :meth:`apply_async`.

    Arguments:
        *args (Any): Positional arguments passed on to the task.
        **kwargs (Any): Keyword arguments passed on to the task.
    Returns:
        celery.result.AsyncResult: Future promise.
    """
    return self.apply_async(args, kwargs)

配置文件

Celery 支持通过加载配置文件的方式来定义各种配置选项。示例如下：

# celery_config.py

# 指定消息代理（Broker）的地址，这里使用的是 RabbitMQ
# 格式为：amqp://<用户名>:<密码>@<主机>/<虚拟主机>
broker_url = 'amqp://guest@localhost//'

# 指定结果后端（Backend）的地址，这里也使用 RabbitMQ 的 RPC 协议
result_backend = 'rpc://'

# 配置任务的序列化方式，推荐使用 JSON 格式
task_serializer = 'json'

# 指定接受的内容类型，确保只处理 JSON 格式的任务
accept_content = ['json']

# 配置任务执行结果的序列化方式，保持与任务一致
result_serializer = 'json'

# 时区设置，默认为 UTC，可以根据项目需求调整
# 注意：如果启用时区支持，需同时设置 enable_utc 为 True
timezone = 'Asia/Shanghai'  # 示例：设置为中国上海时区
enable_utc = False  # 如果不希望强制使用 UTC 时间，可以设置为 False

在 Celery 实例初始化时，可以通过 config_from_object 方法加载上述配置文件：

# celery_app.py
from celery import Celery

app = Celery("tasks")
app.config_from_object('celery_config')

多文件定义

Celery允许在不同文件中定义各自的异步任务。示例如下：

主文件：celery_app.py

# celery_app.py
from celery import Celery

def make_celery_instance():
    """
    创建并配置 Celery 实例
    """
    celery_instance = Celery("tasks")
    celery_instance.conf.update(
        include=[
            'celery_print_hello',
            'celery_add_numbers',
            'celery_add_numbers2'
        ]  # 确保路径正确指向你的任务文件
    )
    celery_instance.config_from_object("celery_config")
    return celery_instance


# Celery 实例
celery_instance = make_celery_instance()

任务文件 1：celery_print_hello.py

# celery_print_hello.py
from celery_app import celery_instance

@app.task(name='print_hello')
def print_hello():
    """
    示例任务 1：无参数打印 Hello
    """
    print("hello Celery")

任务文件 2：celery_add_numbers.py

# celery_add_numbers.py
from celery_app import celery_instance

@app.task(name='add_numbers')
def add_numbers(num1, num2):
    """
    示例任务 2：两个数字相乘
    
    参数:
    num1 (int): 第一个数字
    num2 (int): 第二个数字
    
    返回:
    int: 两数相乘的结果
    """
    return num1 * num2

任务文件 3：celery_add_numbers2.py

# celery_add_numbers2.py
from celery_app import celery_instance

@app.task(name='add_numbers2')
def add_numbers2(**kwargs):
    """
    示例任务 3：接收多个数字列表，返回总和
    
    参数:
    kwargs (dict): 包含关键字参数的字典，必须包含 key 为 'numbers' 的列表
    
    返回:
    int: 数字列表的总和
    
    示例调用：
        add_numbers2.delay(numbers=[10, 20, 30])
    """
    assert 'numbers' in kwargs.keys(), "必须传入 key 为 'numbers' 的参数"
    return sum(kwargs['numbers'])

💡 注意事项：

在每个任务文件中导入 celery_instance，并在任务函数上应用 @app.task 装饰器。
确保所有任务文件的路径正确，且被包含在 celery_app.py 中的 include 列表中。
使用 name 参数为任务指定名称，以便更好地管理和监控任务。

Celery Hook

Celery 提供了丰富的钩子（Hook）机制，允许开发者在任务执行生命周期中的关键节点插入自定义逻辑。常用钩子有：

worker_process_init

触发时机：当每个 Worker 的子进程被创建并启动时调用
用途：适合用于数据库连接池初始化、缓存客户端连接等

from celery import signals

@signals.worker_process_init.connect
def on_worker_init(**kwargs):
    print("Worker 子进程已启动，可以在此初始化资源，如数据库连接")

worker_shutting_down

触发时机：Worker 即将关闭时调用
用途：用于释放资源、保存状态、关闭连接等

1
2
3

@signals.worker_shutting_down.connect
def on_worker_shutdown(**kwargs):
    print("Worker 正在关闭，可以在此释放资源")

task_prerun

触发时机：任务开始执行之前
用途：适合用于记录任务启动时间、验证输入参数、初始化上下文等

from celery import signals

@signals.task_prerun.connect
def task_prerun_handler(task_id, task, args, kwargs, **kw):
    print(f"【Pre-run】Task {task.name} ({task_id}) 正在启动，参数: {args}, {kwargs}")

task_postrun

触发时机：任务执行完成后
用途：适合用于清理资源、记录任务结束时间、处理结果等

1
2
3

@signals.task_postrun.connect
def task_postrun_handler(task_id, task, retval, state, args, kwargs, **kw):
    print(f"【Post-run】Task {task.name} ({task_id}) 已完成，状态: {state}, 返回值: {retval}")

测试代码如下：

# celery_app.py
from celery import Celery, signals

app = Celery('tasks', broker='amqp://guest@localhost//')


# 钩子函数：Worker 子进程初始化
@signals.worker_process_init.connect
def on_worker_init(**kwargs):
    print("【Hook】Worker 子进程已启动，正在初始化资源...")


# 钩子函数：Worker 正在关闭
@signals.worker_shutting_down.connect
def on_worker_shutdown(**kwargs):
    print("【Hook】Worker 正在关闭，正在释放资源...")


# 钩子函数：任务开始执行前
@signals.task_prerun.connect
def task_prerun_handler(task_id, task, args, kwargs, **kw):
    print(f"【Pre-run】Task {task.name} ({task_id}) 正在启动，参数: {args}, {kwargs}")


# 钩子函数：任务执行完成后
@signals.task_postrun.connect
def task_postrun_handler(task_id, task, retval, state, args, kwargs, **kw):
    print(f"【Post-run】Task {task.name} ({task_id}) 已完成，状态: {state}, 返回值: {retval}")


# 示例任务
@app.task
def add(x, y):
    return x + y


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

启动 Worker：

1	celery -A celery_app worker --loglevel=info