第18章：RAGFlow Redis 队列与文档解析任务调度

1 项目背景

业务场景

「云帆科技」的第 16 章综合实战交付后，系统平稳运行了一个月。但周一早晨，HR 部门一次性上传了 30 份新版制度的 PDF，触发了意想不到的问题：前 5 份文档在 2 分钟内就解析完成了，但从第 6 份开始，所有的文档都卡在"等待解析"状态，持续了整整 40 分钟。

运维小李排查后发现：Task Executor 的 Worker 默认只有 1 个，30 份文档按 FIFO（先进先出）顺序排队处理。更麻烦的是，有一份 200 页的扫描 PDF 占用了 Worker 长达 25 分钟，后面的 24 份文档只能干等。小李意识到：默认的 FIFO 队列不适合这种"大小文档混合"的场景——就像超市结账，一个人买了一整车，后面拿一瓶水的也得排半小时。

痛点

简单的 FIFO 任务队列在复杂场景下的局限：

FIFO 队列的"长任务阻塞"问题：

Worker 1（单线程）处理顺序：
[文档1: 2MB PDF, 50页] → 8分钟
[文档2: 200页扫描件] → 25分钟 ← 后面的全卡住
[文档3: 5KB Markdown] → 等25分钟 → 1秒完成
[文档4: 20KB DOCX] → 等25分钟 → 3秒完成
…
后面24个文档平均等待时间：25分钟+

2 项目设计

小胖：（焦急地指着监控屏）“大师！你快看看，Redis 队列里积压了 73 个任务！Task Executor 只有一个 Worker 在吭哧吭哧干活，后面的文档全在排队。我们能不能多开几个 Worker？就像银行柜台——排队人多了多开几个窗口？”

大师：“方向对了，但没那么简单。多开 Worker（多线程/多进程）确实能并行处理多个文档，但有几个问题需要考虑：每个 Worker 会加载一份 Embedding 模型到内存，2 个 Worker = 2 倍内存；不同 Worker 可能同时写入同一个文档引擎索引，需要处理并发冲突。”

技术映射：Worker = 银行柜台窗口，开得越多同时服务的人越多，但每个窗口都要配一个柜员（内存），而且多个柜员不能同时往同一个账户里存钱（索引写入冲突）。

小胖：“那 RAGFlow 具体是怎么做任务调度的？按什么规则？”

大师：“核心是 Redis Stream + 消费组模型。RAGFlow 把每个解析任务包装成一条消息投递到 Redis Stream（ragflow_tasks），Task Executor 的 Worker 作为消费组的消费者从 Stream 中拉取消息。”

RAGFlow 任务调度架构：

Redis Stream: ragflow_tasks
├── Consumer Group: task_executors
│ ├── Worker-1 (consumer_id: worker_1_uuid)
│ ├── Worker-2 (consumer_id: worker_2_uuid)
│ └── Worker-3 (consumer_id: worker_3_uuid)
│
消息格式:
{
"task_id": "task_abc123",
"doc_id": "doc_xyz789",
"dataset_id": "ds_001",
"action": "parse",
"priority": "normal", // normal / high /紧急
"retry_count": 0,
"created_at": 1718000000
}

小白：“Stream 和传统的 List 有什么区别？为什么不用 LPUSH/BRPOP？”

大师：“三个关键差异：”

“最重要的是——Stream 的消费者如果宕机，它已经取走但未确认（XACK）的消息在 PEL 中，其他消费者可以认领（XCLAIM）这些超时未确认的消息继续处理。这就是容错机制。”

技术映射：List = 食堂打饭，打完就没了；Stream = 快递签收系统——快递员取出包裹（XREADGROUP）但必须收件人签收（XACK），否则系统知道这件还没送达，可以换个人送。

小胖：“那我如果想让重要文档优先处理，怎么搞？总不能改代码吧？”

大师：“RAGFlow 在 Stream 消息中预留了 priority 字段。实现优先级队列的思路有多种：”

方案1：多 Stream（简单但粗粒度）
ragflow_tasks:high → 高优先级 Stream
ragflow_tasks:normal → 普通 Stream
ragflow_tasks:low → 低优先级 Stream
Worker 同时 XREADGROUP 三个 Stream，优先处理 high

方案2：加权轮询（单 Stream，对 Worker 逻辑简单）
高优先级消息多分配 Consumer，普通消息少分配
例如 3 个 Worker：2 个处理 normal，1 个处理 high

