大文件分块加密：Multiprocessing与ThreadPoolExecutor性能对比

我正在开发一款基于Python的CLI文件加密工具，采用PyNaCl库实现。待加密文件的典型大小为200-500MB，经实验发现，将数据分割为约5MB的块并使用ThreadPoolExecutor加密，比直接加密整体数据速度更快。由于我对并发/并行技术的细节不甚了解，希望知晓加密大量数据的最优高性能方案。

我想了解使用Multiprocessing是否比ThreadPoolExecutor更快，同时也不确定当前的多线程实现是否为最优方式，恳请相关建议。

当前实现代码：

from concurrent.futures import ThreadPoolExecutor
from os import urandom
from typing import Tuple

from nacl import secret
from nacl.bindings import sodium_increment
from nacl.secret import SecretBox


def encrypt_chunk(args: Tuple[bytes, SecretBox, bytes, int]):
    chunk, box, nonce, macsize = args
    try:
        outchunk = box.encrypt(chunk, nonce).ciphertext
    except Exception as e:
        err = Exception("Error encrypting chunk")
        err.__cause__ = e
        return err
    if not len(outchunk) == len(chunk) + macsize:
        return Exception("Error encrypting chunk")
    return outchunk


def encrypt(
    data: bytes,
    key: bytes,
    nonce: bytes,
    chunksize: int,
    macsize: int,
):
    box = SecretBox(key)
    args = []
    total = len(data)
    i = 0
    while i < total:
        chunk = data[i : i + chunksize]
        nonce = sodium_increment(nonce)
        args.append((chunk, box, nonce, macsize,))
        i += chunksize
    executor = ThreadPoolExecutor(max_workers=4)
    out = executor.map(encrypt_chunk, args)
    executor.shutdown(wait=True)
    return out

多进程 vs 多线程的性能对比 #

PyNaCl底层是基于C实现的加密操作，属于CPU密集型任务。Python的GIL（全局解释器锁）会限制多线程在CPU密集型任务中的并行能力——即便PyNaCl的C扩展会释放GIL，多线程的并行度仍受限于GIL的调度机制；而多进程可以完全绕过GIL，充分利用多核CPU的全部算力。

针对200-500MB的文件，多进程（multiprocessing或ProcessPoolExecutor）通常会比多线程更快，尤其是在CPU核心数较多的机器上。但需注意两点：

• 进程间数据传递存在开销，因此块大小不能过小（比如小于1MB），否则通信开销会抵消并行收益。你当前采用的5MB块大小是合理的，可以保持。
• SecretBox对象无法安全地在进程间共享（底层C结构无法序列化），需要在每个子进程内重新初始化。

当前多线程实现的优化建议 #

1. 动态设置线程数：不要固定max_workers=4，可以设置为os.cpu_count()或os.cpu_count() * 2（若加密操作存在少量IO等待），让程序自动适配机器核心数。
2. 避免预生成所有参数：如果文件过大，预先生成所有chunk和参数会占用大量内存。建议改用生成器动态生成参数，边生成边提交任务。
3. 优化错误处理：当前返回Exception对象的方式不够直观，建议在encrypt_chunk中直接抛出异常，在encrypt迭代结果时捕获，便于定位出错块。
4. 改用流式处理：不要一次性将整个文件读入内存（data: bytes），对于大文件应分块读取并加密，减少内存占用的同时，可与并发操作结合实现流式加密。

优化后的多进程实现示例 #

from concurrent.futures import ProcessPoolExecutor
from os import urandom, cpu_count
from typing import Tuple

from nacl.secret import SecretBox
from nacl.bindings import sodium_increment


def encrypt_chunk(args: Tuple[bytes, bytes, bytes, int]):
    # 每个进程独立初始化SecretBox，规避进程间共享问题
    chunk, key, nonce, macsize = args
    box = SecretBox(key)
    try:
        outchunk = box.encrypt(chunk, nonce).ciphertext
    except Exception as e:
        raise Exception(f"Error encrypting chunk: {str(e)}") from e
    if len(outchunk) != len(chunk) + macsize:
        raise Exception("Encrypted chunk size mismatch")
    return outchunk


def encrypt(
    data: bytes,
    key: bytes,
    nonce: bytes,
    chunksize: int,
    macsize: int,
):
    args = []
    total = len(data)
    current_nonce = nonce
    for i in range(0, total, chunksize):
        chunk = data[i:i+chunksize]
        current_nonce = sodium_increment(current_nonce)
        args.append((chunk, key, current_nonce, macsize))
    
    # 以CPU核心数作为进程数，最大化利用多核算力
    with ProcessPoolExecutor(max_workers=cpu_count()) as executor:
        results = executor.map(encrypt_chunk, args)
    return results

额外实用建议 #

• 测试不同块大小：在目标机器上测试2MB、5MB、10MB等不同块大小的性能，找到最优值——块太小会增加调度开销，块太大则会降低并行度。
• 流式处理大文件：若处理GB级文件，不要一次性加载全部数据，改用文件对象分块读取，读取一块就提交给executor，大幅降低内存占用。
• 做基准测试：用timeit模块对比多线程和多进程的实际运行时间，结合硬件环境选择最优方案——比如单核心机器上，多线程可能和多进程性能相当，甚至更优（进程开销更大）。

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