多进程中全局只读对象的读取开销及最优并行实现方案问询

场景描述 #

处理由自定义类对象组成的海量内存全局数据时，采用多进程并行处理，每个子进程通过索引读取全局数据的文本内容进行计算且不修改数据。

代码示例 #

import concurrent
import numpy as np

data_size = 1_000_000

class DataClass:
    def __init__(self, text):
        self.text = text

def process_text(dataset_idx):
    return dataset[dataset_idx].text.lower()

dataset = [DataClass('SOME TEXT') for _ in range(data_size)]
dataset_indices_to_process = range(data_size)

results = []
with concurrent.futures.ProcessPoolExecutor() as executor:
    for result in executor.map(process_text, dataset_indices_to_process ):
        results.append(result)

问题与解答 #

1. 子进程读取该全局对象时是否会因锁机制产生额外开销？ #

不会。核心原因有两点：

• Python多进程模式下，子进程是通过拷贝父进程内存空间创建的（Unix用fork，Windows用spawn），每个子进程都持有独立的数据集副本，不存在多进程共享同一块内存的情况，自然不需要锁来保护数据。
• 你的场景里数据是只读的，哪怕用共享内存方案，纯读操作也不会触发锁竞争——锁的作用是防止写操作导致数据不一致，只读场景下根本不需要锁。

但要注意：如果用fork创建子进程，海量数据的拷贝会带来巨量内存开销——每个子进程都复制一份100万条数据的列表，内存占用会飙升到父进程的N倍（N是进程数），这反而会成为性能瓶颈。

2. 针对这类只读全局数据，最优的并行化实现方式是什么？ #

根据数据规模和操作系统，推荐三种针对性方案：

方案一：共享内存存储（适合超大规模数据，内存无法承受多份拷贝的场景）

把只读数据放到共享内存区域，避免每个子进程拷贝完整数据集。可以用multiprocessing.shared_memory模块存储序列化后的文本数据，子进程通过共享内存地址直接读取，无需拷贝。
示例思路：

from multiprocessing import shared_memory
import concurrent.futures

data_size = 1_000_000
text = b'SOME TEXT'
# 创建共享内存，存储所有文本（实际场景可序列化所有不同文本）
shm = shared_memory.SharedMemory(create=True, size=len(text)*data_size)
shm.buf[:len(text)*data_size] = text * data_size

def process_text(idx):
    start = idx * len(text)
    end = start + len(text)
    return shm.buf[start:end].decode().lower()

dataset_indices_to_process = range(data_size)
results = []
with concurrent.futures.ProcessPoolExecutor() as executor:
    for result in executor.map(process_text, dataset_indices_to_process):
        results.append(result)

# 清理共享内存
shm.close()
shm.unlink()

方案二：数据分块处理（适合数据规模适中，内存可承受少量拷贝的场景）

把数据集分成若干块，每个子进程处理一整块数据，而非传递索引让子进程读取全局数据。这种方式能减少进程间的索引传递开销，同时如果用spawn模式（Windows），还能避免全局数据被重复初始化。
示例思路：

import concurrent.futures

data_size = 1_000_000

class DataClass:
    def __init__(self, text):
        self.text = text

def process_chunk(chunk):
    return [item.text.lower() for item in chunk]

dataset = [DataClass('SOME TEXT') for _ in range(data_size)]
# 分块，比如每块10000条
chunk_size = 10000
chunks = [dataset[i:i+chunk_size] for i in range(0, data_size, chunk_size)]

results = []
with concurrent.futures.ProcessPoolExecutor() as executor:
    for chunk_result in executor.map(process_chunk, chunks):
        results.extend(chunk_result)

方案三：多线程替代多进程（适合计算量小的场景）

如果你的文本处理逻辑（比如lower()）计算量很小，多进程的进程创建、内存拷贝开销反而会盖过并行收益。这时用ThreadPoolExecutor更合适——线程共享同一内存空间，不需要拷贝数据集，也没有进程间通信的额外开销。
示例：

import concurrent.futures

data_size = 1_000_000

class DataClass:
    def __init__(self, text):
        self.text = text

def process_text(dataset_idx):
    return dataset[dataset_idx].text.lower()

dataset = [DataClass('SOME TEXT') for _ in range(data_size)]
dataset_indices_to_process = range(data_size)

results = []
with concurrent.futures.ThreadPoolExecutor() as executor:
    for result in executor.map(process_text, dataset_indices_to_process):
        results.append(result)

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