方案3：消息优先级 + XREADGROUP 排序（依赖 Redis 7.2+）
Stream 消息可按 score 排序，优先消费 score 高的

小白：“那解析失败的任务呢？RAGFlow 有没有自动重试？”

大师：“目前 RAGFlow 的默认行为是：失败的任务不自动重试，retry_count 字段在消息中递增，超过最大重试次数（默认为 3）后消息被移入死信队列（DLQ）。要实现自动重试，有两个关键点：”

# 源码概念：幂等性保证（简化）
# 文件: rag/svr/task_executor.py

def handle_parse_task(doc_id, retry_count=0):
# 第1步：查询当前文档状态
doc = Document.get_by_id(doc_id)

# 第2步：如果是重试，先清理旧的解析结果
if retry_count > 0:
# 删除已生成的 Chunk
Chunk.delete().where(Chunk.doc_id == doc_id).execute()
# 删除文档引擎中的旧索引
delete_index(doc.dataset_id, doc_id)

# 第3步：执行解析
try:
chunks = parse_and_chunk(doc)
embeddings = embed_chunks(chunks)
save_to_engine(embeddings)

# 第4步：更新状态并确认消息
doc.status = "success"
doc.save()
xack(task_id) # 确认消费

except TemporaryError as e:
# 可重试的错误（网络超时、模型暂时不可用）
if retry_count < MAX_RETRIES:
# 重新入队并设延迟
retry_task(doc_id, retry_count + 1, delay=exponential_backoff(retry_count))
else:
# 超限 → 死信队列
move_to_dlq(doc_id, error=str(e))

except PermanentError as e:
# 不可重试的错误（文件损坏、格式不支持）
doc.status = "failed"
doc.error_msg = str(e)
doc.save()
xack(task_id)

3 项目实战

环境准备

目标：上传 50 份文档，对比不同 Worker 数量下的解析吞吐和队列积压情况。

前提：RAGFlow 拆分部署（第17章），Task Executor 独立运行。

分步实现

步骤1：观察 Redis Stream 内部状态

目标：熟悉 Redis Stream 的监控命令。

# 连接 Redis
docker exec -it ragflow-redis redis-cli

# 查看所有 Stream
> SCAN 0 TYPE stream
# 返回包含 "ragflow_tasks" 的 key

# 查看 Stream 基本信息
> XINFO STREAM ragflow_tasks
# 返回: length（消息总数）, first-entry, last-entry, groups 等

# 查看消费组
> XINFO GROUPS ragflow_tasks
# 返回: group: task_executors, consumers: 2, pending: 3

# 查看待处理（未确认）的消息
> XPENDING ragflow_tasks task_executors
# 返回: (待处理数量, 最早消息ID, 最晚消息ID, 各消费者分布)

# 查看某个消费者未确认的消息列表
> XPENDING ragflow_tasks task_executors – + 10

预期输出示例：

XINFO STREAM ragflow_tasks
length: 47 ← 当前积压 47 个任务
first-entry: 1718000000000-0
last-entry: 1718000045000-0
groups: 1
last-generated-id: 1718000045000-0

XPENDING ragflow_tasks task_executors
1) (integer) 3 ← 3 个任务正在处理中（未确认）
2) "1718000030000-0" ← 最早未确认
3) "1718000040000-0" ← 最晚未确认
4) 1) 1) "worker_1" ← worker_1 有 2 个
2) "2"
2) 1) "worker_2" ← worker_2 有 1 个
2) "1"

步骤2：对比不同 Worker 数量的吞吐量

目标：用 WS=1, WS=3, WS=5 三种配置跑同一批 30 个文档，对比解析耗时。

# 实验设计脚本
#!/bin/bash
DOC_COUNT=30

for WS in 1 3 5; do
echo "=== 实验: WS=$WS ==="

# 设置 Worker 数并重启 Task Executor
docker compose -f docker-compose-split.yml up -d \\
–env WS=$WS ragflow-task-executor
sleep 10 # 等待启动

# 记录开始时间
START=$(date +%s)

# 批量上传 30 个测试文档
for ((i=1; i<=DOC_COUNT; i++)); do
curl -X POST "http://localhost/api/v1/datasets/$DS_ID/documents" \\
-H "Authorization: Bearer $TOKEN" \\
-F "file=@test_docs/sample_${i}.pdf"
done

