RAGFlow 异步消息队列迁移至 Apache Kafka

1. 当前架构分析

1.1 整体架构概览

RAGFlow v0.25.6 使用 Redis (Valkey) Streams 作为异步任务消息代理。 这是一个基于 Redis Stream 消费者组的自研轻量级消息队列系统，无 Celery、RabbitMQ 或 Kafka 等第三方消息中间件。

1.2 核心组件详解

1.2.1 连接层 — rag/utils/redis_conn.py

单例 RedisDB 类封装了 Valkey Python 客户端 (v6.0.2)，提供以下队列操作原语：

1.2.2 消费者 (Worker) — rag/svr/task_executor.py

核心异步 Worker 进程，关键参数：

• 消费者命名: task_executor_{host_id}_{worker_no}
• 并发控制: MAX_CONCURRENT_TASKS=5 (asyncio Semaphore)
• 心跳间隔: 30s，超时时间 120s
• 优先级策略: 先检查 Priority 1 队列，再检查 Priority 0
• 启动恢复: 启动后先重放所有未确认消息
• 死 Worker 清理: 基于 Redis Sorted Set 的心跳检测 + 分布式锁

1.2.3 生产者 (任务提交) — api/db/services/task_service.py

1.3 消息生命周期

1.4 当前架构成熟度评估

2. Kafka 迁移目标与收益

3. 改造技术方案

3.1 总体设计原则

1. 渐进式迁移: 通过抽象层隔离 Kafka 细节，支持 Redis/Kafka 双模式切换
2. 向后兼容: 保留现有 API 接口不变，仅替换底层传输
3. 最小侵入: 对 Task 模型、API 层、业务逻辑层零改动
4. 可观测优先: 所有消息生产/消费均埋点，支持 Prometheus 指标导出

3.2 Kafka Topic 规划

3.3 Producer 层改造

3.3.1 新增消息队列抽象层

新建 rag/utils/message_queue.py，提供统一的 Producer/Consumer 接口，屏蔽底层实现差异：

# 新建: rag/utils/message_queue.py

from abc import ABC, abstractmethod
from typing import Any, Callable
from dataclasses import dataclass
import json

@dataclass
class QueueMessage:
    """统一消息模型"""
    message_id: str
    payload: dict
    topic: str
    partition: int = -1
    offset: int = -1
    timestamp: int = 0

    def ack(self) -> bool:
        """确认消息消费完成"""
        ...

    def nack(self, delay_ms: int = 0) -> bool:
        """拒绝消息 (可延迟重试)"""
        ...

class MessageQueueProducer(ABC):
    """消息生产者抽象"""
    @abstractmethod
    async def send(self, topic: str, key: str, value: dict) -> bool: ...
    @abstractmethod
    async def flush(self) -> None: ...

class MessageQueueConsumer(ABC):
    """消息消费者抽象"""
    @abstractmethod
    async def poll(self, timeout_ms: int = 5000) -> QueueMessage | None: ...
    @abstractmethod
    async def commit(self, message: QueueMessage) -> bool: ...
    @abstractmethod
    async def close(self) -> None: ...

class MessageQueueFactory:
    """工厂方法: 根据配置创建 Redis 或 Kafka 实现"""
    _instance = None

    @classmethod
    def get_producer(cls) -> MessageQueueProducer:
        if settings.MQ_BACKEND == "kafka":
            return KafkaProducer(...)
        return RedisStreamProducer(...)

    @classmethod
    def get_consumer(cls, group_id: str) -> MessageQueueConsumer:
        if settings.MQ_BACKEND == "kafka":
            return KafkaConsumer(group_id, ...)
        return RedisStreamConsumer(group_id, ...)

3.3.2 Producer 调用点改造

所有现有 REDIS_CONN.queue_product(...) 调用替换为统一接口，影响函数：

3.3.3 Kafka Producer 实现

# 新建: rag/utils/kafka_producer.py (简化示意)

