基于mac的设备客户端任务队列调度服务 ——key设计
调度服务前文基于mac的设备客户端任务队列调度服务一、MAC 任务调度相关 Keysensitive:mac:queue:<mac> — List（每个 MAC 一个）每个设备的任务队列。入队时 RPUSH，消费时 RPOP（LIFO 栈行为）。过期检查、死信重投也在此操作。sensitive:mac:active:<mac> — String（SET NX EX）运行锁，防止同一 MAC 被多个 Worker 同时调度。NX 保证原子性获得锁，EX 自动过期（默认 900 秒）防止死锁，消费完成后手动 DELETE 释放。sensitive:mac:schedule — ZSet（全局唯一）调度有序集合，score = 老化等待秒数（封顶 600 秒 × 权重）。ZREVRANGE 按等待时间从高到低获取候选 MAC，实现 "等最久的先调度" 策略。sensitive:mac:meta:<mac> — Hash（每个 MAC 一个）元数据，包含以下字段：字段说明pending_count队列积压任务数oldest_created_at最旧任务的创建时间，用于计算老化分数updated_at最后更新时间unschedulable_until退避到期时间戳（defer 指数退避时设置）二、Worker 管理相关 Keysensitive:worker:<hostname>:<pid> — Hash（每个 Worker 进程一个）Worker 资源注册与心跳，包含 CPU/GPU/内存利用率、活跃任务数、最后心跳时间等。Worker 启动时 HSET 注册，每 15 秒 心跳更新，TTL 过期（默认 30 秒）后自动清理。三、全局监控与过载保护 Keysensitive:global_pending — String（计数器）全局待处理任务总数。入队 INCR，出队 DECR，用于快速判断系统整体负载。sensitive:overloaded_macs — Set过载 MAC 集合。当单个 MAC 积压超过阈值（默认 200）时 SADD，恢复后 SREM。与 global_pending 一起用于 is_queue_overloaded() 过载判断。sensitive:watermark — String（JSON）队列水位快照，每 15 秒 由 publish_watermark 定时任务更新，包含 total_pending、mac_count、active_workers，TTL 为发布间隔 + 5 秒。四、死信与异常 Keysensitive:dead_letter — List死信队列。OCR 任务 defer 超过最大次数（默认 5 次）后移入此处，保留最近 10000 条供人工排查。abnormal:parsed_queue — List异常数据解析队列。decter 路由收到的异常数据 RPUSH 入队，限制上限 5000 条，TTL 20 秒兜底。五、Celery Broker 队列（由 Celery 管理）队列名用途sensitive.dispatch调度器队列，运行 dispatch_next_mac_sensitive_tasks / process_mac_sensitive_taskssensitive.ocrOCR 敏感检测子任务队列sensitive.text文本敏感检测子任务队列Key 关系图 ┌─────────────────────┐ │ sensitive:schedule │ ← ZSet, 按等待时间排序 └──────────┬──────────┘ │ ┌─────────────────────┼─────────────────────┐ ▼ ▼ ▼ ┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐ │ queue:<mac_A> │ │ queue:<mac_B> │ │ queue:<mac_C> │ ← List, 任务数据 │ active:<mac_A> │ │ active:<mac_B> │ │ active:<mac_C> │ ← String, 运行锁 │ meta:<mac_A> │ │ meta:<mac_B> │ │ meta:<mac_C> │ ← Hash, 元数据 └──────────────────┘ └──────────────────┘ └──────────────────┘ │ ┌─────────────┴─────────────┐ ▼ ▼ ┌──────────────────┐ ┌──────────────────────┐ │ global_pending │ │ sensitive:worker:* │ ← Worker 心跳 │ overloaded_macs │ │ watermark │ ← 水位快照 │ dead_letter │ │ parsed_queue │ ← 异常数据 └──────────────────┘ └──────────────────────┘
基于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，不会因为设备增多而爆出大量物理队列。
关于鲁班猫rk3568适配的linux系统使用新版本python的情况
前言开发了一些程序需要放到3568上运行。但是开发环境是python3.12，而最新版本的rk3568适配的系统是ubt20，系统默认的版本是python3.8其次，为了规避linux系统使用python安装依赖与系统依赖冲突的问题。需要采用python虚拟环境来运行情况1 默认版本是3.8需要编译3.12从源码编译（最通用、最干净）这是 ARM 开发板上安装新版 Python 的标准做法，虽然编译时间稍长，但兼容性最好，也不会破坏系统自带的 python3。1.安装编译依赖（已安装过的可跳过）sudo apt update sudo apt install -y wget build-essential libreadline-dev libncursesw5-dev \ libssl-dev libsqlite3-dev tk-dev libgdbm-dev libc6-dev libbz2-dev \ libffi-dev zlib1g-dev2.下载 Python 3.12 源码去 Python 官网 找最新的 3.12 版本链接（例如 3.12.4）：wget https://www.python.org/ftp/python/3.12.4/Python-3.12.4.tgz tar -xf Python-3.12.4.tgz cd Python-3.12.43.配置、编译、安装（使用 altinstall 避免覆盖系统 python3）# 配置（可选 --enable-optimizations 会提升性能但编译时间翻倍） ./configure --enable-optimizations # 使用全部CPU核心编译，RK3568大概需要30-60分钟 make -j$(nproc) # 关键：用 altinstall 而不是 install sudo make altinstall4.验证python3.12 --version你会看到 Python 3.12.x，同时系统默认的 python3 依然是 3.8.2，互不冲突。情况2 安装python3.12-venv报错这里的处理很简单，编译版本的python环境已经带了venv工具，直接使用python3.12 -m venv venv工具既可
前言在开发AwiseOctopus的过程中，最初对于memory的设计直限于经验记忆的设计——设计大致如下：设定一个经验总结的llm，当agent完成整个任务时，交给这个agent总结成功经验并给予权重值，然后记入memory数据库，当agent执行同类任务时即可优先调用这个经验记忆的操作。想象美好，现实骨感。实际操作中，经验memory的注入反而导致了agent的上下文污染，产生了严重幻觉。例如：本来任务是操作A文件，但是由于历史经验操作的是B文件，注入后导致了经验错误，agent实际执行了B文件的操作。处理思路为了解决这一问题，以下记录了处理注入的简单实验过程。数据分层基本思路是，将经验记忆和当前上下文进行基本分层，基于信息提取优先级。让agent对于当前上下文的信息做出较高优先级的提取。再整合当前的经验和上下文。而非全量注入。首先优化一下记忆的读取，将一些关联度较低的经验过滤掉def search_experience(self, task_type, instruction, top_k=3): """搜索历史经验，根据 weight 分为成功和失败两类""" if not self.collection: return "" # 1. 向量检索最相关的 instruction results = self.collection.query( query_texts=[instruction], n_results=top_k * 2, where={"task_type": task_type} ) if not results['ids'] or not results['ids'][0]: return "" # 过滤距离大于 0.3 的结果 exp_ids = [] if 'distances' in results and results['distances']: for i, distance in enumerate(results['distances'][0]): if distance <= 0.38: exp_ids.append(results['ids'][0][i]) else: exp_ids = results['ids'][0] if not exp_ids: return ""对上下文做分层，将历史经验放置到system提示词而不是用户提示词。提示词做解释 # 搜索历史经验 hint = self.memory_manager.search_experience("thinking", user_request) if hint: yield ("RUNNING", "\n[思考Agent 经验记忆] 检索到相关历史经验，已注入上下文。") # 数据分层 for msg in self.messages: if msg["role"] == "system": msg["content"] += "\n" + ( "〖系统注入的历史经验（仅供参考，如与用户最新指令冲突，以用户指令为准）〗\n" f"{hint}\n" "〖经验参考结束〗" ) break else: self.messages.append({ "role": "system", "content": ( "〖系统注入的历史经验（仅供参考，如与用户最新指令冲突，以用户指令为准）〗\n" f"{hint}\n" "〖经验参考结束〗" ) })注: self.memory_manager是一个memory管理工具，会负责从向量数据库提取相关数据。整体结果：agent的智力明显提升，对于已经有经验的相关事件，操作速度和准确率得到了明显提升，出现需要操作A而去操作B的情况大幅下降。虽然概率下降了，但依然偶有找不到，或者用户操作目录过于模糊而导致操作失误的问题。操作工作区新增工作区机制，让agent的每个session的操作都默认限定在某个工作区(文件夹或目录)内。除非得到了用户的指定，否则操作不出工作区。整体结果：对于不确定，不明确的事项，当agent出现偏差操作时，由于有工作区兜底，操作行为风险降低。其次当把所需操作行为，操作对象放到工作区内，由于操作选项范围减少，agent操作更加精确，进一步减少了幻觉。
问题处理串口数据过程中，当前后有连续的数据的情况下，很有可能出现数据粘包的情况。如需要的数据是ef0001和de002但是数据传输和接收过快产生粘包得到数据ef0001de00处理思路(自定协议)再数据包前增加1位或者2位的数据长度帧，通过先获取长度帧判断需要取的数据位。参考代码from .FingerProtocol import * import serial import time from ._Types import * from colorama import Fore, Style __all__ = [ "FingerManager", ] class FingerManager: def __init__(self,serial_conn:serial.Serial) -> None: self.serial_conn = serial_conn def send_cmd(self, packet): """发送命令包""" self.serial_conn.write(packet) self.serial_conn.flush() def read_packet(self,timeout=2): """ 读取一个完整的数据包 1. 读取包头9字节 (Header+Addr+ID+Len) 2. 解析包长度 (大端) 3. 读取剩余数据 (Len指定长度) 1. 检查是否超时 2. 读取数据 3. 检查是否完整 """ data = b'' start_time = time.time() # 1. 读取包头9字节 (Header+Addr+ID+Len) while len(data) < 9: if time.time() - start_time > timeout: return None if self.serial_conn.in_waiting > 0: data += self.serial_conn.read(min(9 - len(data), self.serial_conn.in_waiting)) else: time.sleep(0.001) # 2. 解析包长度 (大端) pkg_len = (data[7] << 8) | data[8] # 3. 读取剩余数据 (Len指定长度) total_len = 9 + pkg_len while len(data) < total_len: if time.time() - start_time > timeout: return None if self.serial_conn.in_waiting > 0: data += self.serial_conn.read(min(total_len - len(data), self.serial_conn.in_waiting)) else: time.sleep(0.001) return data
如何通过python采集windows正在运行的进程信息
基本思路使用psutil采集进程数据，通过pid对比提取出父子关系。示例import psutil import os import win32api import argparse import csv import json from datetime import datetime # 常见 Windows 系统关键进程（不完全列表，用于标记） SYSTEM_PROCESS_NAMES = { "System", "Idle", "smss.exe", "csrss.exe", "wininit.exe", "winlogon.exe", "services.exe", "lsass.exe", "lsm.exe", "svchost.exe", "dwm.exe", "conhost.exe", "fontdrvhost.exe", "audiodg.exe", "taskhostw.exe", "RuntimeBroker.exe", "sihost.exe", "ctfmon.exe", "dllhost.exe" } def get_file_version_info(file_path): try: lang, codepage = win32api.GetFileVersionInfo(file_path, '\\VarFileInfo\\Translation')[0] string_file_info = f'\\StringFileInfo\\{lang:04x}{codepage:04x}\\' company = win32api.GetFileVersionInfo(file_path, string_file_info + 'CompanyName') description = win32api.GetFileVersionInfo(file_path, string_file_info + 'FileDescription') return (company or "Unknown").strip(), (description or "No description").strip() except Exception: return "Unknown", "【版本信息不可读】文件可能损坏、非标准PE格式或权限不足" def get_process_source(proc): """推断进程启动来源""" try: parent_pid = proc.ppid() if parent_pid == 0: return "系统初始进程" parent = psutil.Process(parent_pid) parent_name = parent.name() if parent_name.lower() == "explorer.exe": return "用户启动（通过资源管理器）" elif parent_name.lower() in ["cmd.exe", "powershell.exe", "pwsh.exe", "windowsterminal.exe"]: return "用户启动（通过命令行）" elif parent_name.lower() in ["services.exe", "svchost.exe"]: return "系统服务启动" else: return f"{parent_name}" except (psutil.NoSuchProcess, psutil.AccessDenied): return "父进程已退出或无权限访问" except Exception: return "无法确定来源" def collect_running_processes(): processes_info = [] # 获取当前登录用户名（用于辅助判断） current_user = None try: current_user = psutil.users()[0].name if psutil.users() else "Unknown" except Exception: current_user = "Unknown" for proc in psutil.process_iter(['pid', 'name', 'username']): pid = proc.info['pid'] name = proc.info['name'] username = proc.info.get('username') or "Unknown" exe = None company = "Unknown" description = "" tag = "普通进程" source = "未知来源" # 判断是否为系统进程 if name in SYSTEM_PROCESS_NAMES or pid in (0, 4): tag = "系统进程" # 尝试获取 Source（需先获取 ppid） try: source = get_process_source(proc) except Exception: source = "无法确定来源" try: exe = proc.exe() except (psutil.AccessDenied, psutil.NoSuchProcess): processes_info.append({ 'PID': pid, 'Name': name, 'Path': "[不可访问]", 'Company': "Unknown", 'Description': "【权限受限】无法访问进程信息（可能为受保护系统进程）", 'Tag': tag, 'Source': source }) continue except Exception as e: processes_info.append({ 'PID': pid, 'Name': name, 'Path': "[错误]", 'Company': "Unknown", 'Description': f"【未知错误】{str(e)}", 'Tag': tag, 'Source': source }) continue if not exe or exe == "": processes_info.append({ 'PID': pid, 'Name': name, 'Path': "[无]", 'Company': "Microsoft Corporation", 'Description': "【特殊进程】无对应可执行文件（如 System、Idle 等内核实体）", 'Tag': "系统进程", # 强制标记 'Source': source }) continue if not os.path.exists(exe): processes_info.append({ 'PID': pid, 'Name': name, 'Path': exe, 'Company': "Unknown", 'Description': "【路径无效】可执行文件已删除或为虚拟进程", 'Tag': tag, 'Source': source }) continue # 正常读取版本信息 try: company, file_desc = get_file_version_info(exe) if file_desc.startswith("【"): description = file_desc else: description = file_desc or "No description" except Exception: description = "【版本信息不可读】读取文件资源失败" processes_info.append({ 'PID': pid, 'Name': name, 'Path': exe, 'Company': company, 'Description': description, 'Tag': tag, 'Source': source }) return processes_info def save_to_csv(data, filename): fieldnames = ['PID', 'Name', 'Path', 'Company', 'Description', 'Tag', 'Source'] with open(filename, mode='w', newline='', encoding='utf-8-sig') as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(data) print(f"✅ 数据已保存至 CSV 文件: {filename}") def save_to_json(data, filename): with open(filename, mode='w', encoding='utf-8') as f: json.dump(data, f, indent=4, ensure_ascii=False) print(f"✅ 数据已保存至 JSON 文件: {filename}") def main(): parser = argparse.ArgumentParser(description="采集 Windows 进程信息（含系统标记与启动来源）") parser.add_argument('-f', '--format', choices=['csv', 'json', 'both'], help="导出格式") parser.add_argument('-o', '--output', default=None, help="输出文件名前缀") args = parser.parse_args() print("正在收集进程信息（含 Tag 与 Source 字段）...\n") processes = collect_running_processes() processes.sort(key=lambda x: (x['Tag'] != '系统进程', x['Name'].lower())) # 系统进程排前面 if not args.format: print(f"共收集到 {len(processes)} 个进程。\n") print("示例（前8条）：") print(f"{'PID':<6} {'Name':<20} {'Tag':<8} {'Source'}") print("-" * 70) for p in processes[:8]: print(f"{p['PID']:<6} {p['Name']:<20} {p['Tag']:<8} {p['Source'][:40]}") print("\n使用 -f csv/json 导出完整数据。") return base_name = args.output or f"processes_{datetime.now().strftime('%Y%m%d_%H%M%S')}" if args.format in ['csv', 'both']: save_to_csv(processes, f"{base_name}.csv") if args.format in ['json', 'both']: save_to_json(processes, f"{base_name}.json") if __name__ == "__main__": main()
class InfoLayout(QVBoxLayout): def __init__(self): super().__init__() # 空白窗口 space_window = QWidget() space_window.setFixedHeight(100) # 英文标题 english_title_window = QWidget() english_title_window.setFixedHeight(80) english_title_window.setContentsMargins(0,0,0,-80) english_title_window.setStyleSheet("") ethl = QHBoxLayout() icon = QLabel("●") icon.setFont(QFont("SimHei", 8, QFont.Weight.Bold)) icon.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter) # 设置右对齐垂直居中 icon.setStyleSheet("color: #4CC9F0;") system_name = QLabel("SECURE ACCESS") system_name.setFont(QFont("SimHei", 24, QFont.Weight.DemiBold)) system_name.setAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignVCenter) # 设置左对齐垂直居中 system_name.setStyleSheet("color: white;") ethl.setStretch(0,1) ethl.setStretch(1,3) ethl.addWidget(icon) ethl.addWidget(system_name) english_title_window.setLayout(ethl) # 中文标题 title_window = QWidget() title_window.setFixedHeight(80) thl = QHBoxLayout() space_label = QLabel("") space_label.setAlignment(Qt.AlignmentFlag.AlignRight) title_label = QLabel("安全认证系统") title_label.setFixedHeight(80) title_label.setFont(QFont("SimHei", 18, QFont.Weight.Bold)) title_label.setAlignment(Qt.AlignmentFlag.AlignLeft) title_label.setStyleSheet("color: #8B98A9; margin: 20px;") thl.addWidget(space_label) thl.addWidget(title_label) title_window.setLayout(thl) # 产品介绍 piw = ProjectIntraduce() # 组合窗口 self.addWidget(space_window) self.addWidget(english_title_window) self.addWidget(title_window) self.addWidget(piw,1)
关于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