当前位置：首页 > news >正文

在 Dify 项目中的 Celery：异步任务的实现与集成

news 来源：原创 2025/6/5 14:39:06

Celery 是一个强大而灵活的分布式任务队列系统，旨在帮助应用程序在后台异步运行耗时的任务，提高系统的响应速度和性能。在 Dify 项目中，Celery 被广泛用于处理异步任务和定时任务，并与其他工具（如 Sentry、OpenTelemetry）集成，实现了任务的监控和追踪。

本文将详细介绍 Celery 在 Dify 项目中的应用，包括其集成方式、使用方法，以及涉及的关键文件和代码片段。

概述

Dify 项目采用 Flask 作为 Web 框架，为了提升系统性能和用户体验，引入了 Celery 来处理耗时的后台任务。通过将任务分配到不同的队列，并使用 Celery 的 Worker 进行异步执行，Dify 实现了任务的解耦和并发处理。同时，项目还集成了 Sentry 和 OpenTelemetry，对任务执行进行实时监控和性能追踪。

Celery 的集成与配置

1. 应用工厂 `app_factory.py`

文件路径：

./api/app_factory.py

内容分析：

# 将 Celery 扩展导入应用工厂
ext_celery,

在 Flask 应用工厂中，Celery 被作为扩展（ext_celery）引入。这意味着在创建 Flask 应用实例时，Celery 也会被初始化并与应用集成。这种方式常用于 Flask 框架，便于统一管理应用的各个部分。

2. Celery 扩展 `ext_celery.py`

文件路径：

./api/extensions/ext_celery.py

内容分析：

from celery import Celery, Task
from celery.schedules import crontabdef init_celery_app(app):celery_app = Celery(app.import_name)# 更新 Celery 配置celery_app.conf.update(broker_url=app.config['CELERY_BROKER_URL'],result_backend=app.config['CELERY_RESULT_BACKEND'],task_serializer='json',accept_content=['json'],timezone='UTC',enable_utc=True,)# 注册定时任务和导入任务模块celery_app.conf.update(beat_schedule=beat_schedule,imports=imports,)# 将 Celery 实例注册到 Flask 应用app.extensions["celery"] = celery_appreturn celery_app

解释：

初始化 Celery 应用实例：使用 Celery(app.import_name) 创建 Celery 实例。
配置更新：通过 celery_app.conf.update() 设置消息代理、结果后端、任务序列化等配置。
注册到应用扩展：将 Celery 实例添加到 Flask 应用的扩展中，便于全局访问和使用。

3. 配置文件 `.env` 和 `pyproject.toml`

文件路径：

./docker/.env
./api/pyproject.toml

内容分析：

# .env 配置
# 使用 Redis 作为 Celery 的消息代理，数据库索引为 1
CELERY_BROKER_URL=redis://localhost:6379/1
CELERY_RESULT_BACKEND=redis://localhost:6379/1

# pyproject.toml 中的依赖
"celery~=5.5.2",
"opentelemetry-instrumentation-celery==0.48b0",

解释：

消息代理与结果后端：在 .env 文件中，指定了 Celery 使用 Redis 作为消息代理和结果后端。
依赖管理：在 pyproject.toml 中，明确了项目对 Celery 及其相关监控工具的依赖，确保了环境的一致性。

任务的定义与处理

1. 任务定义文件 `tasks/`

文件路径：

./api/tasks/

内容分析：

from celery import shared_task@shared_task
def some_task(args):# 任务的具体实现pass

解释：

使用 shared_task 装饰器：在任务模块中，使用 @shared_task 装饰器定义任务。这种方式无需直接引用 Celery 应用实例，避免了循环导入的问题，提高了模块的独立性和可重用性。
任务类型：任务包括数据处理、邮件发送、操作追踪等，满足项目的多种业务需求。

2. 定时任务 `schedule/`

文件路径：

./api/schedule/

内容分析：

@app.celery.task(queue="dataset")
def scheduled_task():# 定时任务的逻辑实现pass

解释：

使用 @app.celery.task 装饰器：在定时任务中，直接引用了 Celery 应用实例，便于指定任务的队列和其他配置。
任务调度：定时任务通过 Celery Beat 进行调度，按照预定义的时间间隔自动执行。

3. 操作追踪管理器 `ops_trace_manager.py`

文件路径：

./api/core/ops/ops_trace_manager.py

内容分析：

def send_to_celery(self, tasks: list[TraceTask]):# 将任务发送到 Celery 队列进行异步处理pass

解释：

异步处理任务：定义了方法，将追踪任务发送到 Celery，利用其异步处理能力，提高系统的性能。

Celery 的启动与运行

1. 启动脚本 `entrypoint.sh`

文件路径：

./api/docker/entrypoint.sh

