Celery 与 OCR 调度逻辑梳理

1. 总体架构

系统不是把接收到的 OCR 请求直接投递到 Celery OCR 队列，而是分成两级队列：

1. Redis 逻辑队列：按设备 MAC 隔离，用于削峰、公平调度、任务 TTL 和过载淘汰。(保证对每个设备的处理更公平，避免设备任务过多堵塞)
2. Celery 执行队列：分为调度队列、OCR 队列和文本检测队列，用于真正执行任务。
   核心链路：

业务请求
  -> 图片上传 FS
  -> Redis 按 MAC 逻辑队列
  -> 独立调度循环挑选 MAC
  -> task.dispatch
  -> process_mac_tasks 批量弹出任务
  -> sensitive.ocr
  -> execute_ocr_job
  -> FS 读取图片
  -> OCR 推理
  -> 敏感词检测
  -> 命中后写入业务库和日志
  -> PostgreSQL/DAG 写状态
  -> 清理 FS 临时图片

Redis 同时承担 Celery broker/backend、逻辑任务队列、调度索引、MAC 锁、Worker 心跳与运行槽位记录。

2. Celery 初始化与队列

Celery 使用 REDIS_URL 作为 broker 和 backend，定义三类队列：

任务路由如下：

• dispatch_next_mac_tasks、process_mac_tasks -> 调度队列。
• execute_ocr_job -> OCR 队列。
• execute_text_job -> 文本队列。
  可靠性参数：
• task_acks_late=True：任务完成后确认；Worker 异常退出时可重新投递。
• task_reject_on_worker_lost=True：Worker 丢失时拒绝并回队列。
• worker_prefetch_multiplier=1：避免单 Worker 预取大量重任务。
• ignore_result=True：业务状态写数据库/DAG，不依赖 Celery result backend，减少 Redis 写入。
  注意：晚确认意味着任务是 at-least-once，可能重复执行，并不等同于 exactly-once。

3. 启动方式与实际并发

可以设计支持 worker、beat来发发挥celery的任务执行和beat机制：

• GPU Worker 默认 concurrency=2、pool=solo。
• CPU Worker 默认 concurrency=4、pool=prefork。
• Worker 默认同时消费三个队列。
• Beat 只执行调度循环、维护服务的健康检查。
  可以通过自己写一个脚本把声明的并发、pool、Worker 类型和队列写入环境变量，Worker ready 后注册到 Redis。

关键点：solo 的有效并发固定为 1

即使 GPU Worker 传入 -c 2，solo pool 仍只有一个执行槽。注册逻辑专门将其 effective_concurrency 记为 1，以免调度器按声明值 2 过量投递。
因此：

• 单进程 GPU 推理优先使用 solo + concurrency=1，表达更准确。
• 若确实需要 GPU 并行，通常应启动多个独立 Worker 进程，并谨慎评估每个进程各自加载一份模型造成的显存占用。（worker采用prefork，开启多个运行槽位每次都会加载模型，有显存风险，建议使用solo模式来手动开启worker，便于管理）

4. OCR 请求入队

业务入口先检查逻辑队列是否过载：

• 若过载，优先丢弃当前逻辑队列中最旧任务，为新任务腾出空间。
• 没有可淘汰任务时，返回繁忙和 retry_after=30。
  OCR 入队过程：

1. 标准化 MAC 为 12 位小写十六进制字符串。
2. 将图片 bytes 上传至 FS，Celery/Redis payload 只保存远程 key，避免大二进制进入 Redis broker。
3. 生成 task_id、记录 detection_id、敏感词和创建时间。
4. RPUSH 到该 MAC 的 Redis list。
5. 更新单 MAC pending、全局 pending 和调度 ZSET。
6. 初始化 DAG 的 enqueued 节点。
   按 MAC 分队列避免单个设备持续上传时垄断整个执行队列。

5. 独立调度循环

实际主调度器是 utils/ocr_task_dispatch.py(独立于celery开发的进程，不采用celery地beat来调度，避免抢占woker的任务机会)。
独立进程默认每 0.5 秒执行一轮：

1. 恢复 defer 退避到期的 MAC。
2. O(1) 读取 task:global_pending；无积压则不投递空 Celery 任务。
3. 根据存活 Worker 的资源和有效并发计算容量。
4. 获取可调度 MAC 数。
5. 在 Worker 容量、调度上限、待处理 MAC 数之间取最小值。
6. 从 ZSET 中领取 MAC，并投递 dispatch_next_mac_tasks。
   这样替代“Beat 每秒无条件发一个调度任务”的方式，避免空任务不断进入 broker。

Worker 容量计算

Worker ready 后在 Redis 注册：

• pool、声明并发、有效并发、消费队列；
• CPU、内存、GPU/显存利用率；
• 当前活动任务数；
• TTL 心跳。
  后台线程每 15 秒刷新资源和 slot TTL。调度器过滤超过 CPU/内存高水位的 Worker，并按 effective_concurrency 汇总容量。

当前 _get_idle_worker_capacity() 汇总的是符合资源阈值的 Worker 总槽位，并没有减去 active_tasks。随后再乘 CELERY_SCHEDULER_SLOT_FACTOR 计算动态上限，可能产生超投。更准确的思路是：

idle_slots = max(effective_concurrency - active_tasks, 0)

