修复Zipfile并发写入问题：Python2.7多进程写npz文件内容丢失

使用Python 2.7的multiprocessing模块并发将numpy数组写入zip文件（手动实现增量式npz写入，避免内存溢出），控制台显示共执行8次写入操作，但最终zip归档中仅保留0.npy、1.npy、4.npy、7.npy四个文件，其余文件丢失。

限制条件：

• 无法升级Python版本或添加第三方库
• 需保持zip文件持续打开状态（频繁开闭会导致写入速度下降10倍）
• 不想单独分配进程负责写入（写入并非性能瓶颈，避免资源浪费）

原代码：

import numpy as np                                
import zipfile
import io
import multiprocessing as mp

class ZipStreamEncoder(object):

    def __init__(self, filename, manager):
    
        self.arrayNb = manager.Value('i', 0)
        self.lock = manager.Lock()
        self.fileName = filename
    
    def open(self):
        self.zipfile = zipfile.ZipFile(self.fileName, 'a', compression=zipfile.ZIP_DEFLATED, allowZip64 = True)
    
    def writeArray(self, arrayToExport, proc):
        with self.lock:
            print("Process " + str(proc) + " is writing array " + str(self.arrayNb.value))
            buf = io.BytesIO()
            np.lib.npyio.format.write_array(buf, arrayToExport)
            self.zipfile.writestr(str(self.arrayNb.value) + ".npy", buf.getvalue())
            self.arrayNb.value += 1

    def close(self):        
        self.zipfile.close()

def extractor_worker(proc, zip_object):
    zip_object.open()
    zip_object.writeArray(np.array([[0,1,2],[0,1,2]]), proc)
    zip_object.writeArray(np.array([[0,1,3],[0,1,3]]), proc)
    zip_object.close()

    return None

if __name__ == '__main__':

    nb_processes = 4 #mp.cpu_count()
    manager = mp.Manager()
    q_tasks = manager.Queue() 
    pool = mp.Pool(nb_processes)

    zip_object = ZipStreamEncoder("my_test.zip", manager)

    jobs = []
    for proc in range(nb_processes):
        job = pool.apply_async(extractor_worker, (proc, zip_object))
        jobs.append(job)
    
    for job in jobs:
        job.get()
    
    pool.close()
    pool.join()

控制台输出：

Process 0 is writing array 0
Process 0 is writing array 1
Process 1 is writing array 2
Process 2 is writing array 3
Process 3 is writing array 4
Process 1 is writing array 5
Process 2 is writing array 6
Process 3 is writing array 7

每个子进程都独立调用zip_object.open()创建了自己的ZipFile实例，同时打开同一个物理文件。多个ZipFile实例并发写入时，每个实例维护独立的文件指针和中央目录信息：

• 进程写入的内容会追加到文件中，但中央目录仅记录当前实例写入的文件
• 当进程关闭ZipFile时，会将自身维护的中央目录写入文件末尾，后续进程的写入会覆盖之前的中央目录，导致只有最后几个进程的文件条目被保留，其他文件因无中央目录记录而无法被识别。

由主进程唯一负责打开和维护ZipFile实例，子进程仅生成写入数据并通过共享队列传递给主进程执行写入，既保证文件持续打开，又避免多进程文件操作冲突。

修复后的代码：

import numpy as np                                
import zipfile
import io
import multiprocessing as mp

class ZipStreamEncoder(object):
    def __init__(self, filename, manager):
        self.arrayNb = manager.Value('i', 0)
        self.lock = manager.Lock()
        self.write_queue = manager.Queue()  # 共享队列传递写入任务
        self.fileName = filename

    def open(self):
        # 主进程唯一打开zip文件，使用'w'模式（如需增量写入可改为'a'）
        self.zipfile = zipfile.ZipFile(self.fileName, 'w', compression=zipfile.ZIP_DEFLATED, allowZip64=True)

    def _process_queue(self):
        # 主进程处理队列中的写入任务
        while True:
            task = self.write_queue.get()
            if task is None:  # 终止信号
                break
            file_name, data = task
            self.zipfile.writestr(file_name, data)

    def writeArray(self, arrayToExport, proc):
        with self.lock:
            current_idx = self.arrayNb.value
            print("Process " + str(proc) + " is writing array " + str(current_idx))
            buf = io.BytesIO()
            np.lib.npyio.format.write_array(buf, arrayToExport)
            # 将写入任务放入队列，由主进程执行
            self.write_queue.put((str(current_idx) + ".npy", buf.getvalue()))
            self.arrayNb.value += 1

    def close(self):
        # 发送终止信号，等待所有队列任务处理完成
        self.write_queue.put(None)
        self._process_queue()
        self.zipfile.close()

def extractor_worker(proc, zip_object):
    # 子进程不再打开/关闭zip文件，仅调用writeArray生成数据
    zip_object.writeArray(np.array([[0,1,2],[0,1,2]]), proc)
    zip_object.writeArray(np.array([[0,1,3],[0,1,3]]), proc)
    return None

if __name__ == '__main__':
    nb_processes = 4
    manager = mp.Manager()
    pool = mp.Pool(nb_processes)

    zip_object = ZipStreamEncoder("my_test.zip", manager)
    zip_object.open()  # 主进程提前打开文件

    jobs = []
    for proc in range(nb_processes):
        job = pool.apply_async(extractor_worker, (proc, zip_object))
        jobs.append(job)
    
    # 等待所有子进程完成数据生成
    for job in jobs:
        job.get()
    
    # 关闭zip文件，处理剩余队列任务
    zip_object.close()
    
    pool.close()
    pool.join()

1. 单进程文件管理：仅主进程打开和关闭ZipFile，避免多进程文件句柄冲突。
2. 共享队列传递任务：子进程生成数据后，将文件名和字节内容放入共享队列，由主进程统一写入，保证zip中央目录的一致性。
3. 保留计数器与锁：维持原有的共享计数器和锁，确保文件名的顺序和唯一性。
4. 优雅终止处理：关闭时发送终止信号，确保队列中所有写入任务都被执行完毕。

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