如何加速Python进程间数据交换？解决共享字典性能瓶颈

问题根源 #

你使用的multiprocessing.Manager().dict()是通过跨进程RPC（远程过程调用）实现的，每次执行num not in shared_dict或赋值操作，都需要通过管道在进程间传递数据，高频访问下这种通信开销会成为性能瓶颈。以下是几种针对性的优化方案，兼顾性能和任务重复率需求：

1. 基于共享内存位图的轻量级去重（最优性能，低重复） #

因为你的任务num范围固定为1-100000，非常适合用**位图（BitMap）**跟踪已完成任务。位图存储在共享内存中，所有进程直接操作内存，避免RPC开销，仅需一个锁保证原子性。

实现代码 #

import multiprocessing
import random

def is_processed(num, bitmap):
    byte_idx = (num - 1) // 8
    bit_idx = (num - 1) % 8
    return (bitmap[byte_idx] >> bit_idx) & 1

def mark_processed(num, bitmap, lock):
    with lock:
        byte_idx = (num - 1) // 8
        bit_idx = (num - 1) % 8
        bitmap[byte_idx] |= 1 << bit_idx

def worker(bitmap, lock):
    # 本地缓存已处理的num，减少共享内存访问
    local_processed = set()
    for _ in range(1000000):
        num = random.randint(1, 100000)
        if num in local_processed:
            continue
        if not is_processed(num, bitmap):
            # 执行计算任务
            result = num ** 2
            # 标记为已处理
            mark_processed(num, bitmap, lock)
            local_processed.add(num)

if __name__ == "__main__":
    # 100000位 = 12500字节
    bitmap = multiprocessing.Array('B', 12500, lock=False)
    lock = multiprocessing.Lock()

    processes = []
    for _ in range(5):
        p = multiprocessing.Process(target=worker, args=(bitmap, lock))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

    # 验证结果（可选）
    processed_nums = []
    for num in range(1, 100001):
        if is_processed(num, bitmap):
            processed_nums.append(num)
    print(f"已处理任务数量：{len(processed_nums)}")

优势 #

• 共享内存操作延迟远低于Manager.dict的RPC调用
• 本地缓存进一步减少共享内存访问次数，降低锁竞争
• 仅在锁释放前的极短窗口可能出现重复任务，重复率极低

2. 任务预分配+本地去重（零跨进程开销，无重复） #

如果可以预先生成所有可能的任务num，直接将任务集拆分给各个进程，每个进程处理自己的子集并本地去重，完全不需要共享数据。

实现代码 #

import multiprocessing
import random
from itertools import islice

def worker(task_subset):
    local_result = {}
    for num in task_subset:
        # 本地去重（如果子集有重复）
        if num not in local_result:
            local_result[num] = num ** 2
    return local_result

if __name__ == "__main__":
    # 生成所有可能的任务num并打乱
    all_tasks = list(range(1, 100001))
    random.shuffle(all_tasks)

    # 拆分任务给5个进程
    chunk_size = len(all_tasks) // 5
    task_chunks = [all_tasks[i*chunk_size : (i+1)*chunk_size] for i in range(5)]
    # 分配剩余任务给最后一个进程
    task_chunks[-1].extend(all_tasks[5*chunk_size:])

    # 使用进程池处理
    with multiprocessing.Pool(5) as pool:
        results = pool.map(worker, task_chunks)

    # 合并结果
    shared_dict = {}
    for res in results:
        shared_dict.update(res)
    print(f"已处理任务数量：{len(shared_dict)}")

优势 #

• 完全没有跨进程数据通信开销，性能最优
• 无任何重复任务
• 实现简单，代码可读性高

3. 本地缓存+定期同步（允许少量重复，大幅降低开销） #

如果无法预分配任务，可让每个进程维护本地已处理集合，每隔一定次数将本地集合同步到共享存储，牺牲少量重复换取性能提升。

实现代码 #

import multiprocessing
import random

def worker(shared_set, sync_interval=1000):
    local_processed = set()
    count = 0
    for _ in range(1000000):
        num = random.randint(1, 100000)
        if num in local_processed:
            continue
        # 检查共享集合（频率降低）
        if num not in shared_set:
            result = num ** 2
            local_processed.add(num)
            count += 1
            # 定期同步到共享集合
            if count >= sync_interval:
                with shared_set.get_lock():
                    shared_set.update(local_processed)
                local_processed.clear()
                count = 0
    # 同步剩余数据
    with shared_set.get_lock():
        shared_set.update(local_processed)

if __name__ == "__main__":
    # 使用Manager的set（比dict更适合去重）
    manager = multiprocessing.Manager()
    shared_set = manager.set()

    processes = []
    for _ in range(5):
        p = multiprocessing.Process(target=worker, args=(shared_set,))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

    print(f"已处理任务数量：{len(shared_set)}")

优势 #

• 大幅减少跨进程同步次数，性能比直接用共享字典提升数倍
• 仅在同步窗口内可能出现少量重复任务，可通过调整sync_interval平衡重复率和性能

4. 使用共享内存字典（Python 3.8+） #

利用multiprocessing.shared_memory创建真正的共享内存字典，避免Manager的RPC开销。需要配合锁保证进程安全。

实现思路 #

• 使用shared_memory.SharedMemory创建共享内存块
• 用pickle序列化字典到共享内存，或使用第三方库（如shared_dict）简化操作
• 配合锁保证多进程读写的原子性

注意事项 #

• 实现复杂度较高，适合对性能要求极高且任务范围不固定的场景
• 需要手动管理共享内存的创建和释放，避免内存泄漏

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