Python如何实现内存超阈值时暂停执行、内存恢复后从断点续跑？

你可以通过封装检测逻辑避免重复编写冗余代码，以下两种方案都能实现「断点续跑」的效果，不会重复执行已经完成的步骤：

方案1：装饰器封装（推荐） #

将内存检测逻辑封装为装饰器，给每个执行步骤的函数加上装饰器即可，检测逻辑完全复用：

import psutil
import time
import functools

# 配置项
MEMORY_LIMIT = 60  # 内存使用率阈值，单位%
CHECK_INTERVAL = 30  # 内存不足时的检测间隔，单位秒

def memory_check(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        # 执行函数前先检测内存，不足就等待
        while psutil.virtual_memory().percent > MEMORY_LIMIT:
            time.sleep(CHECK_INTERVAL)
        return func(*args, **kwargs)
    return wrapper

# 业务步骤拆分，每个步骤加装饰器
@memory_check
def step1():
    # 对应原逻辑的 do this stuff
    pass

@memory_check
def step2():
    # 对应原逻辑的 and this other stuff
    pass

@memory_check
def step3():
    # 对应原逻辑的 and this other ...
    pass

if __name__ == "__main__":
    while True:
        step1()
        step2()
        step3()

该方案结构清晰，后续新增步骤只需要加装饰器即可，不需要修改检测逻辑。

方案2：生成器yield实现轻量拆分 #

如果不想拆分太多独立函数，可以用yield拆分执行步骤，统一在迭代环节做内存检测：

import psutil
import time

MEMORY_LIMIT = 60
CHECK_INTERVAL = 30

def task_steps():
    # do this stuff
    yield
    # and this other stuff
    yield
    # and this other ...
    yield

if __name__ == "__main__":
    while True:
        for _ in task_steps():
            while psutil.virtual_memory().percent > MEMORY_LIMIT:
                time.sleep(CHECK_INTERVAL)

该方案更轻量，适合短脚本快速改造，每写完一段逻辑加一个yield即可。

• 这类需求属于业务侧的资源调度逻辑，不属于通用编程语言的核心能力范畴：不同场景的阈值、检测间隔、暂停策略（是否要释放资源、是否告警、是暂停全局还是单个线程）差异极大，没有通用的标准实现可以覆盖所有需求。
• 要实现全局触发式暂停，需要依赖操作系统的信号中断机制，跨Windows、macOS、Linux的信号兼容实现复杂度很高，还容易和业务自行注册的信号处理逻辑冲突。
• 该类需求完全可以通过上层封装的工具逻辑实现，放到Python核心解释器中只会增加不必要的冗余复杂度。

内容的提问来源于stack exchange，提问作者Aguaraguazu