写一个while True很简单，但让这个循环在生产环境里稳定跑上一个月，不崩、不重、不卡、能自动恢复，这完全是另一门学问。我之前见过太多脚本本地测得好好的，一上线就 CPU 飙高、任务被重复处理、进程怎么 kill 都不干净。问题从来不出在“循环能不能转”，而是出在“循环在什么条件下会停、会乱、会挂”。Loop Engineering 就是一套把循环当成独立工程组件来设计的实践：怎么启动、怎么优雅退出、怎么重试、怎么限流、怎么埋点、怎么从崩溃里恢复。这篇文章我会用“目录监听 + 文件处理 + 失败重试”这个真实项目，一步步把循环工程的核心问题拆开讲清楚，代码直接用 Python asyncio 写，你可以抄下来改改就用在业务里。

1. 先搞清楚：Loop Engineering 到底是在解决什么问题

1.1 循环不是“写出来”，而是“养出来”

日常开发里的循环，我会简单分成三类：事件循环、轮询/调度循环、重试循环。事件循环处理外部事件并分发，比如网络服务的主循环；轮询循环每隔一段时间检查条件，比如监听目录有没有新文件；重试循环则是任务失败后按策略反复执行，直到成功或达到上限。

这三类循环在真实项目里经常叠加出现。比如一个文件处理服务：主循环在轮询目录发现新文件，文件处理失败后进入重试循环，同时还要有一个看门狗循环检查任务是不是卡死了。每一层都需要单独设计，不能指望一个while True包打天下。真正上线跑起来，你会发现“循环”不是代码里的那个关键字，而是围绕它的一整套生命周期管理：启动、运行、终止、恢复、观测。

1.2 五个绕不开的工程问题

我把实际踩过的坑归纳成五个问题，也是 Loop Engineering 必须回答的问题：

1. 终止问题：循环怎么停下来？直接break会不会丢任务？进程退出时正在执行的任务怎么处理？
2. 重复问题：任务处理超时了，但下游其实已经处理完，重试会不会造成重复？怎么保证幂等？
3. 堆积问题：生产者速度大于消费者速度，队列内存会涨到什么程度？要不要限流？
4. 恢复问题：进程崩溃后，正在处理的任务状态在哪里？重启后是重新跑还是断点续跑？
5. 观测问题：当前跑了多少任务？失败率多少？重试队列里积压了多少？看不到这些数字，出故障只能靠猜。

1.3 什么时候需要 Loop Engineering

写一次性脚本不需要考虑这些，跑完就结束。但是只要满足下面任一条件，建议直接上工程化思路：

• 服务需要 7×24 小时运行，比如消息消费者、文件监听器；
• 多个线程或协程同时消费同一个队列；
• 任务执行可能失败，且需要自动重试；
• 任务处理耗时不稳定，有快有慢；
• 进程随时可能被重启，需要恢复现场。

说白了，只要循环是“长周期、多并发、会出错”的，就应该用 Loop Engineering 的方法论来设计。

2. 项目实战：从零构建一个可落地的任务循环引擎

2.1 场景设定：一个“文件处理管道”

我把实战项目设定成一个本地目录文件监听处理服务：外部系统持续往inbox目录投放文件，服务监听新文件，读取内容，调用远程接口处理。处理成功后把文件移动到done目录；失败则进入重试队列，按退避策略等待后再试；最多重试 5 次，仍然失败就移入dead目录。

这个场景足够典型：有监听循环、有任务执行循环、有重试循环，还涉及状态目录管理。你可以把它替换成“消息队列消费 + 业务处理”或“数据库扫描 + 数据同步”，核心逻辑完全一样。

目标功能清单：

2.2 技术选型：为什么用 Python asyncio

我选用 Python asyncio，而不是多线程，原因很直接：asyncio 的Queue、Event、Task能让循环逻辑非常清晰；asyncio.sleep是可取消的，重试等待期间能响应退出请求；单线程事件循环避免了多线程共享状态的锁竞争，对中等并发足够使用；后续如果要把内存队列换成 Redis Stream 或 Kafka，消费者逻辑几乎不用大改。下面的代码基于 Python 3.10+ 标准库，不需要安装第三方包，可以直接跑。

2.3 工程目录与代码骨架