import asyncio
from aiokafka import AIOKafkaProducer
from common import settings

class KafkaProducer(MessageQueueProducer):
    def __init__(self):
        self._producer = AIOKafkaProducer(
            bootstrap_servers=settings.KAFKA_BOOTSTRAP_SERVERS,
            key_serializer=lambda k: k.encode('utf-8'),
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            acks='all',                # 等待所有 ISR 副本确认
            compression_type='gzip',   # 压缩传输
            max_in_flight_requests_per_connection=1,  # 保证顺序
            retries=5,
        )

    async def start(self):
        await self._producer.start()

    async def send(self, topic: str, key: str, value: dict) -> bool:
        try:
            result = await self._producer.send_and_wait(topic, key=key, value=value)
            logging.debug(f"Kafka produce OK: {topic}[{result.partition}]@{result.offset}")
            return True
        except Exception as e:
            logging.exception(f"Kafka produce failed: {e}")
            return False

    async def flush(self):
        await self._producer.flush()

3.4 Consumer 层改造

3.4.1 改造 collect() 函数

rag/svr/task_executor.py 的 collect() 函数从 Redis XREADGROUP 模式改造为 Kafka Consumer 模式：

# 改造前后对比 — task_executor.py collect()

# 改造前 (Redis Streams):
async def collect():
    redis_msg = REDIS_CONN.queue_consumer(queue, group, consumer_name)
    if redis_msg:
        msg = redis_msg.get_message()
        task = TaskService.get_task(msg["id"])
        return redis_msg, task

# 改造后 (Kafka):
async def collect():
    msg = await KAFKA_CONSUMER.poll(timeout_ms=5000)
    if msg:
        payload = msg.payload
        task = TaskService.get_task(payload["id"])
        return msg, task  # QueueMessage 替代 RedisMsg

3.4.2 Kafka Consumer 实现

# 新建: rag/utils/kafka_consumer.py (简化示意)

import asyncio
from aiokafka import AIOKafkaConsumer

class KafkaConsumer(MessageQueueConsumer):
    def __init__(self, group_id: str, topics: list[str]):
        self._consumer = AIOKafkaConsumer(
            *topics,
            bootstrap_servers=settings.KAFKA_BOOTSTRAP_SERVERS,
            group_id=group_id,
            key_deserializer=lambda k: k.decode('utf-8'),
            value_deserializer=lambda v: json.loads(v.decode('utf-8')),
            enable_auto_commit=False,           # 手动提交 → 等效 XACK
            auto_offset_reset='earliest',       # 启动策略
            max_poll_records=1,                 # 每次拉取 1 条 → 等效 XREADGROUP count=1
            max_poll_interval_ms=3600000,       # 长任务不触发 rebalance (1h)
            session_timeout_ms=120000,          # 消费者组会话超时 (2m)
        )

    async def start(self):
        await self._consumer.start()

    async def poll(self, timeout_ms: int = 5000) -> QueueMessage | None:
        records = await self._consumer.getmany(timeout_ms=timeout_ms, max_records=1)
        for tp, msgs in records.items():
            for m in msgs:
                return QueueMessage(
                    message_id=f"{tp.partition}:{m.offset}",
                    payload=m.value,
                    topic=tp.topic,
                    partition=tp.partition,
                    offset=m.offset,
                    timestamp=m.timestamp,
                    _consumer_record=m,  # 保存原始记录用于 commit
                )
        return None

    async def commit(self, message: QueueMessage) -> bool:
        """commit offset = Redis XACK 等效操作"""
        try:
            await self._consumer.commit()
            return True
        except Exception as e:
            logging.exception(f"Kafka commit failed: {e}")
            return False

    async def close(self):
        await self._consumer.stop()

3.4.3 多 Topic 消费策略

一个 Consumer Group (task_executor) 同时订阅所有任务 Topic。Kafka 原生支持一个 Consumer 订阅多个 Topic，Partition 级别的并行替代了 Redis 的双队列轮询：

