Python multiprocessing多进程中如何正确共享DB连接池

你遇到的两个报错本质都是违反了SQLAlchemy多进程使用的基本规则：

• 直接向子进程传入Engine对象触发Broken Pipe Error：Engine及其管理的连接池、数据库连接是和创建进程强绑定的资源，底层持有socket文件描述符、线程锁、事务状态等进程私有上下文。多进程fork或spawn时，子进程拿到的只是父进程Engine的内存拷贝，底层socket并不属于子进程，跨进程操作同一个socket句柄必然触发管道断裂。
• 经Queue传递Engine触发序列化错误：Engine本身不是可pickle序列化的对象，内部包含create_engine生成的局部闭包函数、线程锁、内存态连接状态，根本无法通过Manager队列跨进程传输。

绝对不要尝试跨进程共享SQLAlchemy引擎、连接池或单个数据库连接。SQLAlchemy官方从未支持Engine跨进程复用，网上所谓的“共享连接池”方案都会随机出现连接错乱、事务丢失、进程崩溃问题，没有生产可用的稳定实现。

生产环境唯一稳定的方案是每个子进程独立创建属于自己的Engine和连接池，借助multiprocessing.Pool的初始化钩子，在子进程启动时一次性完成引擎初始化，全程不需要跨进程传递任何数据库相关对象。
参考实现代码：

import multiprocessing
import sqlalchemy
from sqlalchemy.pool import QueuePool

# 子进程全局变量，存储当前进程专属的数据库引擎
_process_local_engine = None

def init_worker(db_conn_str: str):
    """子进程初始化函数，每个子进程启动时仅执行一次"""
    global _process_local_engine
    _process_local_engine = sqlalchemy.create_engine(
        db_conn_str,
        poolclass=QueuePool,
        pool_size=2,  # 单进程连接池大小，不要设置过大
        max_overflow=0,  # 禁止超出pool_size创建额外连接
        pool_pre_ping=True,  # 连接探活，自动过滤被数据库回收的死连接
        pool_recycle=3600  # 连接存活超过1小时自动回收，适配数据库wait_timeout配置
    )

def biz_task(data):
    """你的实际业务处理函数"""
    global _process_local_engine
    with _process_local_engine.connect() as conn:
        # 在这里执行你的数据库查询、业务逻辑
        # result = conn.execute(sqlalchemy.text("SELECT ..."))
        pass

if __name__ == "__main__":
    # 替换为你的实际数据库连接串
    DB_CONN_STR = "mysql+pymysql://username:password@host:port/db_name"
    # 替换为你要处理的数据集
    task_data_list = []

    # 初始化进程池时传入初始化函数和参数，每个子进程启动时自动创建自己的引擎
    with multiprocessing.Pool(
        processes=8,
        initializer=init_worker,
        initargs=(DB_CONN_STR,)
    ) as pool:
        # 直接提交任务即可，不需要传递引擎、队列这类对象
        pool.map(biz_task, task_data_list)

• 控制总连接数：因为每个子进程持有独立连接池，总数据库连接数 = 子进程数量 * 单进程pool_size + 主进程使用的连接数，配置时要确保总连接数不超过数据库设置的最大连接上限，避免触发数据库连接满的错误。
• 兼容进程启动模式：Windows、Python3.8+版本的macOS默认使用spawn模式启动子进程，必须把进程池启动、任务提交的逻辑放在if __name__ == "__main__":代码块内，否则会出现递归启动子进程、序列化失败的问题。
• 短任务场景优化：如果你的单任务执行时间极短，不需要复用连接，可以在创建Engine时指定poolclass=sqlalchemy.pool.NullPool关闭连接池，每次执行查询都新建真实连接、用完立即释放，完全规避连接复用带来的风险，缺点是频繁建连会带来少量性能开销。
• 禁止父进程提前创建Engine：不要在主进程中先创建Engine再启动子进程，哪怕你没有主动传递Engine，fork模式下子进程会拷贝父进程的Engine对象，复用父进程的底层连接，依然会触发Broken Pipe错误。

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