先给一个最小但完整的工程结构：

loop_engine/ ├── main.py # 入口：启动/停止事件循环 ├── engine.py # 核心引擎：队列、工作协程、看门狗 ├── retry.py # 重试状态与退避策略 └── config.py # 配置项：并发数、退避参数、目录

retry.py定义任务封装和退避时间计算：

# retry.py import random import time from dataclasses import dataclass, field from enum import Enum class TaskStatus(str, Enum): PENDING = "pending" RUNNING = "running" SUCCESS = "success" FAILED = "failed" DEAD = "dead" @dataclass class Task: file_path: str max_retries: int = 5 attempts: int = 0 status: TaskStatus = TaskStatus.PENDING created_at: float = field(default_factory=time.time) def should_retry(self) -> bool: return self.attempts <= self.max_retries def backoff_delay(attempt: int, base: float = 1.0, factor: float = 2.0, jitter: float = 0.1) -> float: """指数退避 + 抖动，避免同一时刻大量重试""" exp = base * (factor ** attempt) return exp + random.uniform(0, exp * jitter)

任务状态用枚举定义，这是后续断点续跑和幂等判断的基础。

2.4 核心循环代码逐行拆解

engine.py是核心，直接给出可运行的完整代码：

# engine.py import asyncio import logging from pathlib import Path from retry import Task, TaskStatus, backoff_delay logger = logging.getLogger("loop_engine") class LoopEngine: def __init__(self, inbox: Path, done: Path, dead: Path, max_concurrency: int = 4, max_retries: int = 5): self.inbox = Path(inbox) self.done = Path(done) self.dead = Path(dead) self.max_concurrency = max_concurrency self.max_retries = max_retries self._queue: asyncio.Queue = asyncio.Queue(maxsize=100) self._retry_heap: list[Task] = [] self._stop_event = asyncio.Event() self._workers: list[asyncio.Task] = [] self.total_processed = 0 self.total_success = 0 self.total_failed = 0 async def start(self): self.inbox.mkdir(parents=True, exist_ok=True) self.done.mkdir(parents=True, exist_ok=True) self.dead.mkdir(parents=True, exist_ok=True) self._workers = [ asyncio.create_task(self._worker_loop(i), name=f"worker-{i}") for i in range(self.max_concurrency) ] asyncio.create_task(self._watchdog_loop(), name="watchdog") await self._run() async def stop(self): logger.info("stop requested") self._stop_event.set() await asyncio.gather(*self._workers, return_exceptions=True) async def _run(self): logger.info("engine started") await self._stop_event.wait() logger.info("engine stopped") async def _worker_loop(self, worker_id: int): logger.info("worker %s started", worker_id) while not self._stop_event.is_set(): try: task = await asyncio.wait_for(self._queue.get(), timeout=0.5) except asyncio.TimeoutError: continue task.status = TaskStatus.RUNNING try: await self._process(task) task.status = TaskStatus.SUCCESS self.total_success += 1 self._move(task.file_path, self.done) except Exception as exc: task.status = TaskStatus.FAILED task.attempts += 1 self.total_failed += 1 logger.warning("task %s failed: %s (attempt %d)", task.file_path, exc, task.attempts) if task.should_retry(): delay = backoff_delay(task.attempts) loop = asyncio.get_running_loop() self._retry_heap.append((loop.time() + delay, task)) else: logger.error("task %s exceeded max retries", task.file_path) task.status = TaskStatus.DEAD self._move(task.file_path, self.dead) finally: self.total_processed += 1 self._queue.task_done() async def _process(self, task: Task): # 模拟业务处理：读取文件、调用远程接口等 import random await asyncio.sleep(random.uniform(0.1, 0.5)) if random.random() < 0.15: raise RuntimeError("mock failure") data = task.file_path.read_text() logger.info("processed %s (len=%d)", task.file_path, len(data)) async def _watchdog_loop(self): """定期检查重试堆中的任务，到期后重新入队""" while not self._stop_event.is_set(): loop = asyncio.get_running_loop() now = loop.time() ready = [] rest = [] for item in self._retry_heap: if item[0] <= now: ready.append(item[1]) else: rest.append(item) self._retry_heap = rest for task in ready: await self._queue.put(task) await asyncio.sleep(1) def _move(self, file_path: str, target_dir: Path): src = Path(file_path) dst = target_dir / src.name src.replace(dst)

