如何在Python多进程中保持进程存活以维持状态并复用？

你遇到的核心痛点是避免每次调用都序列化/反序列化大型初始数据，要实现每个子进程仅初始化一次数据，之后多次接收小参数处理。下面是三种最实用的解决方案：

方案一：独立进程+队列（完全控制进程生命周期） #

直接创建与数据块数量对应的进程，每个进程初始化自己的MyWorker，通过队列传递任务参数和接收结果。这种方式完全避免了大数据的重复传递，逻辑直观。

import multiprocessing

class MyWorker:
    def __init__(self, data):
        self.data = data  # 仅在进程启动时初始化一次大型数据
    def dosomework(self, arg):
        # 示例计算：用self.data和arg执行逻辑
        return len(self.data) + arg

def worker_process(chunk, task_queue, result_queue):
    """子进程逻辑：初始化worker后循环处理任务"""
    worker = MyWorker(chunk)
    while True:
        arg = task_queue.get()
        if arg is None:  # 收到结束信号时退出
            break
        result = worker.dosomework(arg)
        result_queue.put(result)

if __name__ == "__main__":
    # 模拟大型数组和拆分后的块
    hugearray = list(range(1000000))
    chunksize = 200000
    datachunks = [hugearray[i:i+chunksize] for i in range(0, len(hugearray), chunksize)]
    argstodo = [1,2,3,4,5]
    
    # 创建任务队列（每个进程一个）和结果队列
    task_queues = [multiprocessing.Queue() for _ in datachunks]
    result_queue = multiprocessing.Queue()
    
    # 启动所有子进程
    processes = []
    for chunk, task_q in zip(datachunks, task_queues):
        p = multiprocessing.Process(target=worker_process, args=(chunk, task_q, result_queue))
        p.start()
        processes.append(p)
    
    # 批量处理所有参数
    results = []
    for arg in argstodo:
        # 给每个进程发送当前参数
        for q in task_queues:
            q.put(arg)
        # 收集所有进程的结果
        current_results = [result_queue.get() for _ in datachunks]
        results.append(current_results)
    
    # 发送结束信号并等待进程退出
    for q in task_queues:
        q.put(None)
    for p in processes:
        p.join()
    
    print(results)

方案二：使用Pool的初始化机制（更简洁） #

利用multiprocessing.Pool的initializer参数，让每个进程启动时加载对应的数据集，之后仅传递小参数处理。这种方式代码更简洁，复用了Pool的进程管理能力。

import multiprocessing
from multiprocessing import current_process

class MyWorker:
    def __init__(self, data):
        self.data = data
    def dosomework(self, arg):
        return len(self.data) + arg

# 全局变量存储所有数据块和当前进程的worker
all_chunks = None
worker = None

def init_worker():
    """每个进程启动时初始化自己的worker"""
    global worker, all_chunks
    # 获取当前进程在池中的序号（转换为数组索引）
    process_idx = current_process()._identity[0] - 1
    worker = MyWorker(all_chunks[process_idx])

def process_task(arg):
    """仅接收参数，复用已初始化的worker"""
    return worker.dosomework(arg)

if __name__ == "__main__":
    hugearray = list(range(1000000))
    chunksize = 200000
    global all_chunks
    all_chunks = [hugearray[i:i+chunksize] for i in range(0, len(hugearray), chunksize)]
    argstodo = [1,2,3,4,5]
    
    results = []
    # 进程数与数据块数量一致，确保每个进程对应一个块
    with multiprocessing.Pool(processes=len(all_chunks), initializer=init_worker) as pool:
        for arg in argstodo:
            # 给每个进程发送当前参数，收集所有结果
            current_results = pool.map(process_task, [arg]*len(all_chunks))
            results.append(current_results)
    
    print(results)

方案三：针对Numpy数组的共享内存优化 #

如果你的大型数据是Numpy数组，可以用共享内存直接在进程间共享数据，避免复制开销：

import multiprocessing
import numpy as np

class MyWorker:
    def __init__(self, arr_slice):
        self.arr = arr_slice
    def dosomework(self, arg):
        return self.arr.sum() + arg

def init_worker(shared_arr):
    global worker
    process_idx = multiprocessing.current_process()._identity[0] - 1
    chunksize = shared_arr.size // len(datachunks)
    start = process_idx * chunksize
    end = start + chunksize if process_idx != len(datachunks)-1 else shared_arr.size
    worker = MyWorker(shared_arr[start:end])

def process_task(arg):
    return worker.dosomework(arg)

if __name__ == "__main__":
    hugearray = np.arange(1000000)
    chunksize = 200000
    global datachunks
    datachunks = [hugearray[i:i+chunksize] for i in range(0, len(hugearray), chunksize)]
    
    # 创建共享内存数组
    shared_arr = multiprocessing.Array('i', hugearray.size)
    np_shared_arr = np.frombuffer(shared_arr.get_obj(), dtype='i')
    np_shared_arr[:] = hugearray[:]
    
    argstodo = [1,2,3,4,5]
    results = []
    
    with multiprocessing.Pool(processes=len(datachunks), initializer=init_worker, initargs=(np_shared_arr,)) as pool:
        for arg in argstodo:
            current_results = pool.map(process_task, [arg]*len(datachunks))
            results.append(current_results)
    
    print(results)

关键说明 #

• 所有方案都确保大型数据仅在进程启动时传递/加载一次，后续仅传递轻量的arg参数。
• 方案一适合需要精细控制进程行为的场景，方案二更简洁，适合大多数常规需求。
• 共享内存方案适合Numpy数组，能进一步节省内存开销。

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