GIL锁也拦不住!这个多进程方案让python速度提升400%

off999 2025-06-08 23:23 66 浏览 0 评论

引言

"你的python程序怎么卡死了？"

当产品经理第3次拍我桌子时，我终于意识到——在200万行的数据清洗任务面前，GIL锁正在把我的8核CPU变成单核计算器...

提前准备

为了查看核心使用数量，我们先安装库：

pip install psutil

使用下面代码查看核心使用情况：

import psutil


def get_active_cpu_count():
    per_cpu = psutil.cpu_percent(interval=1, percpu=True)
    active_cores = sum(1 for usage in per_cpu if usage > 20)  # 使用率超过20%认为活跃

    print(f'活跃的CPU核心数: {active_cores}')

一、GIL锁的致命困局

用多线程进行数据处理，我们通过查看活跃核心数量来判断使用核心的情况：

import os
import time
import threading


def get_data(path, event):
    while not event.is_set():
        time.sleep(0.01)
    total = 0
    with open(path, 'r') as f:
        for line in f:
            try:
                s_list = line.split(',')
                v = float(s_list[1])
                total += (v ** 100 + 20 / v + v * 3 - 1000 + (10 ** 24) - 2313.24234)
            except:
                pass
    print('任务完成')


def analyse_file():
    folder = r'D:\xxxxx'
    file_list = [os.path.join(folder, file) for file in os.listdir(folder)][:10]

    # 为了防止创建线程引起的cpu波动，使用信号来控制
    event = threading.Event()
    for path in file_list:
        t = threading.Thread(target=get_data, args=(path, event,))
        t.start()
    
    return event

if __name__ == '__main__':
    get_active_cpu_count()
    event = analyse_file()

    while True:
        event.set()
        get_active_cpu_count()
        time.sleep(1)

我使用了一个文件夹中的数据，它是如下内容：

2023-02-10 09:00:00.025,  0.00011195
2023-02-10 09:00:00.030,  0.00004691
2023-02-10 09:00:00.035,  -0.00006853
...

我取出后面的数据内容，随意进行处理，然后查看核心使用情况如下：

活跃的CPU核心数: 0
活跃的CPU核心数: 1
活跃的CPU核心数: 1
活跃的CPU核心数: 1
任务完成
任务完成
...

GIL（全局解释器锁）如同高速路上的收费站，即使开10个线程，CPU也只能交替放行。

二、核弹级提速方案

绕过GIL的真正秘诀在于跨进程内存隔离，每个 进程独立持有GIL：

import os
import time
from multiprocessing import Process, Event

def get_data(path):
    total = 0
    with open(path, 'r') as f:
        for line in f:
            try:
                s_list = line.split(',')
                v = float(s_list[1])
                total += (v ** 100 + 20 / v + v * 3 - 1000 + (10 ** 24) - 2313.24234)
            except:
                pass
    print('任务完成')


def analyse_file():
    folder = r'D:\xxxxx'
    file_list = [os.path.join(folder, file) for file in os.listdir(folder)][:10]

    event = Event()
    for path in file_list:
        process = Process(target=get_data, args=(path, event,))
        process.start()

    return event


if __name__ == '__main__':
    get_active_cpu_count()
    event = analyse_file()

    while True:
        event.set()
        get_active_cpu_count()
        time.sleep(1)

查看核心使用情况：

活跃的CPU核心数: 0
活跃的CPU核心数: 16
任务完成
...

我的内核是8核，逻辑处理器是16核，可以看到使用了多进程后能够并行处理，并将核心使用吃满。

三、性能对比

方式	耗时（秒）	CPU利用率
线程	6.186309	15%
进程	1.046666	96%

两种方式相差 5.91倍。

四、实战：百万日志文件优化方案

单进程版本

def single_process():
    with open('xxxx.log') as f:
        for line in f:  # 单线程逐行读取
            pass

多进程版本

from multiprocessing import Pool
import os

def split_file(input_file, num_parts=8):
    def line_generator(filename):
        with open(filename, 'r', encoding='utf-8') as f:
            for line in f:
                yield line

    gen = line_generator(input_file)

    output_folder = 'output'

    if not os.path.exists(output_folder):
        os.makedirs(output_folder)

    file_list = []
    for part in range(num_parts):
        output_file = os.path.join(output_folder, f'part_{part + 1}.log')
        with open(output_file, 'w', encoding='utf-8') as out_f:
            try:
                for _ in range(1000):  # 每次处理1000行
                    out_f.write(next(gen))
            except StopIteration:
                break

        file_list.append(output_file)

    return file_list

def multi_process():
    # 第一步：切分日志文件
    chunk_files = split_file('xxxx.log', 8)  # 这里可以使用多进程先将文件拆成8个子文件
    
    # 第二步：并行处理
    with Pool(8) as pool:
        partial_results = pool.map(analyze_chunk, chunk_files)
    
    # 第三步：聚合分块结果
    pass

def analyze_chunk(file):
# 分析分块文件
    results = {}
    with open(file) as f:
       pass
       
    # 将分析的结果返回
    return results