# 等待全部解析完成（轮询）
while true; do
QUEUE_LEN=$(docker exec ragflow-redis redis-cli XLEN ragflow_tasks)
PENDING=$(docker exec ragflow-redis redis-cli XPENDING ragflow_tasks task_executors | head -1)
echo " 队列: $QUEUE_LEN, 处理中: $PENDING"
if [ "$QUEUE_LEN" = "0" ]; then
break
fi
sleep 5
done

END=$(date +%s)
ELAPSED=$((END – START))
THROUGHPUT=$(echo "scale=1; $DOC_COUNT / ($ELAPSED / 60)" | bc)

echo " 完成! 总耗时: ${ELAPSED}s (${THROUGHPUT} 文档/分钟)"
echo ""
done

预期结果：

=== 实验: WS=1 ===
完成! 总耗时: 840s (2.1 文档/分钟)

=== 实验: WS=3 ===
完成! 总耗时: 310s (5.8 文档/分钟)

=== 实验: WS=5 ===
完成! 总耗时: 210s (8.6 文档/分钟)

坑点：Worker 数不是越多越好。5 个 Worker 各自加载 BGE-Large 模型 = 6GB 内存占用。如果机器内存只有 16GB，5 个 Worker 会触发 OOM Killer 随机杀死进程。

步骤3：模拟 Worker 崩溃与任务恢复

目标：验证 Stream 的 PEL 机制如何保证任务不丢失。

# 终端1：监控 PEL
watch -n 2 'docker exec ragflow-redis redis-cli XPENDING ragflow_tasks task_executors'

# 终端2：上传文档并在解析过程中杀死 Worker
# 上传一个大文档，便于在解析过程中操作
curl -X POST "http://localhost/api/v1/datasets/$DS_ID/documents" \\
-H "Authorization: Bearer $TOKEN" \\
-F "file=@test_docs/200page_scan.pdf"

# 等 10 秒（解析开始）
sleep 10

# 强制杀死 Task Executor
docker kill ragflow-task-executor

# 观察 PEL —— 应该有 1 条未确认的消息（正在解析的那个文档）
# XPENDING 应该返回:
# 1) (integer) 1

# 重启 Task Executor
docker start ragflow-task-executor
sleep 15

# 观察 PEL —— 重新启动后，Worker 会从 PEL 中 Claim 超时的消息
# 消息被重新处理，最终确认（PEL 变回 0）

步骤4：实现优先级队列与长任务拆分

目标：扩展 RAGFlow 的任务调度逻辑，区分紧急和普通任务。

# extended_scheduler.py – 优先级调度演示（概念代码）
import redis
import json
import time

class PriorityTaskScheduler:
"""支持优先级的 RAGFlow 任务调度器"""

def __init__(self, redis_host="localhost", redis_port=6379):
self.redis = redis.Redis(host=redis_host, port=redis_port)
self.streams = {
"high": "ragflow_tasks:high",
"normal": "ragflow_tasks:normal",
"low": "ragflow_tasks:low",
}
self.group = "task_executors"
self.consumer_id = f"worker_{os.getpid()}"

# 初始化 Stream 和消费组
for stream in self.streams.values():
try:
self.redis.xgroup_create(stream, self.group, id="0", mkstream=True)
except redis.ResponseError:
pass # 消费组已存在

def enqueue(self, doc_id, priority="normal", metadata=None):
"""入队任务"""
stream = self.streams.get(priority, self.streams["normal"])
message = {
"doc_id": doc_id,
"priority": priority,
"retry_count": 0,
"created_at": time.time(),
"metadata": json.dumps(metadata or {}),
}
self.redis.xadd(stream, message)
print(f"[ENQUEUE] {doc_id} -> {priority} queue")

def dequeue_with_priority(self, count=1, block_ms=5000):
"""带优先级的消费：先取高优先级，再取普通，最后取低优先级"""
# 按优先级顺序尝试消费
for priority in ["high", "normal", "low"]:
stream = self.streams[priority]
result = self.redis.xreadgroup(
self.group, self.consumer_id,
{stream: ">"}, count=count, block=0 # 不阻塞，立即返回
)
if result:
return result
return None

def acknowledge(self, stream, message_id):
"""确认消息处理完成"""
self.redis.xack(stream, self.group, message_id)

def claim_timeout_messages(self, min_idle_ms=60000):
"""认领超时未确认的消息（故障恢复）"""
for stream in self.streams.values():
pending = self.redis.xpending_range(
stream, self.group, min="-", max="+", count=100
)
for msg in pending:
msg_id = msg["message_id"]
idle_time = msg.get("time_since_delivered", 0)
if idle_time > min_idle_ms:
# 认领此超时消息
claimed = self.redis.xclaim(
stream, self.group, self.consumer_id,
min_idle_ms, msg_id
)
if claimed:
print(f"[RECOVERY] Claimed timeout message: {msg_id}")

