工作流中断通常不是孤立的异常，而是错误处理缺位引发的连锁反应。比如一个包含下单、扣款、发货三个节点的电商处理流程，扣款成功后发货节点超时，如果直接抛出异常，整个任务会停留在已扣款未发货的中间态，重跑又可能再次扣款。要解决这类问题，必须同时做好错误分类、重试策略和断点续传设计。本文会从这三个层面展开，并给出可落地的状态机示例。
一、先分清错误类型：不是所有失败都适合重试
工作流中的异常可以粗略分成三类。第一类是瞬时故障，例如网络抖动、第三方接口偶尔超时、数据库锁等待，这类错误通常在几百毫秒到几秒内自行恢复。第二类是业务异常，比如库存不足、用户状态不合法、参数校验失败，这类错误重试没有意义，需要直接终止或走人工处理。第三类是系统性故障，比如配置错误、服务不可用、磁盘写满，这类错误可能持续较长时间，需要熔断和告警。
把错误分类之后，策略才清晰。瞬时故障适合自动重试，业务异常适合记录原因并跳过或终止，系统性故障适合快速失败并通知运维。很多工作流框架默认对所有异常都采用同一种处理方式，这是导致重复执行和脏数据的常见原因。可以定义一个异常基类，并让瞬时异常继承自 RetryableError，业务异常继承自 NonRetryableError，这样错误处理层就能根据类型自动决策。
例如在一个异步任务系统中，调用支付网关超时属于瞬时故障，但如果返回的是余额不足错误码，就属于业务异常。两者的处理路径完全不同：前者进入重试队列，后者直接写入失败表并触发用户通知。没有这种区分，重试队列会被永远无法成功的任务占满，真正可恢复的任务反而被阻塞。
二、错误处理策略：重试、退避与熔断如何配合
重试是解决瞬时故障最直接的手段，但直接无脑重试可能放大故障。例如下游服务已经过载，大量客户端同时立即重试会把服务进一步打垮。因此需要在重试中加入指数退避和随机抖动。指数退避让每次重试的间隔逐渐增加，例如1秒、2秒、4秒，随机抖动则避免多个客户端在同一时刻集体重试。下面的代码展示了一个带退避的重试装饰器。
import time
import random
def retry_with_backoff(func, max_retries=3, base_delay=1):
for attempt in range(max_retries):
try:
return func()
except TransientError as e:
if attempt == max_retries - 1:
raise
delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
time.sleep(delay)
单纯重试还不够，当某个依赖服务持续失败时，应该启用熔断机制。熔断器在一段时间内快速拒绝请求，给下游恢复时间，同时也让上游线程不必长时间等待超时。熔断状态通常有三种：关闭、打开、半开。关闭状态正常放行；失败率超过阈值后切换到打开状态，直接抛出熔断异常；经过冷却时间后进入半开状态，允许少量探测请求，如果探测成功则恢复关闭，失败则重新打开。
此外还可以设置降级路径。比如主流程依赖推荐服务，推荐服务不可用时，可以返回默认列表或基于缓存的兜底结果，保证核心流程不被旁路服务拖垮。错误处理的目标不是消除所有失败，而是控制失败的影响范围，让系统在部分依赖异常时仍然能够提供可用的服务。
三、断点续传的核心：状态持久化与幂等写入
断点续传的目标是让工作流从失败节点继续执行，而不是从头重跑。要做到这一点，每个节点的完成状态必须持久化到外部存储。可以设计一张检查点表，记录任务ID、当前步骤、状态、重试次数和更新时间。每次节点成功执行后立即更新检查点，即使进程崩溃，也能从表中恢复出最后成功的位置。
CREATE TABLE workflow_checkpoint (
task_id VARCHAR(64) PRIMARY KEY,
current_step VARCHAR(32) NOT NULL,
status VARCHAR(16) NOT NULL,
retry_count INT DEFAULT 0,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
状态持久化只是基础，更关键的是每个节点必须幂等。假设扣款节点执行成功后更新检查点之前进程崩溃，恢复时系统会重跑扣款节点，如果扣款接口不具备幂等性，就会重复扣款。解决方法是在调用外部接口时传入唯一的幂等键，例如任务ID加步骤名，下游根据幂等键去重。或者在节点内部先查询是否已经完成，再决定是否执行。
另一种常见做法是使用事务性发件箱或本地消息表。在业务数据写入的同时，将需要执行的下一步任务写入数据库，通过后台任务扫描并触发，这样业务操作和任务状态的更新可以放在同一个本地事务中，避免不一致。断点续传的难点不在于保存状态，而在于如何保证状态更新与业务操作之间的原子性。
四、完整示例：基于状态机实现可恢复工作流
下面给出一个简化的工作流执行器，它根据持久化的状态判断当前应该执行哪一步。状态机包含五个状态：待处理、步骤A完成、步骤B完成、步骤C完成、整体完成。任务每次被调度时，从状态存储中读取当前状态，只执行当前状态对应的后续步骤，并在步骤成功后更新状态。如果某一步失败，状态不会推进，下次调度会从失败步骤重试。
from enum import Enum
class WorkflowState(Enum):
PENDING = 0
STEP_A_DONE = 1
STEP_B_DONE = 2
STEP_C_DONE = 3
COMPLETED = 4
def execute_workflow(task_id, state_store):
state = state_store.get(task_id, WorkflowState.PENDING)
if state == WorkflowState.PENDING:
do_step_a(task_id)
state_store.save(task_id, WorkflowState.STEP_A_DONE)
state = WorkflowState.STEP_A_DONE
if state == WorkflowState.STEP_A_DONE:
do_step_b(task_id)
state_store.save(task_id, WorkflowState.STEP_B_DONE)
state = WorkflowState.STEP_B_DONE
if state == WorkflowState.STEP_B_DONE:
do_step_c(task_id)
state_store.save(task_id, WorkflowState.COMPLETED)
这个设计的关键是每个步骤的入口条件是明确的状态标记，而不是依赖内存中的执行位置。即使执行进程重启，只要状态存储还在，任务就能继续。配合前面提到的重试和幂等控制，就可以构建一个相对可靠的工作流恢复机制。对于更复杂的场景，还可以把节点抽象成有向无环图，用状态表记录每个节点的完成情况，恢复时从所有未完成节点中找出可执行的节点继续调度。
实际生产环境中还需要考虑并发执行的风险。同一个任务可能被多个调度器实例同时拉取，导致同一节点被重复执行。解决方式是在状态表上加乐观锁或使用分布式锁，例如在更新检查点时带上版本号，只有版本号匹配才允许更新。如果更新失败说明其他实例已经推进了状态，当前实例应该放弃本次执行并重新读取最新状态。这样可以避免两个实例同时执行同一个扣款节点。
工作流中断的治理不是单一技术点，而是错误分类、重试策略、熔断降级、状态持久化和幂等控制共同作用的结果。可以先从错误分类和检查点表入手，逐步引入退避重试与幂等键，再根据实际流量增加熔断和并发控制。任何可靠的工作流引擎，本质上都是让每一步都可被追踪、可被恢复、可被安全重复执行。