3.5 消息格式设计

保留现有 Task 字典结构不变，仅增加 Kafka 元数据 header：

// Kafka Message Headers
{
  "x-task-type":      "parse",           // 任务类型
  "x-task-id":        "uuid-xxxx",       // 任务 ID
  "x-tenant-id":      "tenant-xxxx",     // 租户 ID
  "x-priority":       "0",               // 优先级
  "x-source":         "document_upload", // 来源
  "x-timestamp":      "1718123456789",   // 生产时间戳
  "x-retry-count":    "0",               // 重试次数
  "x-schema-version": "1"                // 消息格式版本
}

// Kafka Message Value (JSON)
{
  "id": "task-uuid",
  "doc_id": "doc-uuid",
  "from_page": 0,
  "to_page": 12,
  "task_type": "parse",
  "priority": 0,
  "digest": "xxhash64-hex",
  "progress": 0.0,
  "begin_at": "2026-06-17 10:30:00",
  "parser_id": "pdf",
  // ... 其余现有字段
}

3.6 错误处理与死信队列

当前系统无 DLQ 机制。Kafka 迁移后引入标准的三层错误处理：

  Message consumed
       │
       ▼
  ┌─────────────┐
  │ Attempt 1    │── success ──▶ commit() → DONE
  └──────┬──────┘
         │ fail
         ▼
  ┌─────────────┐
  │ Attempt 2    │── success ──▶ commit() → DONE
  └──────┬──────┘
         │ fail
         ▼
  ┌─────────────┐
  │ Attempt 3    │── success ──▶ commit() → DONE
  └──────┬──────┘
         │ fail (exceeds max_retries=3)
         ▼
  ┌──────────────┐
  │ Send to DLQ   │ → ragflow.task.dlq  (保留现场，人工介入)
  └──────────────┘
  │ commit() original offset (避免阻塞)
  

3.7 监控与可观测性

4. 逐文件改造清单

5. Docker 部署变更

5.1 Kafka 服务定义 (docker-compose-base.yml)

# 新增 Kafka 服务 (KRaft 模式，无需 Zookeeper)
kafka:
  image: bitnami/kafka:3.9
  hostname: kafka
  ports:
    - "9092:9092"              # Internal (PLAINTEXT)
    - "${KAFKA_EXT_PORT:-9093}:9093"  # External
  environment:
    # KRaft consensus
    KAFKA_CFG_NODE_ID: 1
    KAFKA_CFG_PROCESS_ROLES: broker,controller
    KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@kafka:9094
    KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
    # Listeners
    KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,EXTERNAL://:9093,CONTROLLER://:9094
    KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,EXTERNAL://${KAFKA_EXT_HOST:-localhost}:${KAFKA_EXT_PORT:-9093}
    KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT
    KAFKA_CFG_INTER_BROKER_LISTENER_NAME: PLAINTEXT
    # Topic defaults
    KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: "false"     # 禁止自动创建 Topic
    KAFKA_CFG_DEFAULT_REPLICATION_FACTOR: 1           # 单节点 = 1
    KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
    # Retention
    KAFKA_CFG_LOG_RETENTION_HOURS: 168                # 7 days default
  volumes:
    - kafka_data:/bitnami/kafka
  networks:
    - ragflow
  restart: unless-stopped
  healthcheck:
    test: ["CMD", "kafka-topics.sh", "--bootstrap-server", "localhost:9092", "--list"]
    interval: 15s
    timeout: 10s
    retries: 30

5.2 Topic 初始化脚本

# 新建: docker/kafka_init.sh
#!/bin/bash
# Kafka Topic 初始化 — 在容器启动前执行

BOOTSTRAP="kafka:9092"
PARTITIONS=${KAFKA_DEFAULT_PARTITIONS:-4}
REPLICATION=${KAFKA_REPLICATION_FACTOR:-1}

