本文目录导读:

- 使用
time.sleep()主动让出CPU - 使用
psutil限制进程优先级 - 使用
resource模块限制CPU时间 (Unix) - 使用
multiprocessing控制CPU核心数 - 使用任务调度手动控制
- 在Linux中使用
cpulimit工具 - 使用
signal实现定时让步 - 专业方案:使用
setproctitle+ 调度 - 推荐方案
限制Python脚本的CPU占用,有以下几种常用方法,从简单到复杂:
使用 time.sleep() 主动让出CPU
最基础的方法,在循环中主动休眠:
import time
while True:
# 你的计算任务
result = heavy_computation()
time.sleep(0.01) # 让出CPU 10ms
使用 psutil 限制进程优先级
降低进程的优先级,让系统优先分配CPU给其他进程:
import psutil import os # 获取当前进程 process = psutil.Process(os.getpid()) # 设置优先级 (Unix系统) process.nice(19) # 最低优先级 # Windows系统 process.nice(psutil.BELOW_NORMAL_PRIORITY_CLASS)
使用 resource 模块限制CPU时间 (Unix)
import resource
import signal
# 设置CPU时间限制(秒)
resource.setrlimit(resource.RLIMIT_CPU, (1, 2)) # 软限制1秒,硬限制2秒
# 设置信号处理
def timeout_handler(signum, frame):
raise TimeoutError("CPU时间超限")
signal.signal(signal.SIGXCPU, timeout_handler)
try:
# 你的计算密集型代码
while True:
pass
except TimeoutError:
print("已限制CPU使用")
使用 multiprocessing 控制CPU核心数
from multiprocessing import Pool, cpu_count
import time
def worker(n):
# 计算任务
result = sum(i * i for i in range(n))
return result
# 只使用CPU核心数的50%
cpu_usage = max(1, cpu_count() // 2)
with Pool(processes=cpu_usage) as pool:
results = pool.map(worker, range(1000))
使用任务调度手动控制
创建自己的CPU占用控制器:
import time
import psutil
class CPULimiter:
def __init__(self, max_cpu_percent=50):
"""
max_cpu_percent: 最大CPU占用百分比
"""
self.max_cpu = max_cpu_percent
self.process = psutil.Process()
def check_and_wait(self):
"""检查CPU占用,如果超过限制则等待"""
cpu_percent = self.process.cpu_percent(interval=0.1)
if cpu_percent > self.max_cpu:
# 计算需要等待的时间
wait_time = (cpu_percent - self.max_cpu) / 100.0 * 0.5
time.sleep(wait_time)
return True
return False
# 使用示例
limiter = CPULimiter(max_cpu_percent=30)
while True:
# 你的计算任务
for i in range(1000000):
_ = i * i
# 检查并限制CPU
limiter.check_and_wait()
在Linux中使用 cpulimit 工具
外部工具限制Python进程:
# 安装cpulimit sudo apt-get install cpulimit # 运行Python脚本,限制CPU占用为50% python your_script.py & cpulimit -e python -l 50 # 限制所有python进程为50%
使用 signal 实现定时让步
import signal
import time
class CPUSaver:
def __init__(self, work_ratio=0.5):
"""
work_ratio: 工作时间的比例 (0-1)
"""
self.work_time = 0
self.work_ratio = work_ratio
def run_with_limit(self, func, *args, **kwargs):
"""在限制CPU的情况下执行函数"""
start = time.time()
result = func(*args, **kwargs)
elapsed = time.time() - start
# 计算需要休眠的时间
sleep_time = elapsed * (1 - self.work_ratio) / self.work_ratio
time.sleep(sleep_time)
return result
# 使用示例
saver = CPUSaver(work_ratio=0.3) # 只使用30%的CPU
def heavy_task(n):
return sum(i ** 2 for i in range(n))
for _ in range(10):
result = saver.run_with_limit(heavy_task, 1000000)
专业方案:使用 setproctitle + 调度
import time
import threading
from setproctitle import setproctitle
class CPUBalancer:
def __init__(self, max_cpu_usage=50):
self.max_cpu = max_cpu_usage
self.is_running = True
self.thread = threading.Thread(target=self._monitor)
def _monitor(self):
"""监控并调整CPU使用"""
while self.is_running:
# 获取当前CPU使用率(需要psutil)
import psutil
cpu_percent = psutil.cpu_percent(interval=1)
if cpu_percent > self.max_cpu:
# CPU过高,让当前线程休眠
time.sleep(0.5)
else:
# CPU正常,继续工作
time.sleep(0.1)
def start(self):
self.thread.start()
def stop(self):
self.is_running = False
# 使用
balancer = CPUBalancer(max_cpu_usage=40)
balancer.start()
# 你的主程序
while True:
# 计算密集型任务
pass
balancer.stop()
推荐方案
- 简单场景:使用
time.sleep()是最简单有效的方法 - 生产环境:使用
cpulimit或nice命令 - 精细控制:结合
psutil实现动态调节 - 多进程:使用
multiprocessing控制并发数
根据你的具体需求(临时限制还是持续限制、精度要求、操作系统)选择合适的方案。