这段代码有几个细节必须展开讲。

第一个细节：worker 循环的停止条件。我没有写裸的while True，而是while not self._stop_event.is_set()。停止不是“突然打断”，而是一个信号，让 worker 在空闲时自然退出。如果 worker 正好阻塞在_queue.get()，我加了asyncio.wait_for(..., timeout=0.5)，让它最多半秒醒来一次检查停止标志。

第二个细节：重试任务不是通过asyncio.sleep单独等待，而是交给看门狗统一扫描。这样做避免了成千上万个 sleep 协程占用任务列表，也让退出时容易清理。真实项目里这个_retry_heap建议用heapq或者asyncio.PriorityQueue，我为了演示直白用了 list。

第三个细节：_queue.get()和_queue.task_done()必须成对出现。如果你用了queue.join()等待所有任务完成，漏掉task_done()会让 join 永远卡住。这个坑特别隐蔽，写的时候务必留意。

2.5 main.py 入口与优雅退出

# main.py import asyncio import logging import signal from pathlib import Path from engine import LoopEngine logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s") async def main(): engine = LoopEngine( inbox=Path("./inbox"), done=Path("./done"), dead=Path("./dead"), max_concurrency=4, max_retries=5, ) loop = asyncio.get_running_loop() for sig in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(sig, lambda: asyncio.create_task(engine.stop())) await engine.start() if __name__ == "__main__": asyncio.run(main())

signal.add_signal_handler是 asyncio 的标准做法，不借助额外线程池。收到 Ctrl+C 后调engine.stop()，触发优雅关闭。千万不要在信号处理里直接os.kill自己，否则正在处理的任务会直接丢掉。

3. 循环工程的四个关键设计点（保姆级细节）

3.1 终止条件：如何优雅退出循环

循环必须有一个可触发的退出条件。我在上面的代码里用了asyncio.Event作为停止信号，这样每个循环在下一个迭代开始前检查事件。但“优雅退出”不止是退出，还包含两个层次：

第一层，停止接收新任务。退出信号一旦触发，就不能再往队列里塞新任务，否则永远消费不完。在文件监听场景里，这意味着停止向队列添加新文件；在消息队列场景里，则是先取消订阅。

第二层，等待当前任务完成并决定未完成任务去向。最简单的策略是等所有 worker 处理完当前任务再退出。但如果有任务一直卡死，退出就会被无限拖住。我建议给每个任务包一层处理超时：

try: await asyncio.wait_for(self._process(task), timeout=30) except asyncio.TimeoutError: task.status = TaskStatus.FAILED

我在早期项目里踩过一个大坑：循环里调用第三方接口，超时设了 10 分钟，进程关闭时直接杀掉，数据库里任务状态永远停留在 running。后来加上超时控制和状态标记，才算彻底解决。所以请记住：所有while True都必须有一个对应的退出条件，并且每个业务调用都要有超时。

3.2 重试策略：指数退避与抖动

重试是循环工程最容易出错的地方。第一版往往写成“失败后等 1 秒再重试”甚至“立即重试”。低并发场景勉强能跑，一旦任务数量多，所有失败任务同时重试，会对下游造成二次冲击。

指数退避的公式是：

delay = base * factor ^ attempt

base=1, factor=2时，第 1 次重试延迟 2 秒，第 2 次 4 秒，第 3 次 8 秒。再加 10% 抖动，就是delay + random(0, delay*0.1)，最终落在 2.0~2.2 秒、4.0~4.4 秒之间。

为什么要加抖动？想象 100 个任务同时失败，如果大家都等 8 秒，第 8 秒时 100 个请求会同时在同一个瞬间打到下游，这叫“惊群效应”。加一点随机抖动，重试请求会像下雨一样均匀散开。这个技巧在分布式系统里几乎是标配。

另一个必须设置的参数是最大重试次数。我不建议无限重试，因为如果是代码 Bug 或依赖彻底故障，无限重试只会刷爆日志、打死下游。业务场景重试 3~5 次通常足够，超过上限应该进入死信或者告警，让运维介入。

3.3 幂等与去重：避免重复处理的两种手段

