celery与多ocr任务调度的一些思路梳理和情况
Celery 与 OCR 调度逻辑梳理1. 总体架构系统不是把接收到的 OCR 请求直接投递到 Celery OCR 队列，而是分成两级队列：Redis 逻辑队列：按设备 MAC 隔离，用于削峰、公平调度、任务 TTL 和过载淘汰。(保证对每个设备的处理更公平，避免设备任务过多堵塞)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，定义三类队列：队列默认名称职责MAC 调度队列task.dispatch执行 MAC 领取和逻辑队列拆分OCR 队列task.ocr执行图片读取、OCR 和后续敏感词检测文本队列task.text执行已有文本的敏感词检测任务路由如下：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 入队过程：标准化 MAC 为 12 位小写十六进制字符串。将图片 bytes 上传至 FS，Celery/Redis payload 只保存远程 key，避免大二进制进入 Redis broker。生成 task_id、记录 detection_id、敏感词和创建时间。RPUSH 到该 MAC 的 Redis list。更新单 MAC pending、全局 pending 和调度 ZSET。初始化 DAG 的 enqueued 节点。按 MAC 分队列避免单个设备持续上传时垄断整个执行队列。5. 独立调度循环实际主调度器是 utils/ocr_task_dispatch.py(独立于celery开发的进程，不采用celery地beat来调度，避免抢占woker的任务机会)。独立进程默认每 0.5 秒执行一轮：恢复 defer 退避到期的 MAC。O(1) 读取 task:global_pending；无积压则不投递空 Celery 任务。根据存活 Worker 的资源和有效并发计算容量。获取可调度 MAC 数。在 Worker 容量、调度上限、待处理 MAC 数之间取最小值。从 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：根据是否存在饥饿 MAC 决定 batch size。每弹一条任务前刷新 MAC 锁 TTL，并检查本机资源是否过载。从 MAC 逻辑队列弹出任务，再投递 OCR 或文本 Celery 子任务。处理结束后更新调度分数或移除空队列。在 finally 中释放 MAC 锁。若其他 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 次：检查本机 CPU、内存和 GPU 显存高水位；过载时指数退避。写入 Worker slot，用于活动任务观测。从 FS 读取图片。调用 _run_ocr_job。异常时调用 Celery retry。finally 清除 Worker slot。OCR 业务流程：OCR 引擎记录、敏感检测记录置为 processing。DAG ocr_processing 置为 running。图片转 RGB，最长边限制为 1920，降低内存/显存压力。延迟初始化 OCR 引擎：CUDA 优先 PaddleOCR，CPU 使用 RapidOCR。使用全局锁包住推理，保护单例 OCR 引擎线程安全。保存文字、行数、平均置信度和耗时。对识别文本执行敏感词检测。命中时发送，写业务日志，并把图片存档到审计目录。更新 PostgreSQL 记录和 DAG。配置允许时删除 FS 临时图片。8. 重试、defer 与失败语义这里有两层不同的重试：调度前 defer当 MAC 任务出队后发现没有可用 Worker时：任务重新放回逻辑队列；defer count 增加；对整个 MAC 设置指数退避；超过阈值后进入 task:dead_letter。Celery 执行重试OCR 子任务本机过载或抛异常时，使用 Celery retry，最多 10 次，最长退避 300 秒。当前实现的语义问题has_capable_worker() 调用没有传 gpu_required=True，所以检查的是任意 Worker，而不是 GPU Worker。配置声明 reject/defer/skip_ocr，实际分支只有日志文案不同，最终都调用 requeue_task_with_defer()；skip_ocr 没有真正降级到 CPU 文本队列，reject 也没有立即进死信。_run_ocr_job() 捕获大部分 OCR 异常后只写失败状态并 return，外层 Celery task 看起来成功，因此不会触发 self.retry()。当前真正能触发 Celery retry 的主要是 RustFS 读取等向外抛出的异常。如果 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，降级时显式改路由。
基于mac的设备客户端任务队列调度服务
一、概述基于mac的设备客户端任务队列调度服务是一个基于 Redis 的、按设备标识（MAC）维度隔离的异步任务调度中间件，核心解决以下问题：多租户公平调度：大量设备共享有限的 Worker 资源，需保证各设备的任务都能被及时处理资源感知调度：不同任务对 CPU/GPU 资源需求不同，需感知 Worker 资源状况动态分配稳定性自保：防止某设备异常发量冲垮系统，也防止系统过载后雪崩该服务的核心思想是：为每个设备维护一个独立逻辑队列，通过老化优先级计算各设备的调度得分，结合分布式互斥锁实现多 Worker 安全的公平调度。二、数据模型整个调度系统的状态全部存储在 Redis 中，核心数据结构分为五类：┌─────────────────────────────────────────────────────────────────┐ │ MAC 队列调度系统 │ │ ┌───────────────────────────────────────────────────────────┐ │ │ │ 设备逻辑队列 │ 元数据 │ 调度集 │ 全局计数器 │ │ │ │ (List) │ (Hash) │ (ZSET) │ (String) │ │ │ │ │ │ │ │ │ │ │ task_mac_q:m1 │ meta:m1 │ schedule │ global_pend │ │ │ │ task_mac_q:m2 │ meta:m2 │ [m1:15.2] │ → 128 │ │ │ │ task_mac_q:m3 │ meta:m3 │ [m2:42.5] │ │ │ │ │ ... │ ... │ [m3:8.1] │ │ │ │ └────────────────┴──────────────┴───────────┴─────────────┘ │ │ │ │ ┌───────────────────────────────────────────────────────────┐ │ │ │ 运行锁 │ 过载集合 │ 死信队列 │ │ │ │ (String) │ (Set) │ (List) │ │ │ │ │ │ │ │ │ │ active:m1 → TTL │ overloaded: │ dead_letter │ │ │ │ active:m2 → TTL │ {m3, m7} │ [{task1}, ...] │ │ │ └──────────────────┴─────────────────┴──────────────────────┘ │ └─────────────────────────────────────────────────────────────────┘2.1 数据结构详表用途Redis 类型键格式值说明设备队列Listsensitive:mac:queue:{mac}JSON 序列化的任务 payload，按入队顺序排列设备元数据Hashsensitive:mac:meta:{mac}字段：pending_count(积压数)、oldest_created_at(最早任务时间)、unschedulable_until(退避截止时间)、updated_at(更新时间)调度集合ZSetsensitive:mac:schedulemember = mac，score = 老化调度分数全局计数器Stringsensitive:global_pending所有设备积压任务总数（原子计数器）运行锁Stringsensitive:mac:active:{mac}SET NX + TTL，防止多 Worker 同时调度同一设备过载集合Setsensitive:overloaded_macs积压超过阈值的设备集合死信队列Listsensitive:dead_letter超过最大退避次数的任务记录水位快照Stringsensitive:watermark定期发布的系统运行状态 JSON三、核心操作3.1 入队流程（Enqueue）客户端请求 │ ▼ 检查全局过载 ──> 过载 ──> 拒绝入队，返回 503 │ ▼ (正常) 构建任务 payload (含 task_id, task_type, mac, created_at 等) │ ▼ Redis Pipeline: ├─ RPUSH queue_key, serialized_payload (追加到队列尾部) ├─ HINCRBY meta_key pending_count 1 (递增积压计数器) └─ INCR sensitive:global_pending (递增全局计数器) │ ▼ 如果是该设备的第一个任务 └─ HSET meta_key oldest_created_at = payload.created_at │ ▼ 如果 pending_count > 单设备阈值 └─ SADD sensitive:overloaded_macs mac │ ▼ refresh_mac_schedule_score(mac) └─ 重新计算该设备的调度分数，更新 ZSET关键代码视角：def enqueue_payload(payload): pipeline = client.pipeline() pipeline.rpush(queue_key, serialized_payload) pipeline.hincrby(meta_key, "pending_count", 1) pipeline.incr("sensitive:global_pending") _, pending_count, _ = pipeline.execute() if int(pending_count) == 1: client.hset(meta_key, mapping={"oldest_created_at": payload["created_at"]}) if int(pending_count) > CELERY_MAX_PER_MAC_PENDING: client.sadd("sensitive:overloaded_macs", mac) refresh_mac_schedule_score(mac)3.2 调度流程（Scheduling）系统采用两级调度模型，在 Celery 的 Worker 进程中通过链式任务调用实现：第一级：全局调度器Celery Beat (每 15s) │ ▼ dispatch_next_mac_sensitive_tasks ├─ _resume_expired_backoff_macs() ← 恢复退避到期的设备 └─ claim_next_schedulable_mac() │ ├─ ZREVRANGE schedule 0 -1 ← 从高到低遍历所有候选设备 ├─ 跳过 exclude_mac（刚处理完的设备） ├─ 跳过退避未到期的设备（check unschedulable_until） └─ try_acquire_mac_dispatch() ← SET NX 加分布式锁 │ 成功 → 从 ZSET 移除该设备 → 投递 process task 失败 → 遍历下一个候选第二级：设备级调度器process_mac_sensitive_tasks(mac) │ ├─ 计算动态批大小 │ ├─ has_starved_mac()? → batch = MIN_BATCH (1) │ └─ 无饥饿 → batch = NORMAL_BATCH (10) │ ├─ loop (batch_size 次): │ ├─ _is_local_worker_overloaded()? → break (背压) │ ├─ refresh_mac_dispatch() → EXPIRE 续约锁 │ ├─ pop_next_mac_task() │ │ └─ RPOP 队列 → 过期检查 → 返回有效任务 │ └─ 根据 task_type 提交子任务到相应队列 │ └─ finally: ├─ 释放运行锁 (DEL active key) ├─ 有剩余任务? → 更新 ZSET 分数 / 从 ZSET 移除 └─ 有饥饿设备? → 续投全局调度器 (带 exclude_mac)3.3 出队流程（Dequeue）pop_next_mac_task(mac) │ ▼ RPOP queue_key ← 从队列尾部弹出（FIFO 语义） │ ▼ 任务为空? ├─ Yes → refresh_mac_schedule_score() → 返回 None │ └─ No → 检查 TTL 过期? ├─ 过期 → HINCRBY pending_count -1 → DECR global_pending │ → 日志记录 → continue（继续取下一条） │ └─ 有效 → HINCRBY pending_count -1 → DECR global_pending → pending_count <= 0? ├─ Yes → 删除 meta key，从 ZSET 移除 └─ No → 更新 oldest_created_at → refresh_mac_schedule_score() → pending_count <= 阈值? └─ SREM overloaded_macs → 返回 payload3.4 Defer 与指数退避当 OCR 任务投递时无可用 GPU Worker，触发 defer 逻辑：requeue_task_with_defer(payload) │ ├─ defer_count = payload.defer_count + 1 │ ├─ defer_count > MAX_DEFER(5)? │ └─ Yes → LPUSH dead_letter × LTRIM 0 9999 → 返回 False (死信) │ └─ No → RPUSH queue_key (重新入队) → pending_count +1 → global_pending +1 → _apply_defer_backoff(mac, defer_count) │ ├─ backoff = min(30 × 2^(defer-1), 600) ├─ HSET meta_key unschedulable_until = now + backoff └─ ZREM schedule mac (从调度集中移除) │ ▼ 返回 True (已重新入队)退避时间序列：#退避时长累计等待130s30s260s90s3120s210s4240s450s5480s930s → 死信3.5 任务 TTL 过期双向过期策略确保任务不会无限堆积：机制一：惰性过期（出队时判断） ─────────────────────────── pop_next_mac_task 中： if (now - created_at) > TTL(3600s): 丢弃，更新计数器，继续取下一条 机制二：定时批量清理（周期性全队列扫描） ─────────────────────────── purge_expired_mac_tasks (每 300s 执行一次): get_all_mac_queues() → SCAN 非阻塞扫描所有队列 for each mac: LRANGE 头部 100 条 统计连续过期数 LTRIM 保留未过期部分（批量删除） 更新全局计数器3.6 过载保护is_queue_overloaded() ├─ GET "sensitive:global_pending" > 5000? → True (全局限流) └─ SCARD "sensitive:overloaded_macs" > 0? → True (单设备限流) Worker 侧背压（process_mac_sensitive_tasks 循环中）: ├─ psutil.cpu_percent() > 75%? → break ├─ psutil.virtual_memory().percent > 75%? → break └─ GPU 显存 > 75%? → break四、简化数学模型4.1 调度分数设备 d 在时刻 t 的调度优先级分数为：S(d, t) = min(W(d, t), W_max) × α 其中： W(d, t) = t - t_first(d) ← 最早任务的等待时间（秒） W_max = 600 ← 等待时间上限（可配置） α = 1.0 ← 老化权重（可配置）分数越高，优先级越高。调度器使用 ZREVRANGE 按分数降序获取候选设备。4.2 动态批大小当前设备 d_cur 的批处理大小：B(d_cur) = / B_min = 1 , if ∃ d ≠ d_cur : W(d, t) ≥ T_starve \ B_norm = 10 , otherwise 其中： T_starve = 30 (饥饿判定阈值，秒)4.3 指数退避设备 d 在第 n 次退避时的禁调时长：Backoff(n) = min(Base × 2^(n-1), Backoff_max) 其中： Base = 30 (秒) Backoff_max = 600 (秒) n ∈ [1, 5]，n > 5 进入死信4.4 过载判定系统过载条件：Overloaded = (G > G_max) ∨ (|O| > 0) 其中： G = global_pending (全局积压总数) G_max = 5000 O = overloaded_macs 集合（pending_count > 200 的设备集合）五、模型代码以下是一个独立可运行的 Python 模型代码，实现了 MAC 队列调度服务的核心逻辑，不依赖 Celery、Flask 等框架，仅依赖 redis 库。""" MAC Queue Scheduling Service — Simplified Model A standalone demonstration of the core MAC-based task scheduling algorithm. Dependencies: pip install redis """ import json import time import uuid import logging from datetime import datetime, timezone from typing import Optional # ============================================================ # Configuration # ============================================================ class Config: REDIS_URL = "redis://127.0.0.1:6379/0" # Queue keys MAC_QUEUE_PREFIX = "model:mac:queue:" MAC_META_PREFIX = "model:mac:meta:" MAC_ACTIVE_PREFIX = "model:mac:active:" SCHEDULE_KEY = "model:mac:schedule" GLOBAL_PENDING_KEY = "model:global_pending" OVERLOADED_MACS_KEY = "model:overloaded_macs" DEAD_LETTER_KEY = "model:dead_letter" # Thresholds MAX_TOTAL_PENDING = 5000 MAX_PER_MAC_PENDING = 200 ACTIVE_TTL = 900 TASK_TTL = 3600 AGING_MAX_WAIT = 600 AGING_WEIGHT = 1.0 STARVATION_HINT = 30 BATCH_SIZE = 10 MIN_BATCH_SIZE = 1 MAX_DEFER_COUNT = 5 DEFER_BACKOFF_BASE = 30 DEFER_BACKOFF_MAX = 600 # ============================================================ # Redis client # ============================================================ import redis as redis_lib _redis: Optional[redis_lib.Redis] = None def get_redis(): global _redis if _redis is None: _redis = redis_lib.from_url(Config.REDIS_URL, decode_responses=True) return _redis # ============================================================ # Helper functions # ============================================================ def _iso_now() -> str: return datetime.now(timezone.utc).isoformat() def _serialize(payload: dict) -> str: return json.dumps(payload, ensure_ascii=False) def _deserialize(raw: Optional[str]) -> Optional[dict]: return json.loads(raw) if raw else None def _now_ts() -> float: return time.time() def _wait_seconds(created_at_iso: Optional[str]) -> float: if not created_at_iso: return 0.0 elapsed = (datetime.now(timezone.utc) - datetime.fromisoformat(created_at_iso)).total_seconds() return max(0.0, elapsed) def _schedule_score(wait_sec: float) -> float: capped = min(wait_sec, Config.AGING_MAX_WAIT) return capped * Config.AGING_WEIGHT # ============================================================ # Core API — Enqueue # ============================================================ def enqueue_task(mac: str, task_type: str, payload_data: dict) -> str: """ 将任务入队到指定 MAC 的逻辑队列。 返回 task_id。 """ r = get_redis() task_id = uuid.uuid4().hex payload = { "task_id": task_id, "task_type": task_type, "mac": mac, "created_at": _iso_now(), **payload_data, } queue_key = f"{Config.MAC_QUEUE_PREFIX}{mac}" meta_key = f"{Config.MAC_META_PREFIX}{mac}" # 原子入队 pipe = r.pipeline() pipe.rpush(queue_key, _serialize(payload)) pipe.hincrby(meta_key, "pending_count", 1) pipe.incr(Config.GLOBAL_PENDING_KEY) _, pending_count, _ = pipe.execute() # 首个任务记录 oldest_created_at if int(pending_count) == 1: r.hset(meta_key, "oldest_created_at", payload["created_at"]) # 单设备过载标记 if int(pending_count) > Config.MAX_PER_MAC_PENDING: r.sadd(Config.OVERLOADED_MACS_KEY, mac) # 刷新调度分数 _refresh_schedule_score(mac) return task_id # ============================================================ # Core API — Schedule # ============================================================ def _refresh_schedule_score(mac: str) -> Optional[float]: """重新计算并更新设备的调度分数。""" r = get_redis() queue_key = f"{Config.MAC_QUEUE_PREFIX}{mac}" meta_key = f"{Config.MAC_META_PREFIX}{mac}" pending = r.llen(queue_key) if pending <= 0: r.delete(meta_key) r.zrem(Config.SCHEDULE_KEY, mac) return None oldest = r.hget(meta_key, "oldest_created_at") if not oldest: # 从队列头部读取 head = r.lindex(queue_key, 0) head_payload = _deserialize(head) oldest = head_payload.get("created_at") if head_payload else None wait = _wait_seconds(oldest) score = _schedule_score(wait) r.hset(meta_key, mapping={ "pending_count": pending, "oldest_created_at": oldest or "", "updated_at": _iso_now(), }) r.zadd(Config.SCHEDULE_KEY, {mac: score}) return score def try_acquire_lock(mac: str) -> bool: """尝试获取设备的调度锁（SET NX + TTL）。""" r = get_redis() active_key = f"{Config.MAC_ACTIVE_PREFIX}{mac}" return bool(r.set(active_key, _iso_now(), nx=True, ex=Config.ACTIVE_TTL)) def release_lock(mac: str): """释放设备的调度锁。""" r = get_redis() active_key = f"{Config.MAC_ACTIVE_PREFIX}{mac}" r.delete(active_key) def claim_next_mac(exclude_mac: Optional[str] = None) -> Optional[str]: """ 从调度集中选出优先级最高的、可调度的设备。 返回值：选中的 mac，或 None（无可用设备）。 """ r = get_redis() now = _now_ts() for candidate in r.zrevrange(Config.SCHEDULE_KEY, 0, -1): if exclude_mac and candidate == exclude_mac: continue # 跳过退避中的设备 meta_key = f"{Config.MAC_META_PREFIX}{candidate}" until = r.hget(meta_key, "unschedulable_until") if until and float(until) > now: continue if try_acquire_lock(candidate): r.zrem(Config.SCHEDULE_KEY, candidate) return candidate return None # ============================================================ # Core API — Dequeue # ============================================================ def pop_task(mac: str) -> Optional[dict]: """ 从设备队列中取出一条有效任务（自动过滤过期任务）。 返回 payload dict，或 None（队列为空）。 """ r = get_redis() queue_key = f"{Config.MAC_QUEUE_PREFIX}{mac}" meta_key = f"{Config.MAC_META_PREFIX}{mac}" while True: raw = r.rpop(queue_key) payload = _deserialize(raw) if not payload: _refresh_schedule_score(mac) return None # 惰性过期检查 created_at = payload.get("created_at") if created_at and _wait_seconds(created_at) > Config.TASK_TTL: remaining = r.hincrby(meta_key, "pending_count", -1) r.decr(Config.GLOBAL_PENDING_KEY) if remaining <= 0: r.delete(meta_key) r.zrem(Config.SCHEDULE_KEY, mac) continue # 跳过过期任务 # 有效任务 remaining = r.hincrby(meta_key, "pending_count", -1) r.decr(Config.GLOBAL_PENDING_KEY) if remaining <= 0: r.delete(meta_key) r.zrem(Config.SCHEDULE_KEY, mac) else: # 更新 oldest_created_at head = r.lindex(queue_key, 0) head_payload = _deserialize(head) if head_payload: r.hset(meta_key, "oldest_created_at", head_payload["created_at"]) _refresh_schedule_score(mac) # 恢复过载标记 if remaining <= Config.MAX_PER_MAC_PENDING: r.srem(Config.OVERLOADED_MACS_KEY, mac) return payload # ============================================================ # Defer & Backoff # ============================================================ def requeue_with_defer(payload: dict) -> bool: """ 将任务重新入队并应用指数退避。 返回 True = 已重入队，False = 已入死信。 """ r = get_redis() mac = payload["mac"] queue_key = f"{Config.MAC_QUEUE_PREFIX}{mac}" meta_key = f"{Config.MAC_META_PREFIX}{mac}" defer_count = int(payload.get("defer_count", 0)) + 1 if defer_count > Config.MAX_DEFER_COUNT: # 移入死信队列 dead_entry = { "task_id": payload.get("task_id", "unknown"), "mac": mac, "task_type": payload.get("task_type"), "error": f"defer 超过阈值 ({defer_count})", "created_at": payload.get("created_at"), "dead_at": _iso_now(), } r.lpush(Config.DEAD_LETTER_KEY, _serialize(dead_entry)) r.ltrim(Config.DEAD_LETTER_KEY, 0, 9999) return False payload["defer_count"] = defer_count r.rpush(queue_key, _serialize(payload)) r.hincrby(meta_key, "pending_count", 1) r.incr(Config.GLOBAL_PENDING_KEY) # 计算退避时长并设置 backoff = min( Config.DEFER_BACKOFF_BASE * (2 ** (defer_count - 1)), Config.DEFER_BACKOFF_MAX, ) unschedulable_until = _now_ts() + backoff r.hset(meta_key, "unschedulable_until", str(unschedulable_until)) r.zrem(Config.SCHEDULE_KEY, mac) return True def resume_expired_backoffs() -> int: """扫描所有退避到期的设备，恢复其可调度状态。""" r = get_redis() now = _now_ts() cursor = 0 resumed = 0 pattern = f"{Config.MAC_META_PREFIX}*" while True: cursor, keys = r.scan(cursor=cursor, match=pattern, count=200) for key in keys: until_raw = r.hget(key, "unschedulable_until") if not until_raw: continue if float(until_raw) > now: continue mac = key[len(Config.MAC_META_PREFIX):] r.hdel(key, "unschedulable_until") pending = int(r.hget(key, "pending_count") or 0) if pending > 0: _refresh_schedule_score(mac) resumed += 1 if cursor == 0: break return resumed # ============================================================ # Overload Detection # ============================================================ def is_overloaded() -> bool: """全局过载检查（O(1)）。""" r = get_redis() total = int(r.get(Config.GLOBAL_PENDING_KEY) or 0) if total > Config.MAX_TOTAL_PENDING: return True if r.scard(Config.OVERLOADED_MACS_KEY) > 0: return True return False # ============================================================ # Starvation Detection & Dynamic Batch Size # ============================================================ def get_oldest_wait(mac: str) -> float: """获取设备最早任务的已等待秒数。""" r = get_redis() meta_key = f"{Config.MAC_META_PREFIX}{mac}" oldest = r.hget(meta_key, "oldest_created_at") return _wait_seconds(oldest) def has_starved_mac(exclude_mac: Optional[str] = None) -> bool: """检测是否存在等待超过饥饿阈值的设备。""" r = get_redis() for candidate in r.zrevrange(Config.SCHEDULE_KEY, 0, -1): if exclude_mac and candidate == exclude_mac: continue meta_key = f"{Config.MAC_META_PREFIX}{candidate}" pending = int(r.hget(meta_key, "pending_count") or 0) if pending > 0 and get_oldest_wait(candidate) >= Config.STARVATION_HINT: return True return False def get_batch_size(current_mac: str) -> int: """动态批大小：有饥饿时缩减，无饥饿时正常。""" if has_starved_mac(exclude_mac=current_mac): return Config.MIN_BATCH_SIZE return Config.BATCH_SIZE # ============================================================ # Purge Expired Tasks # ============================================================ def purge_expired_tasks() -> int: """扫描所有队列，批量清理过期任务。""" r = get_redis() prefix = Config.MAC_QUEUE_PREFIX total_purged = 0 cursor = 0 while True: cursor, keys = r.scan(cursor=cursor, match=f"{prefix}*", count=200) for key in keys: mac = key[len(prefix):] queue_len = r.llen(key) if queue_len == 0: continue scan_batch = min(queue_len, 100) raw_tasks = r.lrange(key, 0, scan_batch - 1) expired = 0 cutoff = None for i, raw in enumerate(raw_tasks): payload = _deserialize(raw) if not payload: continue if _wait_seconds(payload.get("created_at")) > Config.TASK_TTL: expired += 1 else: if cutoff is None: cutoff = i break if expired == 0: continue if cutoff is not None: r.ltrim(key, cutoff, -1) else: r.delete(key) r.decrby(Config.GLOBAL_PENDING_KEY, expired) pending = r.llen(key) meta_key = f"{Config.MAC_META_PREFIX}{mac}" if pending == 0: r.delete(meta_key) r.zrem(Config.SCHEDULE_KEY, mac) else: r.hset(meta_key, "pending_count", pending) _refresh_schedule_score(mac) total_purged += expired if cursor == 0: break return total_purged # ============================================================ # Watermark Publishing # ============================================================ def publish_watermark(): """发布系统水位快照。""" r = get_redis() prefix = Config.MAC_QUEUE_PREFIX total = 0 count = 0 cursor = 0 while True: cursor, keys = r.scan(cursor=cursor, match=f"{prefix}*", count=200) for key in keys: total += r.llen(key) count += 1 if cursor == 0: break worker_cursor = 0 worker_count = 0 while True: worker_cursor, keys = r.scan(cursor=worker_cursor, match="model:worker:*", count=200) worker_count += len(keys) if worker_cursor == 0: break watermark = { "total_pending": total, "mac_count": count, "active_workers": worker_count, "timestamp": _iso_now(), } r.set("model:watermark", json.dumps(watermark, ensure_ascii=False)) # ============================================================ # Demo: Basic usage scenario # ============================================================ def demo(): """ 演示 MAC 队列调度核心流程： 1. 三个设备入队 5 个任务 2. 模拟调度器选取最优设备 3. 出队处理 4. 模拟 defer + 退避 5. 饥饿检测 6. 过载检测 """ r = get_redis() r.flushdb() # Clean start print("=" * 60) print("MAC 队列调度服务 — 模型演示") print("=" * 60) # ── 1. 三个设备各入队若干任务 ── print("\n[1] 设备 A 入队 3 个任务") for i in range(3): tid = enqueue_task("A", "text", {"seq": i, "data": f"task-{i}"}) print(f" task_id={tid[:8]}..., seq={i}") print("\n[2] 设备 B 入队 1 个任务") tid = enqueue_task("B", "ocr", {"seq": 0, "data": "image-0"}) print(f" task_id={tid[:8]}...") print("\n[3] 设备 C 入队 5 个任务") for i in range(5): tid = enqueue_task("C", "text", {"seq": i, "data": f"task-{i}"}) print(f" task_id={tid[:8]}..., seq={i}") # ── 2. 查看调度分数 ── print("\n[4] 调度分数 (等待相同时间，队列越长分数越高)") scores = r.zrevrange(Config.SCHEDULE_KEY, 0, -1, withscores=True) for mac, score in scores: print(f" {mac}: score={score:.1f}") # ── 3. 模拟调度 ── print("\n[5] 模拟调度 (claim_next_mac)") selected = claim_next_mac() print(f" 选中设备: {selected}") # ── 4. 出队 ── print(f"\n[6] 从 {selected} 出队 2 个任务") for _ in range(2): task = pop_task(selected) if task: print(f" 出队: {task['task_id'][:8]}... type={task['task_type']}") else: print(" 队列为空") # ── 5. 释放锁 ── release_lock(selected) print(f"\n[7] 释放 {selected} 的运行锁") # ── 6. 饥饿检测 ── print(f"\n[8] 饥饿检测: has_starved_mac(exclude={selected}) = {has_starved_mac(selected)}") print(f" batch_size = {get_batch_size(selected)} (有饥饿时=1, 正常={Config.BATCH_SIZE})") # ── 7. Defer 退避 ── print(f"\n[9] 模拟 defer 退避") fake_payload = { "task_id": uuid.uuid4().hex, "task_type": "ocr", "mac": "B", "created_at": _iso_now(), } for d in range(1, 4): result = requeue_with_defer(fake_payload) print(f" defer #{d}: {'已入队' if result else '进入死信'}, " f"backoff={Config.DEFER_BACKOFF_BASE * (2**(d-1))}s") # ── 8. 过载检测 ── print(f"\n[10] 过载检测: is_overloaded() = {is_overloaded()}") print(f" global_pending = {r.get(Config.GLOBAL_PENDING_KEY)}") # ── 9. 最终调度图 ── print("\n[11] 最终调度集状态") scores = r.zrevrange(Config.SCHEDULE_KEY, 0, -1, withscores=True) if scores: for mac, score in scores: wait = get_oldest_wait(mac) print(f" {mac}: score={score:.1f}, oldest_wait={wait:.1f}s") else: print(" 调度集为空") print("\n" + "=" * 60) print("演示完毕") print("=" * 60) r.flushdb() if __name__ == "__main__": demo()运行方式pip install redis python mac_queue_model.py预期输出示例：============================================================ MAC 队列调度服务 — 模型演示 ============================================================ [1] 设备 A 入队 3 个任务 task_id=a1b2c3d4..., seq=0 task_id=e5f6g7h8..., seq=1 task_id=i9j0k1l2..., seq=2 [2] 设备 B 入队 1 个任务 task_id=m3n4o5p6... [3] 设备 C 入队 5 个任务 task_id=q7r8s9t0..., seq=0 ... [4] 调度分数 (等待相同时间，队列越长分数越高) C: score=0.0 A: score=0.0 B: score=0.0 # 所有设备等待时间接近 0，分数均为 0 # ZREVRANGE 返回顺序由 ZADD 顺序决定 [5] 模拟调度 (claim_next_mac) 选中设备: C [6] 从 C 出队 2 个任务 出队: q7r8s9t0... type=text 出队: ... [9] 模拟 defer 退避 defer #1: 已入队, backoff=30s defer #2: 已入队, backoff=60s defer #3: 已入队, backoff=120s [10] 过载检测: is_overloaded() = False六、与 Celery 框架的集成在实际系统中，上述模型代码由 Celery Task 驱动执行：Celery Beat (定时触发) │ ▼ dispatch_next_mac_sensitive_tasks (Task) ├─ 调用 claim_next_mac() └─ 投递 process_mac_sensitive_tasks │ ▼ process_mac_sensitive_tasks (Task) ├─ 调用 get_batch_size() → pop_task() × batch ├─ Worker 过载自检（背压） └─ 提交子任务到 ocr/text 队列 │ ▼ execute_ocr_sensitive_job (Task, 路由到 ocr 队列) execute_text_sensitive_job (Task, 路由到 text 队列)Celery 的 acks_late=True、worker_prefetch_multiplier=1 配置确保任务不丢失且 Worker 不会过度预取。七、关键设计决策决策选择理由队列类型Redis List (RPUSH/RPOP)O(1) 入队出队，天然 FIFO，无需额外排序调度优先级Redis ZSet天然支持分数排序，O(log N) 操作复杂度分布式锁SET NX + TTL无外部依赖，TTL 自动死锁恢复全局计数器Redis String + INCR/DECRO(1) 原子操作，高并发安全过期检查惰性 + 定时双向惰性无额外开销，定时防积累退避状态Hash 字段 + 时间戳结构简单，过期由定时任务恢复饥饿检测遍历 ZSet 计算等待时间每次调度前检查，精度高死信队列List + LTRIM 限容简单可靠，避免无限增长
分布式任务调度与资源管理系统设计
一、系统概述本系统是一个面向内容检测场景的分布式异步任务调度与资源管理平台，核心提供三方面能力：任务调度：基于设备维度的逻辑队列 + 老化优先级调度，实现多租户公平调度与饥饿感知资源调度：Worker 节点资源注册与心跳，CPU/GPU/内存感知的任务分发稳定性保障：过载保护、背压机制、指数退避、死信队列、任务 TTL 过期等多层可靠性策略系统基于 Flask + Celery + Redis + PostgreSQL 技术栈构建，采用三层架构（Web 路由层 → 业务服务层 → 异步任务层），所有异步任务通过 Celery 分发至专用队列，由 Worker 集群消费。二、整体架构┌─────────────────────────────────────────────────────────┐ │ Web 路由层 (Flask) │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ 检测结果上报 │ │ 图片处理任务 │ │ 文本处理任务 │ │ │ │ /api/decter │ │ /api/ocr/* │ │ /api/ocr/* │ │ │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ └─────────┼──────────────────┼──────────────────┼──────────┘ │ │ │ ▼ ▼ ▼ ┌─────────────────────────────────────────────────────────┐ │ 服务层 (Services) │ │ ┌────────────────┐ ┌──────────────────────────────┐ │ │ │ 检测结果处理服务 │ │ 任务处理服务 │ │ │ │ (同步处理) │ │ (异步入队 + 过载检查) │ │ │ └────────────────┘ └──────────┬───────────────────┘ │ │ │ │ │ ┌───────────────────────────────▼────────────────────┐ │ │ │ MAC 队列调度服务 │ │ │ │ 按设备维度逻辑队列 · 老化优先级排序 · 分布式互斥锁 │ │ │ │ 指数退避 · 死信 · 饥饿检测 · 动态批大小 │ │ │ └───────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────┐ │ 异步任务层 (Celery + Redis) │ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ 调度队列 │ │ 图片处理队列 │ │ 文本处理队列 │ │ │ │ dispatch │ │ ocr │ │ text │ │ │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌──────────────────────────────────────────────────┐ │ │ │ Worker 集群 │ │ │ │ GPU Worker (消费所有队列) · CPU Worker (消费部分) │ │ │ │ Worker 注册/心跳 · 过载自检 · 自适应 CPU/GPU │ │ │ └──────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────┐ │ 数据持久层 (PostgreSQL) │ │ 任务记录 · 处理结果 · 统计报表 │ └─────────────────────────────────────────────────────────┘三、任务调度系统3.1 核心调度模型系统采用两级调度模型：第一级：全局调度器 (Celery Beat + dispatch Task) → 从 Sorted Set 中选取最优 MAC 设备 → 加分布式互斥锁防止多 Worker 冲突 第二级：设备级调度器 (process_mac_sensitive_tasks Task) → 从该设备的 Redis List 中批量拉取任务 → Worker 过载自检（背压） → 提交具体子任务到专用队列 → 处理完毕后释放锁，触发下一轮全局调度3.2 按设备维度的逻辑队列每个设备拥有独立的 Redis List 作为任务队列：队列键格式：sensitive:mac:queue:{normalized_mac}元数据键格式：sensitive:mac:meta:{normalized_mac}（存储积压数、最早任务时间、退避状态）入队操作：RPUSH + 原子递增全局计数器 + 更新调度分数出队操作：RPOP + 原子递减全局计数器 + 惰性过期检查每设备独立队列的设计带来以下优势：优势说明故障隔离单设备大量任务不影响其他设备的调度公平调度每个设备独立计算调度优先级，防止少数设备占满所有 Worker退避隔离设备级指数退避，失败设备自动降权不影响正常设备3.3 老化优先级调度使用 Redis Sorted Set（sensitive:mac:schedule）维护所有活跃设备的调度优先级：调度分数 = min(等待秒数, CELERY_MAC_AGING_MAX_WAIT_SECONDS) × CELERY_MAC_AGING_WEIGHT关键特性：等待越久优先级越高：等待时间长的设备获得更高分数，ZREVRANGE 获取最高分设备等待时间上限：CELERY_MAC_AGING_MAX_WAIT_SECONDS（默认 600s）防止极端情况元数据缓存：最早任务的 created_at 缓存在 meta hash 中，避免反复读取队列头部惰性更新：每次入队/出队/心跳时触发 refresh_mac_schedule_score() 更新分数3.4 调度流程详解1. Beat 触发调度（周期 15s） │ ▼ 2. dispatch_next_mac_sensitive_tasks │ ├─ 恢复退避到期的 MAC │ └─ claim_next_schedulable_mac() │ ├─ 从 Sorted Set 中 ZREVRANGE 遍历候选 MAC │ ├─ 跳过退避中的 MAC（check unschedulable_until） │ └─ try_acquire_mac_dispatch() — SET NX + TTL 加锁 │ ▼ 3. process_mac_sensitive_tasks(mac) │ ├─ 计算动态批大小（get_dynamic_batch_size） │ │ ├─ 有饥饿设备 → 最小批大小（默认 1） │ │ └─ 无饥饿设备 → 正常批大小（默认 10） │ │ │ ├─ 循环处理批内任务 │ │ ├─ Worker 过载自检（背压） │ │ ├─ 刷新 MAC 运行锁 TTL │ │ ├─ pop_next_mac_task() — 惰性过期检查 │ │ ├─ 无可用 GPU Worker → 执行 defer/死信策略 │ │ └─ 提交子任务到专用队列 │ │ │ └─ finally: │ ├─ 释放 MAC 运行锁 │ ├─ 更新调度分数 / 移除空队列 │ └─ 续投全局调度器（有饥饿时传 exclude_mac） │ ▼ 4. execute_ocr_sensitive_job / execute_text_sensitive_job │ 子任务在专用队列上由 Worker 消费 │ ├─ 过载自检（retry 背压） │ ├─ 执行具体处理逻辑 │ └─ 结果持久化到 PostgreSQL3.5 饥饿检测与动态批大小饥饿检测机制确保没有设备被长期忽略：判定条件：下一个待调度设备的等待时间 ≥ CELERY_MAC_STARVATION_HINT_SECONDS（默认 30s）响应动作：当前设备的批大小从默认值（10）缩减为最小值（1），让出调度机会续投机制：处理完当前设备后，若有饥饿设备存在，全局调度器以 exclude_mac 参数续投，跳过刚处理完的设备def get_dynamic_batch_size(current_mac): if has_starved_mac(exclude_mac=current_mac): return max(1, CELERY_MAC_SCHEDULER_MIN_BATCH_SIZE) # 默认 1 return max(1, CELERY_MAC_SCHEDULER_BATCH_SIZE) # 默认 103.6 分布式互斥锁同一设备可能被多个 Worker 同时调度，需互斥保证：加锁：SET sensitive:mac:active:{mac} {timestamp} NX EX {TTL}续约：处理期间定期 EXPIRE 刷新 TTL解锁：处理完成后 DEL 删除锁超时：默认 900s（CELERY_MAC_ACTIVE_TTL），防止 Worker 崩溃导致死锁四、资源调度系统4.1 Worker 资源注册每个 Worker 启动时通过 @signals.worker_ready 钩子自动注册资源信息到 Redis：HSET sensitive:worker:{hostname}:{pid} hostname "worker-01" pid "12345" gpu_available "1" # PaddlePaddle CUDA 检测 gpu_util "23.5" # GPUtil gpu_memory_util "45.2" # GPUtil cpu_count "16" cpu_percent "35.0" # psutil memory_total "33559728128" memory_available "12559728128" memory_percent "62.5" # psutil active_tasks "0" last_heartbeat "2026-06-18T10:00:00+00:00" EXPIRE sensitive:worker:{hostname}:{pid} 604.2 Worker 心跳周期：15s（与 Beat 调度周期一致）更新内容：CPU 使用率、内存使用率、GPU 利用率、GPU 显存利用率TTL：60s（允许连续 3 次心跳丢失才判定死亡）扫描方式：SCAN 非阻塞扫描，避免 KEYS * 阻塞4.3 资源感知的任务分发系统根据 Worker 资源情况智能分发任务：任务类型目标队列资源要求调度策略图片处理任务（OCR）sensitive.ocr无硬性要求（可 CPU 可 GPU）检查是否有任意可用 Worker文本处理任务sensitive.text无特殊要求直接投递到队列调度任务sensitive.dispatch无特殊要求仅在调度器队列运行Worker 可用性过滤条件（get_available_workers）：Worker 的 Redis key 存在（TTL 未过期）内存使用率 < CELERY_WORKER_MEM_HIGH_WATERMARK（默认 75%）CPU 使用率 < CELERY_WORKER_CPU_HIGH_WATERMARK（默认 75%）4.4 自适应 CPU/GPU 模式Worker 在处理图片任务时，OCR 引擎根据运行时环境自动适配：Worker 启动 → is_cuda_available() 检测 ├─ 有 GPU → use_gpu=True, rec_batch_num=4, enable_mkldnn=False └─ 无 GPU → use_gpu=False, rec_batch_num=2, enable_mkldnn=True, cpu_threads=4运行时发生 OOM/CUDA 错误时自动降级：GPU OOM → 标记 _is_cuda=False → 重置 OCR 引擎 → 下次以 CPU 模式重初始化五、稳定性保障5.1 过载保护（多层防线）请求入口 │ ├─ 防线一：全局积压限流 │ ├─ 全局积压 > CELERY_MAX_TOTAL_PENDING（默认 5000）→ 拒绝入队 │ └─ 返回 HTTP 503 + retry_after: 30 │ ├─ 防线二：单设备积压限流 │ ├─ 单设备队列 > CELERY_MAX_PER_MAC_PENDING（默认 200） │ └─ 标记为过载设备（sensitive:overloaded_macs） │ ├─ 防线三：Worker 过载自检（调度时） │ ├─ CPU > 75% → 暂停拉取新任务 │ ├─ 内存 > 75% → 暂停拉取新任务 │ └─ GPU 显存 > 75% → 暂停拉取新任务 │ └─ 防线四：子任务级过载自检 ├─ CPU 过载 → self.retry(countdown=30) ├─ 内存过载 → self.retry(countdown=30) └─ GPU 显存过载 → self.retry(countdown=30)过载状态判断（is_queue_overloaded）基于 O(1) 的 Redis 计数器和集合：def is_queue_overloaded(): # O(1) 检查，无阻塞 total_pending = GET "sensitive:global_pending" if total_pending > CELERY_MAX_TOTAL_PENDING: return True overloaded_count = SCARD "sensitive:overloaded_macs" if overloaded_count > 0: return True return False5.2 背压机制（Backpressure）Worker 侧主动背压防止资源耗尽：调度时背压：process_mac_sensitive_tasks 循环中每次迭代检查本地 CPU/内存/GPU，超阈值立即 break执行时背压：execute_ocr_sensitive_job / execute_text_sensitive_job 启动时过载自检，超阈值则 self.retry(countdown=N)任务降级：无 GPU Worker 可用时的三种策略：策略行为适用场景reject直接丢弃，记录失败GPU 完全不可用，不接受降级defer指数退避后重试期望 GPU 资源即将恢复skip_ocr跳过 OCR 处理业务允许降级5.3 指数退避（Exponential Backoff）设备级退避机制避免频繁重试打满系统：backoff_seconds = min( CELERY_MAC_DEFER_BACKOFF_BASE × 2^(defer_count - 1), # 30, 60, 120, 240, 480... CELERY_MAC_DEFER_BACKOFF_MAX # 600s cap )退避效果：退避次数退避时长累计等待130s30s260s90s3120s210s4240s450s5480s930s (阈值) → 死信退避 MAC 通过 unschedulable_until 时间戳记录在 meta hash 中，调度器在 claim_next_schedulable_mac() 中跳过，退避到期后由 _resume_expired_backoff_macs() 自动恢复。5.4 死信队列（Dead Letter Queue）超过最大重试次数的任务移入死信队列：LPUSH sensitive:dead_letter { task_id, mac, task_type, error: "defer 次数超过阈值", requeue_count: 6, created_at, dead_at } LTRIM sensitive:dead_letter 0 9999 # 保留最近 10000 条死信队列通过 LTRIM 限制大小，避免无限制增长耗尽 Redis 内存。5.5 任务 TTL 过期机制所有任务携带 created_at 时间戳，确保不会无限排队：机制触发方式说明惰性过期pop_next_mac_task() 出队时消费时判断 TTL，过期则丢弃并继续取下一条定时清理Beat 每 300s 触发 purge_expired_mac_tasks使用 LRANGE 扫描队列头部最多 100 条，LTRIM 批量删除def _is_task_expired(payload): elapsed = now - created_at return elapsed > CELERY_MAC_TASK_TTL_SECONDS # 默认 3600s5.6 水位监控（Watermark）定期发布系统水位到 Redis，支撑弹性扩缩容决策：SET sensitive:watermark { total_pending: 1234, # 全局积压数 mac_count: 42, # 活跃设备数 active_workers: 8, # 存活 Worker 数 timestamp: "2026-06-18T10:00:00" } EXPIRE sensitive:watermark 20发布周期：CELERY_WATERMARK_PUBLISH_INTERVAL（默认 15s），扫描使用 SCAN 非阻塞模式。六、系统配置总览6.1 核心配置项配置项默认值说明CELERY_MAX_TOTAL_PENDING5000全局最大积压任务数CELERY_MAX_PER_MAC_PENDING200单设备最大积压任务数CELERY_WORKER_CPU_HIGH_WATERMARK75%Worker CPU 过载阈值CELERY_WORKER_MEM_HIGH_WATERMARK75%Worker 内存过载阈值CELERY_WORKER_GPU_MEM_HIGH_WATERMARK75%Worker GPU 显存过载阈值CELERY_MAC_ACTIVE_TTL900s设备运行锁超时时间CELERY_MAC_TASK_TTL_SECONDS3600s任务存活时间CELERY_MAC_AGING_MAX_WAIT_SECONDS600s老化等待时间上限CELERY_MAC_STARVATION_HINT_SECONDS30s饥饿判定阈值CELERY_MAC_SCHEDULER_BATCH_SIZE10正常批处理大小CELERY_MAC_SCHEDULER_MIN_BATCH_SIZE1饥饿时最小批处理大小OCR_FALLBACK_STRATEGYreject降级策略CELERY_OCR_MAX_DEFER_COUNT5最大退避次数CELERY_MAC_DEFER_BACKOFF_BASE30s退避基时CELERY_MAC_DEFER_BACKOFF_MAX600s退避最大时长CELERY_WORKER_TTL60sWorker 心跳过期时间CELERY_WATERMARK_PUBLISH_INTERVAL15s水位发布间隔CELERY_MAC_AGING_WEIGHT1.0老化权重系数6.2 Celery 配置配置项值BrokerRedisSerializerJSONtask_acks_lateTruetask_reject_on_worker_lostTrueworker_prefetch_multiplier1result_expires180signore_resultTrue6.3 PostgreSQL 连接池配置项值pool_size10max_overflow20pool_timeout30spool_recycle1800s七、定时任务机制系统通过 Celery Beat 驱动三个核心定时任务，保障分布式系统的自运维能力：任务周期说明worker_heartbeat15sWorker 心跳更新，维持资源感知调度的基础数据publish_watermark15s队列水位发布，输出系统运行状态快照purge_expired_mac_tasks300s过期任务主动清理，防止积压任务耗尽内存这些定时任务与事件驱动流程（入队、出队、退避恢复）共同构成系统的完整调度闭环。八、数据流完整链路┌──────────┐ ┌──────────┐ ┌────────────┐ ┌──────────┐ │ 客户端 │───>│ Web 路由 │───>│ Service │───>│ Redis │ │ │ │ /api/* │ │ 入队+过载检查 │ │ MAC 队列 │ └──────────┘ └──────────┘ └────────────┘ └─────┬────┘ │ ┌────────────────────────────────────┘ │ ▼ ┌─────────────────┐ │ Celery Beat │ │ (定时触发调度器) │ └────────┬────────┘ │ ▼ ┌─────────────────┐ │ 全局调度器 │ │ 选取最优 MAC │ │ 加分布式锁 │ └────────┬────────┘ │ ▼ ┌─────────────────┐ │ 设备级调度器 │ │ 批量拉取任务 │ │ Worker 背压 │ │ 提交子任务 │ └──┬──────────┬───┘ │ │ ▼ ▼ ┌────────────┐ ┌────────────┐ │ OCR 子任务 │ │ 文本子任务 │ │ GPU/CPU │ │ CPU │ │ OCR 识别 │ │ 敏感词匹配 │ │ 敏感词匹配 │ │ 结果入库 │ │ 结果入库 │ │ │ └────────────┘ └────────────┘九、关键设计决策决策选择理由设备维度队列Redis List简单可靠，LPUSH/RPOP O(1) 操作调度优先级Redis Sorted Set天然支持按分数排序，ZREVRANGE O(log N)分布式锁Redis SET NX + TTL无额外依赖，TTL 防死锁全局计数器Redis INCR/DECRO(1) 操作，高并发友好Worker 注册Redis Hash + TTLTTL 自动清理死 Worker过载检测psutil + GPUtil已在依赖中，零额外成本子任务拆分独立 Celery Task避免调度 Task 阻塞，支持路由到不同队列扫描而非 KEYSSCAN 游标避免 Redis 单线程阻塞死信队列LPUSH + LTRIM简单可靠，限制大小防内存泄漏十、可观测性设计10.1 关键可观测指标系统在设计上内置了丰富的可观测性指标，以支撑运行时状态推断：指标数据来源采集复杂度全局积压数Redis sensitive:global_pendingO(1)，原子计数器设备活跃数ZSET sensitive:mac:schedule 大小O(1) ZCARD存活 Worker 数SCAN sensitive:worker:*非阻塞游标扫描过载设备数SET sensitive:overloaded_macs 大小O(1) SCARD死信队列深度sensitive:dead_letter 长度O(1) LLEN队列水位快照sensitive:watermark JSONO(1) GET10.2 异常场景的容错设计场景设计策略Worker 进程崩溃Celery acks_late=True + worker_lost 自动重新入队，保证 at-least-once 语义Worker 集体不可用任务安全积压于 Redis 队列，Worker 恢复后自动继续消费，无人工介入需要Redis 实例故障任务数据丢失（无持久化），系统设计上接受此风险，依赖上游重试补偿单 Worker 内存泄漏Worker 进程级别隔离，单个 Worker OOM 不影响集群中其他 Worker调度器进程崩溃Beat 与 Worker 进程分离部署，Beat 重启后按周期恢复调度，不丢失调度状态资源竞争条件分布式锁 + defer 重试计数 + 死信队列三层兜底，防止任务无限卡死
运行逻辑图flowchart TD A[客户端请求 /text/handler 或 /calculate/handler] --> B[Router 参数校验/解密/取敏感词] B --> C[handlerService.text_handler / calculate_handler] C --> D[生成业务 task_id] D --> E[按 object 标准化 queue_object] E --> F[写入 Redis 逻辑队列 handler:object:queue:{queue_object}] F --> G{是否抢到该 object 的运行锁 handler:object:active:{queue_object}} G -- 否 --> H[结束提交 等待已有 dispatcher 继续消费] G -- 是 --> I[投递 Celery dispatcher 任务 dispatch_object_handler_tasks(object)] I --> J[Celery worker 固定消费 handler.dispatch] J --> K[dispatcher 启动] K --> L[循环处理最多 N 条 CELERY_object_DISPATCH_BATCH_SIZE] L --> M[从 Redis 队列 lpop 一条任务] M --> N{还有任务吗} N -- 否 --> O[释放 object 运行锁] O --> P[本轮 dispatcher 结束] N -- 是 --> Q{任务类型} Q -- text_handle --> R[_run_text_handler_job] Q -- calculate_handler --> S[_run_calculate_handler_job] R --> T[写 handlerDetectRecord processing] T --> U[执行敏感词匹配] U --> V{命中敏感词?} V -- 是 --> W[send_commend + write_log] V -- 否 --> X[正常结束] W --> Y[写 handlerDetectRecord success/failed] X --> Y S --> Z[写 calculateEngineRecord/handlerDetectRecord processing] Z --> AA[perform_calculate] AA --> AB{calculate 成功?} AB -- 否 --> AC[写 calculate failed + Detect failed] AB -- 是 --> AD[保存 calculate 成功结果] AD --> AE[执行敏感词匹配] AE --> AF{命中敏感词?} AF -- 是 --> AG[send_commend + write_log] AF -- 否 --> AH[正常结束] AG --> AI[写 handlerDetectRecord success/failed] AH --> AI Y --> AJ[刷新 object 锁 TTL] AI --> AJ AJ --> L L --> AK{本轮处理后 Redis 仍有剩余任务?} AK -- 否 --> O AK -- 是 --> AL[再次 apply_async 投递 dispatcher] AL --> P简化时序图sequenceDiagram participant Client as 客户端 participant Router as Router participant Service as handlerService participant Redis as Redis逻辑栈 participant Celery as Celery Dispatcher participant Job as 业务执行函数 participant DB as DB/日志/命令下发 Client->>Router: 提交 text/calculate 请求 Router->>Service: 调用 text_handler / calculate_handler Service->>Redis: rpush 到 object 对应逻辑队列 Service->>Redis: setnx active 锁 alt 抢锁成功 Service->>Celery: apply_async(dispatch_object_handler_tasks) else 抢锁失败 Service-->>Client: 返回 task_id，等待已有 dispatcher 消费 end Celery->>Redis: lpop 一条 object 任务 Redis-->>Celery: 返回 payload Celery->>Job: 调用 _run_text_handler_job / _run_calculate_handler_job Job->>DB: 写 processing 记录 Job->>Job: calculate/敏感词检测 opt 命中敏感词 Job->>DB: write_log Job->>DB: send_commend end Job->>DB: 写 success/failed 记录 Celery->>Redis: 检查是否还有剩余任务 alt 还有剩余 Celery->>Celery: 续投 dispatcher else 队列已空 Celery->>Redis: 删除 active 锁 end核心理解同一个 object：只会有一个 dispatcher 持有锁，所以严格串行。不同 object：会进入各自 Redis 逻辑栈，由多个 Celery worker 并行推进。Celery 侧：只维护固定 dispatcher 队列 handler.dispatch，不会因为设备增多而爆出大量物理队列。
关于celery安装由于部分依赖的需要，pip需要版本降级到24版本以下，如23.1版本python -m pip install pip==23.1关于windows下celery运行任务Celery 4 在 Windows 上存在一些限制，主要是由于其默认使用的 prefork 进程池不被 Windows 支持。Windows 系统不支持 fork() 系统调用，因此需要采用其他方式来运行 Celery 任务。使用 Eventlet 或 Gevent 池。这是在 Windows 上运行 Celery 4 的推荐方法。需要安装额外的库：pip install eventlet # 或 gevent然后使用 eventlet 或 gevent 作为 worker 池启动 Celery：celery -A SVAD worker --loglevel=info --pool=eventlet -c 1000