内容分析：

# 启动 Celery Worker
exec celery -A app.celery worker -P ${CELERY_WORKER_CLASS:-gevent} $CONCURRENCY_OPTION \--loglevel ${LOG_LEVEL:-INFO} --queues ${CELERY_QUEUES:-default}# 启动 Celery Beat（定时任务调度器）
exec celery -A app.celery beat --loglevel ${LOG_LEVEL:-INFO}

解释：

启动 Celery Worker：使用 celery -A app.celery worker 命令，指定应用实例为 app.celery，并使用 gevent 并发池，提高异步任务的执行效率。
启动 Celery Beat：使用 celery -A app.celery beat 命令，启动定时任务调度器，按计划执行定时任务。

2. 启动命令示例 `README.md`

文件路径：

./api/README.md

内容分析：

uv run celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion

解释：

指定任务队列：使用 -Q 参数，指定 Worker 监听的任务队列，如 dataset、generation、mail 等，实现任务的分类处理和资源优化。
设置并发和日志：-P gevent 指定并发池，-c 1 设置并发数量，--loglevel INFO 设置日志级别为 INFO，便于监控任务执行情况。

与其他工具的集成

1. Sentry 集成 `ext_sentry.py`

文件路径：

./api/extensions/ext_sentry.py

内容分析：

from sentry_sdk.integrations.celery import CeleryIntegrationdef init_sentry():sentry_sdk.init(dsn="your_sentry_dsn",integrations=[CeleryIntegration()],# 其他配置)

解释：

错误监控：通过集成 Sentry，Celery 任务执行中的异常将被捕获并发送到 Sentry，方便开发者及时发现和解决问题。

2. OpenTelemetry 集成 `ext_otel.py`

文件路径：

./api/extensions/ext_otel.py

内容分析：

from opentelemetry.instrumentation.celery import CeleryInstrumentordef init_tracing():# 判断是否为 Celery Worker 进程if not is_celery_worker():CeleryInstrumentor().instrument()else:# 在 Celery Worker 初始化时进行 Instrumentationworker_init.connect(init_celery_worker)def init_celery_worker(*args, **kwargs):CeleryInstrumentor().instrument()

解释：

性能监控：通过 OpenTelemetry，对 Celery 任务的执行进行性能监控和分布式追踪，助力分析系统的瓶颈和优化方向。

数据库支持与依赖管理

1. 数据库迁移 `64b051264f32_init.py`

文件路径：

./api/migrations/versions/64b051264f32_init.py

内容分析：

op.create_table('celery_taskmeta', ...)
op.create_table('celery_tasksetmeta', ...)# 删除表
op.drop_table('celery_tasksetmeta')
op.drop_table('celery_taskmeta')

解释：

任务状态存储：通过 Alembic 迁移，创建 Celery 用于存储任务元数据和结果的数据库表，实现任务状态的持久化管理。

2. 依赖管理 `uv.lock`

文件路径：

./api/uv.lock

内容分析：

name = "celery"
version = "5.5.2"
...name = "opentelemetry-instrumentation-celery"
version = "0.48b0"
...

解释：

锁定依赖：uv.lock 文件记录了项目的依赖库和版本信息，确保了在不同环境下安装一致的依赖，防止版本冲突。

开发与调试支持

1. VSCode 调试配置 `launch.json.example`

文件路径：

./api/.vscode/launch.json.example

内容分析：

{"name": "Celery Worker","type": "python","request": "launch","module": "celery","args": ["worker","-A","app.celery","--loglevel=INFO"],"console": "integratedTerminal"
}

解释：

调试支持：提供了 VSCode 的调试配置，方便开发人员在 IDE 中对 Celery Worker 进行调试和测试。

2. 开发脚本 `start-worker`

文件路径：

./dev/start-worker

内容分析：

#!/bin/bash
# 启动开发环境下的 Celery Worker
celery -A app.celery worker --loglevel=INFO

解释：

快速启动：提供了脚本，简化了开发环境下 Celery Worker 的启动命令，提高了开发效率。

实践建议

任务队列划分：根据任务的性质和资源需求，将任务分配到不同的队列，合理配置 Worker，提高系统的性能和可靠性。
监控与日志：利用 Sentry 和 OpenTelemetry，对任务的执行状态和性能进行实时监控，及时发现潜在问题。
资源优化：根据任务类型（I/O 密集型或 CPU 密集型），选择合适的并发模式（如 gevent、eventlet 或 prefork），优化资源利用率。
依赖管理：通过 pyproject.toml 和 uv.lock 等文件，明确项目的依赖版本，确保环境的一致性。
安全性考虑：在配置文件和环境变量中，注意保护敏感信息，如数据库连接字符串和 Sentry 的 DSN，避免泄漏。