Python脚本如何限制运行CPU占用

wen python案例 28

本文目录导读:

Python脚本如何限制运行CPU占用

  1. 使用 time.sleep() 主动让出CPU
  2. 使用 psutil 限制进程优先级
  3. 使用 resource 模块限制CPU时间 (Unix)
  4. 使用 multiprocessing 控制CPU核心数
  5. 使用任务调度手动控制
  6. 在Linux中使用 cpulimit 工具
  7. 使用 signal 实现定时让步
  8. 专业方案:使用 setproctitle + 调度
  9. 推荐方案

限制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() 是最简单有效的方法
  • 生产环境:使用 cpulimitnice 命令
  • 精细控制:结合 psutil 实现动态调节
  • 多进程:使用 multiprocessing 控制并发数

根据你的具体需求(临时限制还是持续限制、精度要求、操作系统)选择合适的方案。

抱歉,评论功能暂时关闭!