Python并行处理代码运行慢排查及multiprocessing相关接口使用咨询

• 核心性能问题：进程启动后立即调用Q.get()导致串行执行。循环中每启动1个进程就阻塞等待队列返回结果，4个进程完全按顺序执行，没有任何并行效果，这是耗时过长的根本原因。
• 逻辑报错问题：进程对象存储逻辑错误。你把队列返回的筛选结果存入procs列表，后续调用p.join()时遍历的是数字列表，会直接抛出属性错误，代码实际无法正常运行结束。
• 其他细节问题：缺失计时起始点，仅定义了end = time.time()没有对应的start = time.time()，无法统计实际耗时；预定义的dictionary变量未使用，筛选逻辑硬写[1,2,3]属于冗余代码；Windows系统下没有加if __name__ == '__main__'保护，会触发无限启动进程的报错。

import time
import numpy as np
import multiprocessing

data = [1,2,3,4,10,2,3,4,5,6,7,8,9,10,1,2,3,11,12,1,2,3,100,101]
dictionary = [1,2,3]
data_split = np.array_split(data,4)
Q = multiprocessing.Queue()

def recog_func(data_chunk, valid_values, queue):
    result = [w for w in data_chunk if w in valid_values]
    queue.put(result)

if __name__ == '__main__':
    start = time.time()
    procs = []
    # 第一步：先启动所有进程，实现真正并行
    for s in data_split:
        p = multiprocessing.Process(target=recog_func, args=(s, dictionary, Q))
        p.start()
        procs.append(p) # 存储进程对象而非结果
    # 第二步：收集所有进程的返回结果
    final_result = []
    for _ in range(len(procs)):
        final_result.extend(Q.get())
    # 第三步：统一等待进程资源释放，避免僵尸进程
    for p in procs:
        p.join()
    end = time.time()

    print("筛选结果：", final_result)
    print("耗时：", end - start)

注：当前示例的数据集过小，多进程的创建、通信开销会大于并行收益，实际业务中只有待处理数据量较大时使用多进程才能获得耗时优化

Process 核心使用逻辑 #

• 实例创建：传入target为子进程要执行的函数，args为函数参数元组，单参数需要加逗号避免语法解析错误。
• 启动流程：所有进程先全部调用start()启动，不要启动一个就等待结果，否则失去并行意义。
• 回收流程：所有进程启动完成后，先收集执行结果，再统一调用join()等待进程退出释放资源，避免产生僵尸进程。
• 平台兼容：Windows系统下多进程代码必须放在if __name__ == '__main__'代码块内，否则会触发无限递归启动进程的报错。

Queue 核心使用逻辑 #

• 特性：Queue是进程安全的通信组件，多进程同时读写不需要额外加锁，内部已经实现了同步机制。
• 读写规则：put()用于向队列写入数据，get()用于从队列读取数据，默认是阻塞模式，队列为空时get()会一直等待直到有数据写入。
• 数量匹配：启动N个进程写入队列，就要对应调用N次get()读取队列，避免遗漏数据或者程序永久阻塞。
• 适用场景：适合传递小体积的结果数据，超大量数据建议用共享内存、管道等组件降低序列化开销。

内容的提问来源于stack exchange，提问作者유한정