再对所有 Worker 汇总，并明确 slot factor 是允许排队的倍率还是实际并发倍率。

6. 按 MAC 的公平调度

每个 MAC 在 ZSET 中有 aging score：等待越久，优先级越高。领取时通过 Redis SET NX EX 获取 MAC 运行锁，并从 ZSET 移除，防止两个调度器同时处理同一 MAC。
process_mac_tasks：

1. 根据是否存在饥饿 MAC 决定 batch size。
2. 每弹一条任务前刷新 MAC 锁 TTL，并检查本机资源是否过载。
3. 从 MAC 逻辑队列弹出任务，再投递 OCR 或文本 Celery 子任务。
4. 处理结束后更新调度分数或移除空队列。
5. 在 finally 中释放 MAC 锁。
6. 若其他 MAC 已饥饿，优先续投其他 MAC。

FIFO 方向存在隐患

正常入队使用 RPUSH，但消费使用 RPOP，因此实际是 LIFO，新任务先执行。代码中维护 oldest_created_at 时读取 LINDEX 0，过载淘汰也用 LPOP，这些又按左端是最旧任务理解。
这会导致：

• 同一 MAC 的旧任务可能长期饥饿；
• aging score 基于最旧任务，但实际先处理最新任务；
• 维护逻辑和消费语义不一致。
  建议统一为 FIFO：RPUSH + LPOP；若重入队需要回到队首，则使用 LPUSH。修改时应同时核对 defer、过期清理和元数据更新方向。

7. OCR 子任务执行

execute_ocr_job 最多重试 10 次：

1. 检查本机 CPU、内存和 GPU 显存高水位；过载时指数退避。
2. 写入 Worker slot，用于活动任务观测。
3. 从 FS 读取图片。
4. 调用 _run_ocr_job。
5. 异常时调用 Celery retry。
6. finally 清除 Worker slot。
   OCR 业务流程：
7. OCR 引擎记录、敏感检测记录置为 processing。
8. DAG ocr_processing 置为 running。
9. 图片转 RGB，最长边限制为 1920，降低内存/显存压力。
10. 延迟初始化 OCR 引擎：CUDA 优先 PaddleOCR，CPU 使用 RapidOCR。
11. 使用全局锁包住推理，保护单例 OCR 引擎线程安全。
12. 保存文字、行数、平均置信度和耗时。
13. 对识别文本执行敏感词检测。
14. 命中时发送，写业务日志，并把图片存档到审计目录。
15. 更新 PostgreSQL 记录和 DAG。
16. 配置允许时删除 FS 临时图片。

8. 重试、defer 与失败语义

这里有两层不同的重试：

调度前 defer

当 MAC 任务出队后发现没有可用 Worker时：

• 任务重新放回逻辑队列；
• defer count 增加；
• 对整个 MAC 设置指数退避；
• 超过阈值后进入 task:dead_letter。

Celery 执行重试

OCR 子任务本机过载或抛异常时，使用 Celery retry，最多 10 次，最长退避 300 秒。

当前实现的语义问题

1. has_capable_worker() 调用没有传 gpu_required=True，所以检查的是任意 Worker，而不是 GPU Worker。
2. 配置声明 reject/defer/skip_ocr，实际分支只有日志文案不同，最终都调用 requeue_task_with_defer()；skip_ocr 没有真正降级到 CPU 文本队列，reject 也没有立即进死信。
3. _run_ocr_job() 捕获大部分 OCR 异常后只写失败状态并 return，外层 Celery task 看起来成功，因此不会触发 self.retry()。当前真正能触发 Celery retry 的主要是 RustFS 读取等向外抛出的异常。
4. 如果 perform_ocr() 返回 success=False，同样不会 Celery retry。
   建议先明确错误分类：

• 参数错误、图片损坏：不可重试，直接失败。
• RustFS/网络临时错误：Celery retry。
• CPU/内存/GPU 高水位：退避 retry。
• CUDA OOM：重置引擎后是否在同任务切 CPU 重试，或让任务失败后重新投递到专用 CPU OCR 队列，需要明确策略。
• 无目标 Worker：按配置真正执行 reject、defer 或 CPU OCR fallback。

9. OCR 引擎与 GPU 的坑

9.1 全局单例与 pool 模型

OCR 引擎是进程内单例：

• solo：一个进程一份模型，最可控。
• prefork：每个子进程可能各加载一份模型，显存成倍增长。
• threads：共享模型，但依赖全局锁，OCR 推理仍串行，并发收益有限。
  GPU 场景不宜简单通过增大 -c 提升吞吐，应先测量单模型显存、峰值临时显存和任务耗时。

9.2 OOM 降级可能不完整

perform_ocr() 遇到 OOM/CUDA 等关键字时把加速器改为 CPU并重置引擎，但当前调用立即返回失败，没有在本次调用内重新初始化 CPU 引擎并执行一次。下一任务才可能使用 CPU。
同时，清空 Python 引用不保证 Paddle/CUDA allocator 立即把显存归还系统。可选思路：

• GPU Worker 使用独立进程，设置每处理若干任务后回收子进程；
• 避免 prefork 多模型副本；
• 对 Paddle/CUDA 做针对性的缓存释放与实测，而不是只依赖 gc.collect()；
• 将 GPU OCR 与 CPU OCR 拆成不同队列和 Worker，降级时显式改路由。