重试必然带来一个问题：重复执行。任务处理到一半网络超时，下游可能已经处理成功，但客户端认为失败并重试，同一任务被执行两次。解决思路有两个方向。

一是消费端幂等。无论同一个任务被投递多少次，最终效果都一样。比如文件处理，处理前先记录文件 hash 到已处理列表，遇到相同 hash 直接跳过。或者给每个任务生成唯一request_id，下游据此去重。

二是状态判断。任务数据落库时，用状态机PENDING/RUNNING/SUCCESS/FAILED/DEAD标记。处理前查状态，如果是SUCCESS，就不再处理。数据库唯一约束也可以做到天然去重。

我的实战项目里，把文件移动到done目录就是最简单的幂等：同一个文件处理成功后不可能再出现在inbox，所以天然不会重复。如果是消息队列场景，就必须面对“至少一次语义”下的幂等性问题，这一点在第 5 章我再展开讲。

3.4 背压与限流：循环不能无脑跑

循环每分钟能处理 100 个任务，但生产者每分钟扔进来 1000 个，会发生什么？内存队列达到maxsize后，put会阻塞。我在引擎里把asyncio.Queue的maxsize设为 100，这就是一个最简单的背压机制：队列满了，生产者协程会被阻塞，进而迫使上游减速。

但要注意，如果生产者和消费者在同一个进程里，queue.put阻塞整个生产者循环，可能导致目录监听停止。更合理的做法是：监听循环只负责发现文件并入队，入队时用put_nowait，一旦队列满就记录告警，而不是无限阻塞。或者把“队列深度”作为核心告警指标，超过阈值就通知运维扩容。

并发数也不是越大越好。每个任务如果都要占用一个数据库连接，并发数设为 100 可能直接打爆连接池。建议从下游能力的 50% 起步，再压测调整。我代码里的max_concurrency=4是保守默认值，生产环境建议根据实测调整。

4. 可观测性：让循环“看得见”

4.1 打点与日志规范

没有日志的循环等于盲飞。我踩过最深的坑是生产环境任务积压了，代码却没有任何输出，只能靠猜。从那以后，我给所有循环操作定了三条日志规矩：

• 任务进入循环时打一条 info，记录任务 ID、来源、时间；
• 任务成功或失败时各打一条日志，记录耗时、尝试次数、异常信息；
• 循环本身的生命周期（启动、停止、退出）必须有日志。

日志格式建议用结构化形式，方便采集分析：

2025-01-15 10:00:01 INFO worker-0 task=/data/inbox/a.json attempt=1 status=running 2025-01-15 10:00:02 INFO worker-0 task=/data/inbox/a.json attempt=1 status=success duration_ms=980 2025-01-15 10:00:03 WARNING worker-1 task=/data/inbox/b.json attempt=2 status=failed error="timeout"

不要小看这条规范，出问题的时候可以快速定位是哪个任务、哪个 worker、哪次尝试。

4.2 三类指标

日志是事后排查用的，指标是实时监控用的。最简单的指标至少要有三个：

不用一上来就上 Prometheus，先用代码里的计数器打印指标也可以。核心思想是：指标必须反映循环的健康度，而不是好看的数字。

4.3 断点续跑与恢复

进程崩溃了，重启后怎么办？如果文件还在inbox，重新扫描就能处理。但如果任务已经出队正在执行中，崩溃后这个任务既不在队列里，也不在原始位置，就会丢失。

解决方案是给任务落一个持久化状态。数据库表或本地 JSON 文件都可以，记录这些字段：

task_id, file_path, status, attempts, next_retry_time, updated_at

每次状态变化时更新。启动时先扫描数据库，把所有status=RUNNING的任务更新为PENDING并重新入队。为什么？因为崩溃时 RUNNING 任务是否真的完成了不确定，为了不丢数据宁可重跑一次，配合幂等逻辑保证最终一致。

这个“状态机 + 启动恢复”模式，是 Loop Engineering 里最核心的韧性设计。只要符合“长周期、多并发、可失败”特征的循环任务，我强烈建议配上持久化状态。

5. 常见问题与排查实录

5.1 循环吃满 CPU：怎么定位死循环

现象是进程启动后 CPU 直接 100%，日志却没有任何输出。大多数情况是循环里有一个没有await的死等：

