一代明君的小屋 - celery
type: channel
link: https://yidaimingjvn.xyz/index.php/tag/celery/
items: celery与多ocr任务调度的一些思路梳理和情况
items: 基于mac的设备客户端任务队列调度服务
items: 分布式任务调度与资源管理系统设计
items: celery基于对象的逻辑栈分配设计
items: 关于celery使用的一些问题

celery与多ocr任务调度的一些思路梳理和情况
type: item
date: 2026-07-13T11:16:38+08:00
description: 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的设备客户端任务队列调度服务
type: item
date: 2026-06-18T17:39:19+08:00
description: 一、概述基于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…

分布式任务调度与资源管理系统设计
type: item
date: 2026-06-18T17:37:04+08:00
description: 一、系统概述本系统是一个面向内容检测场景的分布式异步任务调度与资源管理平台，核心提供三方面能力：任务调度：基于设备维度的逻辑队列 + 老化优先级调度，实现多租户公平调度与饥饿感知资源调度：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 中选取…

celery基于对象的逻辑栈分配设计
type: item
date: 2026-06-10T15:26:05+08:00
description: 运行逻辑图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…

关于celery使用的一些问题
type: item
date: 2025-07-08T17:17:01+08:00
description: 关于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