—— coding: utf-8 ——

""“ 生产级 Loop 循环系统最小演示 多 Worker 并发处理 + 优雅停机 + 失败重试 ”"" import asyncio import random class ProductionLoop: "““一个能扛流量的循环系统：任务队列 + 多Worker + 排空停机””" def init(self, workers=3, max_retries=3): self.queue = asyncio.Queue() self.workers = workers self.max_retries = max_retries self.running = True self.done_count = 0 self.idle_count = 0 self.retry_map = {} async def _process(self, task, wid): "““模拟一次真实的循环处理：有 30% 概率随机失败””“ await asyncio.sleep(0.1) if random.random() < 0.3: raise RuntimeError(“模拟故障”) print(f“[worker{wid}] 完成第 {task} 个任务”) async def worker(self, wid): ”““单个 Worker 主循环：取任务 -> 处理 -> 确认/重试””“ while self.running or not self.queue.empty(): if self.running: try: task = await asyncio.wait_for(self.queue.get(), timeout=0.3) except asyncio.TimeoutError: self.idle_count += 1 continue else: task = self.queue.get_nowait() try: await self._process(task, wid) except Exception as exc: times = self.retry_map.get(task, 0) if times >= self.max_retries: print(f“[worker{wid}] 任务 {task} 重试 {times} 次仍失败，丢弃：{exc}”) else: self.retry_map[task] = times + 1 print(f“[worker{wid}] 任务 {task} 失败，回队重试第 {times + 1} 次”) await self.queue.put(task) await asyncio.sleep(0.05) else: self.done_count += 1 async def shutdown(self): ”““优雅停机：停止接收新任务，让 Worker 排空队列””“ print(“收到停机信号，Worker 完成手头任务后排空队列……”) self.running = False async def main(): engine = ProductionLoop(workers=3, max_retries=3) for i in range(20): engine.queue.put_nowait(i) print(“放入 20 个任务，启动 3 个 Worker……”) tasks = [asyncio.create_task(engine.worker(i)) for i in range(engine.workers)] await asyncio.sleep(2) await engine.shutdown() await asyncio.gather(*tasks) print(f“停机完成：成功 {engine.done_count} 个，Worker 空闲 {engine.idle_count} 次”) if name == ”main": random.seed(42) asyncio.run(main()) ```