while True: if queue.empty(): continue # 没有 await，事件循环被占死 task = queue.get() process(task)

这个continue没有让出控制权，事件循环里的其他协程全部饿死。解法很简单：用await asyncio.sleep(0.01)让出 CPU，或者直接用await queue.get()阻塞等待。我习惯用asyncio.wait_for(queue.get(), timeout=0.5)，既能避免空转，又能定期检查停止标志。

如果已经线上出了问题，怎么定位？用py-spy dump --pid <pid>可以看到当前 Python 进程的线程栈，直接找到死循环所在行。这个工具是救命稻草，比pstack好用得多。

5.2 消息重复消费怎么解决

这个和第 3 章的幂等有关，但我再讲一个实战案例。之前做一个订单同步服务，从消息队列消费消息，调用对方接口更新订单状态。网络超时很常见——对方接口可能已经更新成功，但返回超时，我们的重试逻辑会把同一消息再发一次，结果就出现重复更新。

当时给的方案是：每个请求生成request_id = md5(order_id + event_type)，下游接口按request_id去重；消费端在本地 Redis 里存最近处理过的 request_id，处理前检查，成功后写入，过期时间设为 1 小时。即使消息重复投递，也不会重复更新。

后来更严谨的做法是消费端状态检查：处理前查数据库，如果订单已经处于目标状态，直接确认消费并跳过。所以，重复消费的终极解法永远是两个词：业务幂等 + 状态判断。

5.3 优雅关闭失效：退出卡住不动

信号触发了，stop_event也 set 了，但进程就是不退出。通常有三种原因：

1. worker 卡在某个不可中断的同步 I/O 里，比如 socketrecv或同步文件读，asyncio.wait_for能取消协程，但中断不了同步调用；
2. 还有非 daemon 线程或者没清理干净的asyncio.Task在阻挡事件循环退出；
3. 重试队列时间判断写错，导致等待时间无限长。

排查顺序：先打印asyncio.all_tasks()看所有 Task 状态；再检查是否有同步阻塞操作，有就改为await asyncio.to_thread(...)；最后给stop()加一个总超时，超时后强制取消：

try: await asyncio.wait_for(asyncio.gather(*self._workers), timeout=10) except asyncio.TimeoutError: logger.warning("force shutdown after timeout") for w in self._workers: w.cancel()

强制取消不完美，但好过进程永远退不掉。

5.4 常见问题速查表

6. 扩展与心得

6.1 从单机循环到分布式循环

单机 asyncio 版本只是一个起点，但 Loop Engineering 的思想在分布式系统里完全适用。把内存队列换成 Redis Stream 或 Kafka，把_workers换成多个服务实例，把本地状态机换成数据库状态机，核心依然是：可终止、可重试、幂等、可观测、可恢复。循环从来不是某个进程里的while，而是整个数据流里的“闭环处理”。

如果要扩展到多实例，优先选择一致性更强的队列组件，并为每个任务设置全局唯一 ID。消费端使用手动确认模式，处理成功后再确认，这样即使消费过程中崩溃，未确认消息也会被重新投递。这和我们单机代码里“成功之后才task_done”是同一个道理。

6.2 我的几点实践经验

最后分享几条我自己的习惯，都是踩坑踩出来的：

• 所有循环必须有名字。协程一多，没名字的 Task 根本分不清谁是谁。asyncio.create_task(coro, name="worker-0")带来的调试便利远超命名成本。
• 重试必须有上限，上限必须告警。重试的目的是扛住瞬时抖动，不是掩盖系统故障。连续失败超过阈值，日志必须出现 ERROR 级别内容。
• 循环里的状态变化都要有记录。不要只在成功和失败时打日志，排队、重试、进入死信这些中间态也要保留，否则没法复盘。
• 先保证能停止，再谈别的。一个不能按信号退出的循环，迟早让你在发布时崩溃。我写任何循环的第一版，都会先测试退出信号。

上面这套代码，你完全可以直接抄下来改着用：把_process换成你的业务逻辑，把inbox/done/dead换成队列或表，把退避参数调一调，就是一个能上生产的循环引擎。Loop Engineering 说起来玄，做起来其实就是把这些细节一个个抠到位。希望这篇保姆级教程能帮你少踩几个我当年踩过的坑。