如何根据CPU使用率动态创建线程以加速Python脚本执行？

问题背景 #

需要处理大型URL列表，每个URL需执行网络请求与HTML解析任务，希望根据系统实时CPU使用率（兼顾后台应用）动态调整并发线程数：后台无负载时用最大并发数提速，后台有任务时自动减少并发，避免占用过多资源。当前可手动实现多线程，需实现动态调整逻辑。

解决方案 #

核心思路 #

1. 用psutil获取系统CPU使用率：实时监控系统整体CPU负载
2. 动态计算并发数：基于CPU使用率调整线程池大小，IO密集型任务默认最大并发数设为CPU核心数的2-4倍（网络请求等待时GIL会释放，多线程效率更高）
3. 线程安全的结果收集：避免全局列表的线程安全问题，用线程池返回值或线程安全容器存储结果

实现步骤 #

1. 安装依赖

pip install psutil requests beautifulsoup4

2. 编写安全的任务函数

改造原函数，去掉全局列表，直接返回处理结果（线程安全）：

import requests
from bs4 import BeautifulSoup

def process_url(url):
    try:
        # 设置超时避免任务卡住
        r = requests.get(url, timeout=10)
        # 抛出HTTP状态码错误
        r.raise_for_status()
        soup = BeautifulSoup(r.content, 'html.parser')
        
        op_div = soup.find('div', attrs={'class': 'offer_price'})
        if not op_div:
            return None
            
        price_span = op_div.find('span', attrs={'class': 'm-w'})
        if not price_span or not price_span.text.strip():
            return None
            
        # 解析价格并转换为浮点数
        price_text = price_span.text.strip()[1:].replace(',', '')
        return float(price_text)
    except Exception:
        # 可按需添加错误日志
        return None

3. 动态计算并发数的函数

import psutil

def get_dynamic_max_workers():
    # IO密集型任务默认最大并发数为CPU核心数*4
    max_possible = psutil.cpu_count() * 4
    # 获取1秒内的CPU平均使用率
    cpu_usage = psutil.cpu_percent(interval=1)
    # CPU使用率阈值，可按需调整
    threshold = 70
    
    if cpu_usage < threshold:
        # CPU负载低，用最大并发数
        return max_possible
    else:
        # 根据负载超额比例减少并发数，最少保留2个线程
        reduce_amount = int((cpu_usage - threshold) / 10) * 2
        return max(max_possible - reduce_amount, 2)

4. 主逻辑：动态调整线程池处理任务

from concurrent.futures import ThreadPoolExecutor, as_completed

def main(url_list):
    results = []
    # 分批次处理，每批处理后重新调整并发数
    batch_size = 20
    for i in range(0, len(url_list), batch_size):
        batch_urls = url_list[i:i+batch_size]
        # 每批开始前重新计算并发数
        current_workers = get_dynamic_max_workers()
        print(f"处理批次 {i//batch_size + 1}，当前CPU使用率: {psutil.cpu_percent(interval=0)}%，调整并发数为: {current_workers}")
        
        with ThreadPoolExecutor(max_workers=current_workers) as executor:
            futures = [executor.submit(process_url, url) for url in batch_urls]
            for future in as_completed(futures):
                results.append(future.result())
    
    return results

# 测试调用
if __name__ == "__main__":
    # 替换为你的实际URL列表
    url_list = ["https://example.com/offer" + str(i) for i in range(100)]
    min_prices = main(url_list)
    valid_prices = [p for p in min_prices if p is not None]
    print(f"处理完成，共{len(min_prices)}个任务，有效价格{len(valid_prices)}个")

关键注意事项 #

• 线程 vs 进程：你的任务是IO密集型（网络请求），多线程比多进程更高效，进程切换开销远大于线程
• 线程安全：绝对不要用全局列表收集多线程结果，会出现数据竞争导致结果错乱，用线程池返回值或queue.Queue更安全
• 动态调整时机：通过分批次处理实现动态调整，因为ThreadPoolExecutor创建后无法修改最大线程数，每批任务开始前重新计算并发数
• 超时与错误处理：给网络请求加超时时间，避免单个任务长期阻塞，影响整体效率

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