Python多进程下实时更新共享数据的实现方案咨询

Python multiprocessing 模块默认采用进程内存隔离机制，单进程下用global声明的全局变量会在每个子进程启动时拷贝一份独立副本，子进程对副本的修改完全不会同步到主进程或其他子进程，这是全局变量写法在多进程场景失效的根本原因。

可选跨进程状态实现方案对比 #

• 共享内存基础类型（multiprocessing.Value/Array）：适合存储单个数值、定长数组这类简单状态，性能开销最小，基础读写不需要额外通信，仅并发修改关联状态时需要搭配锁保证原子性。质因数分解场景需要同步的剩余待分解值、遍历上限、终止标记都属于单个数值类状态，优先选用该方案。
• 进程安全队列（multiprocessing.Queue/JoinableQueue）：适合传递动态长度的结果数据，入队出队操作本身进程安全，不需要手动维护内存扩容，适合存储动态增长的质因数结果列表。
• 代理共享对象（multiprocessing.Manager）：支持list、dict等复杂Python对象的跨进程共享，但性能比原生共享内存低30%以上，当前场景无使用必要。

质因数分解场景的实现要点 #

该算法的状态更新逻辑存在明确的单向特征：所有状态更新的触发前提是某个工作进程找到了一个质因数，此时所有工作进程都需要立刻读取新的剩余待分解值和遍历上限，一旦剩余值的平方根小于当前正在遍历的质数，所有进程需要立刻终止任务。实现时需遵循以下规则：

1. 固定状态用共享内存存储：
   • 用Value('d', 0)存储浮点型遍历上限new_upto_target
   • 用Value('L', 0)存储无符号长整型的剩余待分解值full_number_target、最后剩余值last_divisor
   • 用Value('b', False)存储布尔型终止标记found
   • 搭配1把Lock，在更新剩余值、遍历上限这组强关联状态时加锁，保证三个值的更新是原子操作，避免进程读到半更新的无效状态。
2. 动态结果用队列收集：不要尝试直接在共享内存中存储numpy数组做动态追加，numpy数组扩容需要重新分配内存，不适配跨进程共享场景，直接用Queue收集所有找到的质因数，最后由主进程统一整理排序即可，实现简单且性能更高。
3. 任务拆分逻辑：将质数列表按固定块大小切分后分给进程池的工作进程，每个工作进程处理质数块时，每次做整除判断前先读取最新的终止标记和遍历上限：
   • 如果终止标记为True，直接退出当前任务
   • 如果当前质数大于最新遍历上限，就设置终止标记，将最后剩余值推入结果队列，退出任务
   • 如果当前质数可整除剩余待分解值，统计该质数的幂次，将所有质因数推入结果队列，加锁更新剩余值、新的遍历上限后释放锁
4. 性能优化点：终止标记只会单向从False变为True，不存在反向修改，读取该标记时不需要加锁，减少不必要的锁开销。

核心改造代码示例 #

import math
import numpy as np
from multiprocessing import Pool, Value, Lock, Queue

def init_shared(shared_target, shared_upto, shared_last, shared_found, shared_lock, result_q, prime_arr):
    """进程池初始化函数，将共享状态映射为每个子进程的全局变量"""
    global full_number_target
    global new_upto_target
    global last_divisor
    global found
    global lock
    global factor_queue
    global prime_array
    full_number_target = shared_target
    new_upto_target = shared_upto
    last_divisor = shared_last
    found = shared_found
    lock = shared_lock
    factor_queue = result_q
    prime_array = prime_arr

def exponent_finder(num, input_target):
    """统计质因数的幂次，返回剩余值和找到的质因数列表"""
    other_divisor = input_target
    factors = []
    while other_divisor % num == 0:
        factors.append(num)
        other_divisor = other_divisor // num
    return other_divisor, factors

def worker(prime_chunk):
    """工作进程函数，处理分配到的质数块"""
    for p in prime_chunk:
        # 无锁读终止标记，状态单向变更无脏读问题
        if found.value:
            return
        current_upto = new_upto_target.value
        if p > current_upto:
            # 加锁修改终止标记，避免多个进程重复写入剩余值
            with lock:
                if not found.value:
                    found.value = True
                    if last_divisor.value > 1:
                        factor_queue.put(last_divisor.value)
            return
        current_target = full_number_target.value
        if current_target % p == 0:
            remaining, p_factors = exponent_finder(p, current_target)
            # 加锁原子更新所有关联状态
            with lock:
                for f in p_factors:
                    factor_queue.put(f)
                full_number_target.value = remaining
                last_divisor.value = remaining
                new_upto_target.value = math.sqrt(remaining)
                # 剩余值已是质数，直接终止所有任务
                if remaining < p * p:
                    found.value = True
                    if remaining > 1:
                        factor_queue.put(remaining)
                    return

def main_multi_process(n, process_num=4):
    # 初始化共享变量
    shared_target = Value('L', n)
    shared_upto = Value('d', math.sqrt(n))
    shared_last = Value('L', n)
    shared_found = Value('b', False)
    lock = Lock()
    result_q = Queue()
    prime_array = np.load('prime_10000.npy')
    # 切分质数列表为等大块
    chunk_size = len(prime_array) // process_num + 1
    prime_chunks = [prime_array[i:i+chunk_size] for i in range(0, len(prime_array), chunk_size)]
    # 启动进程池执行任务
    with Pool(processes=process_num, initializer=init_shared, initargs=(shared_target, shared_upto, shared_last, shared_found, lock, result_q, prime_array)) as pool:
        pool.map(worker, prime_chunks)
    # 收集结果并排序
    factors = []
    while not result_q.empty():
        factors.append(result_q.get())
    return sorted(factors)

上述实现中，共享内存的读写延迟和普通进程内变量基本一致，锁的粒度仅覆盖状态更新的几行代码，不会成为性能瓶颈；切分质数块的大小可以根据CPU核心数调整，块过小会增加进程调度开销，块过大会导致状态更新不及时，单块包含100~1000个质数时性能表现最优。

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