declare -A TOPICS=(
  ["ragflow.task.parse"]="8"
  ["ragflow.task.parse.priority"]="4"
  ["ragflow.task.dataflow"]="4"
  ["ragflow.task.graphrag"]="2"
  ["ragflow.task.memory"]="2"
  ["ragflow.task.dlq"]="2"
)

echo "Waiting for Kafka..."
until kafka-topics.sh --bootstrap-server $BOOTSTRAP --list &>/dev/null; do
  sleep 2
done

for topic in "${!TOPICS[@]}"; do
  partitions=${TOPICS[$topic]}
  if kafka-topics.sh --bootstrap-server $BOOTSTRAP --describe --topic "$topic" &>/dev/null; then
    echo "Topic $topic already exists, skipping."
  else
    kafka-topics.sh --bootstrap-server $BOOTSTRAP \
      --create --topic "$topic" \
      --partitions "$partitions" \
      --replication-factor "$REPLICATION"
    echo "Created topic: $topic ($partitions partitions)"
  fi
done

echo "Kafka topic initialization complete."

6. 配置变更

6.1 service_conf.yaml.template 新增配置段

# 新增 Kafka 配置段
kafka:
  bootstrap_servers: '${KAFKA_BOOTSTRAP_SERVERS:-kafka:9092}'
  client_id: 'ragflow-server'
  security_protocol: '${KAFKA_SECURITY_PROTOCOL:-PLAINTEXT}'
  sasl_mechanism: '${KAFKA_SASL_MECHANISM:-PLAIN}'
  sasl_username: '${KAFKA_SASL_USERNAME:-}'
  sasl_password: '${KAFKA_SASL_PASSWORD:-}'

# 消息队列后端选择
mq:
  backend: '${MQ_BACKEND:-redis}'        # 'redis' 或 'kafka'
  consumer_group: 'ragflow-task-executor' # Kafka Consumer Group ID

6.2 .env 新增环境变量

# ===== Kafka Configuration =====
MQ_BACKEND=kafka                                # redis | kafka
KAFKA_BOOTSTRAP_SERVERS=kafka:9092
KAFKA_EXT_PORT=9093
KAFKA_EXT_HOST=localhost
KAFKA_DEFAULT_PARTITIONS=4
KAFKA_REPLICATION_FACTOR=1                      # 单节点=1, 生产=3
KAFKA_SECURITY_PROTOCOL=PLAINTEXT               # PLAINTEXT | SASL_PLAINTEXT | SASL_SSL
# KAFKA_SASL_MECHANISM=PLAIN
# KAFKA_SASL_USERNAME=
# KAFKA_SASL_PASSWORD=

6.3 common/settings.py 新增

# 新增全局配置
MQ_BACKEND = os.getenv('MQ_BACKEND', 'redis')
KAFKA_BOOTSTRAP_SERVERS = os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092')
KAFKA_CONSUMER_GROUP = os.getenv('KAFKA_CONSUMER_GROUP', 'ragflow-task-executor')

def get_mq_producer():
    """延迟初始化 MQ Producer"""
    from rag.utils.message_queue import MessageQueueFactory
    return MessageQueueFactory.get_producer()

def get_mq_consumer():
    """延迟初始化 MQ Consumer"""
    from rag.utils.message_queue import MessageQueueFactory
    return MessageQueueFactory.get_consumer(KAFKA_CONSUMER_GROUP)

# 保留兼容 (废弃标记)
def get_svr_queue_name(priority: int) -> str:
    """@deprecated: 仅在 MQ_BACKEND=redis 时使用"""
    ...

7. 迁移实施路线图

时间线总览

总计预估: 8 周 (2 人团队，含测试和灰度)

8. 风险评估与缓解

9. 附录

9.1 当前系统消息流全景图

9.2 依赖项对比

9.3 关键文件索引

9.4 术语对照表