Python脚本如何正确实现多进程利用多核CPU加速特征向量生成

1. 多线程无提速的核心原因：CPython存在GIL全局解释器锁，同一时刻仅允许一个线程执行Python字节码。你的任务属于纯CPU密集型计算（逐行匹配、计数均为CPU运算，无大量网络/磁盘IO等待场景），多线程无法实现真正的并行执行，只会反复切换上下文增加额外开销，因此表现为所有核心负载偏低、处理速度无明显提升。
2. 现有代码存在明确逻辑与性能问题：

• 线程分配逻辑完全错误：判断条件使用if/elif/if/else结构，当计数值为4的倍数时，会同时启动t4、t2两个线程处理同一个文件，造成无意义的重复计算；且未做并发数控制，6000余个文件会瞬间触发数千个线程创建，系统资源大量消耗在线程调度上，无法用于实际计算。
• 核心计算逻辑效率极低：每次匹配到行内容后，都通过列表.index()方法查找位置，列表的成员判断和索引查找均为O(n)时间复杂度，相当于每一行每一类特征都要遍历两次列表，单文件处理耗时被人为拉高；另外每处理一行就重复给特征向量末尾赋值家族标签，属于完全冗余的操作。
• 路径处理存在兼容性问题：直接用字符串拼接文件路径，在不同操作系统下容易出现路径解析错误，应使用标准库的路径方法处理。

CPU密集型任务要实现多核并行，必须使用多进程而非多线程：多进程会启动多个独立的Python解释器实例，每个实例持有独立的GIL，可真正将任务负载分摊到所有CPU核心上。
实现时直接使用标准库concurrent.futures.ProcessPoolExecutor即可，无需手动管理进程创建、分配、回收逻辑，组件内置的调度机制会自动均衡负载，默认进程数等于设备CPU逻辑核心数，可最大化硬件利用率。
完整实现代码如下，同时对原有计算逻辑做了性能优化：

import os
import csv
from concurrent.futures import ProcessPoolExecutor, as_completed

# -------------------------- 提前初始化的全局配置，替换为实际值 --------------------------
script_path = os.path.dirname(os.path.abspath(__file__))
malware_families = set()  # 原恶意家族列表，转成set提升成员判断速度
feature_ls = []           # 原特征名列表
unique_ls = []            # 原每个特征对应的唯一值列表
# ----------------------------------------------------------------------------------------

# 预处理：将每个特征的唯一值列表转为 {值:索引} 的字典，查找复杂度从O(n)降到O(1)
unique_idx_map = []
for u_ls in unique_ls:
    idx_map = {val: idx for idx, val in enumerate(u_ls)}
    unique_idx_map.append(idx_map)


def generate_feature_vector(task_tuple):
    """核心特征生成函数，输入为单文件任务参数，返回生成的特征向量"""
    file_path, file_name, family = task_tuple
    # 初始化特征向量
    feature_vector = []
    for i in range(len(feature_ls)):
        feat_col = [0] * len(unique_ls[i])
        feat_col[-1] = family  # 家族标签仅需赋值一次，无需逐行重复写入
        feature_vector.append(feat_col)

    # 逐行读取文件处理，避免全量读入占用过多内存
    with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            # 遍历特征组匹配计数
            for i in range(len(feature_ls)):
                if line in unique_idx_map[i]:
                    idx = unique_idx_map[i][line]
                    feature_vector[i][idx] += 1
    return (file_name, family, feature_vector)


def find_all_files():
    """遍历目录收集所有待处理文件的任务参数"""
    task_list = []
    output_path = os.path.join(script_path, "output")
    if not os.path.exists(output_path):
        return task_list
    fam_dirs = os.listdir(output_path)
    for fam_dir in fam_dirs:
        if fam_dir not in malware_families:
            continue
        fam_path = os.path.join(output_path, fam_dir)
        if not os.path.isdir(fam_path):
            continue
        for file in os.listdir(fam_path):
            file_path = os.path.join(fam_path, file)
            if os.path.isfile(file_path):
                task_list.append((file_path, file, fam_dir))
    return task_list


def main():
    # 收集所有待处理任务
    all_tasks = find_all_files()
    print(f"共收集待处理文件：{len(all_tasks)}个")
    results = []

    # 启动进程池执行任务，默认进程数=CPU逻辑核心数，可通过max_workers参数手动指定
    # 例：i5-8300H（4核8线程）可设max_workers=8，i9-10900（10核20线程）可设max_workers=20
    with ProcessPoolExecutor() as executor:
        future_map = {executor.submit(generate_feature_vector, task): task for task in all_tasks}
        # 按完成进度收集结果
        for idx, future in enumerate(as_completed(future_map), 1):
            res = future.result()
            results.append(res)
            if idx % 100 == 0:
                print(f"处理进度：{idx}/{len(all_tasks)}")

    # 汇总结果写入CSV，可根据实际需求调整输出格式
    with open("feature_result.csv", "w", encoding="utf-8", newline="") as csvf:
        writer = csv.writer(csvf)
        # 写入表头
        header = ["file_name", "family"] + feature_ls
        writer.writerow(header)
        # 逐行写入特征
        for file_name, family, feat_vec in results:
            row = [file_name, family]
            for feat_col in feat_vec:
                row.extend(feat_col[:-1])  # 最后一位为家族标签，已单独存储无需重复写入
            writer.writerow(row)
    print("所有特征生成完成，结果已写入feature_result.csv")


if __name__ == "__main__":
    main()

• 多进程代码必须放在if __name__ == "__main__":块内执行，否则Windows系统下会出现递归启动子进程的报错。
• 不要手动通过取模方式分配任务，进程池内置的任务调度机制已经做了最优的负载均衡，可避免任务分配不均、重复计算、进程数过多导致的调度开销问题。
• 并发数不是越大越好，默认设置为CPU逻辑核心数即可，设置过高会因进程切换、磁盘IO争抢反而降低处理速度。
• 提前将列表查找转为字典查找，单文件处理速度可提升3-10倍，优化收益远高于并发调整。

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