# 使用示例
scheduler = PriorityTaskScheduler()

# 入队不同优先级
scheduler.enqueue("doc_urgent_report", "high")
scheduler.enqueue("doc_daily_update", "normal")
scheduler.enqueue("doc_archive_scan", "low")

# 消费（高优先级总是先被消费）
while True:
tasks = scheduler.dequeue_with_priority()
if tasks:
for stream, messages in tasks:
for msg_id, data in messages:
doc_id = data.get(b"doc_id", b"").decode()
priority = data.get(b"priority", b"normal").decode()
print(f"[PROCESS] {doc_id} (priority={priority})")
# … 执行解析 …
scheduler.acknowledge(stream, msg_id)
else:
time.sleep(1)

步骤5：死信队列与失败告警

目标：建立失败任务的兜底机制。

# 创建死信队列监控脚本
cat > dlq_monitor.sh << 'EOF'
#!/bin/bash
# RAGFlow 死信队列监控

DLQ_STREAM="ragflow_tasks:dlq"
MAX_RETRIES=3
ALERT_WEBHOOK="https://hooks.slack.com/xxx" # 替换为实际告警 Webhook

while true; do
DLQ_COUNT=$(docker exec ragflow-redis redis-cli XLEN $DLQ_STREAM 2>/dev/null || echo "0")

if [ "$DLQ_COUNT" -gt 0 ]; then
echo "[$(date)] ⚠ 死信队列中有 $DLQ_COUNT 个失败任务"

# 获取死信详情
FAILED=$(docker exec ragflow-redis redis-cli XRANGE $DLQ_STREAM – + COUNT 5)

# 发送告警
curl -s -X POST "$ALERT_WEBHOOK" \\
-H "Content-Type: application/json" \\
-d "{
\\"text\\": \\"⚠ RAGFlow 死信队列告警\\n死信数量: $DLQ_COUNT\\n最近失败: $FAILED\\"
}"
fi

sleep 300 # 每 5 分钟检查一次
done
EOF

chmod +x dlq_monitor.sh

测试验证

# test_task_scheduler.py
import pytest
import redis
import time

class TestTaskQueue:
def test_fifo_order_preserved(self):
"""验证 FIFO 顺序被保持"""
r = redis.Redis(host="localhost", port=6379, db=0)
stream = "test_fifo_stream"
r.delete(stream)

# 入队 5 个任务
for i in range(5):
r.xadd(stream, {"task_id": f"task_{i}", "order": str(i)})

# 消费并验证顺序
messages = r.xread({stream: "0"}, count=5)[0][1]
orders = [m[1][b"order"].decode() for _, m in messages]
assert orders == ["0", "1", "2", "3", "4"]

def test_worker_recovery(self):
"""验证 Worker 崩溃后消息可恢复"""
r = redis.Redis(host="localhost", port=6379, db=0)
stream = "test_recovery_stream"
group = "test_group"
r.delete(stream)

try:
r.xgroup_create(stream, group, id="0", mkstream=True)
except:
pass

# 入队一个任务
msg_id = r.xadd(stream, {"task": "important"})

# Worker 1 消费但不确认（模拟崩溃）
r.xreadgroup(group, "worker_dead", {stream: ">"}, count=1)

# 确认消息在 PEL 中
pending = r.xpending(stream, group)
assert pending["pending"] == 1

# Worker 2 认领超时消息（min_idle=0 用于测试）
claimed = r.xclaim(stream, group, "worker_recovery", 0, [msg_id])
assert len(claimed) > 0

# 确认完成
r.xack(stream, group, msg_id)
pending = r.xpending(stream, group)
assert pending["pending"] == 0

完整代码清单

4 项目总结

优点 & 缺点

适用场景

不适用场景：

注意事项

常见踩坑经验

思考题

（答案提示见第19章末尾或附录 D。）

延伸阅读与资源

10倍开发者的 Dify 魔法书：从零构建全栈 AI 应用 后端工程师转型AI第一课-Ollama 与私有化大模型实战 大型语言模型(LLM) vLLM 高性能推理落地实战 Agent开发之LlamaIndex 实战修炼与源码进阶 大语言模型Transformers 实战修炼与源码剖析