使用psycopg2 ThreadedConnectionPool结合多进程池时出现连接已关闭错误的技术咨询

你遇到的核心问题是**ThreadedConnectionPool是为多线程设计的，完全不支持跨进程共享**，结合Python多进程的特性就会导致连接失效：

1. 多进程的内存复制机制：当你在父进程初始化连接池后创建子进程，每个子进程会复制父进程的内存空间（包括连接池对象）。但数据库连接是和父进程的文件描述符绑定的，子进程里的连接副本根本无法和数据库建立有效通信，所以会出现SSL error: decryption failed or bad record mac这类底层通信错误，最终触发connection already closed。
2. 额外的代码问题：你的CursorFromPool类的__init__方法有语法错误（缺少self参数），虽然你说单独调用没问题，但这会导致上下文管理器使用时出现NameError: name 'self' is not defined，需要先修复这个笔误。

方法1：每个子进程独立初始化连接池 #

既然多进程无法共享连接池，那我们让每个子进程启动时自己初始化专属的连接池。可以通过multiprocessing.Pool的initializer参数实现：

修改后的代码示例： #

import psycopg2
import os
import pandas as pd
import time
from psycopg2.pool import ThreadedConnectionPool
from multiprocessing import Pool

# 先修复CursorFromPool的__init__方法
class CursorFromPool:
    ''' Wrapper class to connect to PostgresDB '''
    def __init__(self):  # 这里添加self参数
        self.connection = None
        self.cursor = None

    def __enter__(self):
        self.connection = Database.get_connection()
        self.cursor = self.connection.cursor()
        return self.cursor

    def __exit__(self, exc_type, exc_val, exc_tb):
        # 先判断连接是否有效，再执行操作
        if self.connection and not self.connection.closed:
            if exc_val is not None:
                self.connection.rollback()
            else:
                self.cursor.close()
                self.connection.commit()
            Database.return_connection(self.connection)

# 定义子进程初始化函数
def init_child_process():
    # 每个子进程自己初始化连接池
    Database.initialize(*db_args_passed_here*)

def clean(skiprows=None):
    df_flags = pd.read_csv('filename.csv')
    data_list = [tuple(row) for row in df_flags.itertuples(index=False)]
    start = time.time()
    try:
        with CursorFromPool() as cursor:
            psycopg2.extras.execute_batch(cursor,'INSERT INTO db_name.table_name VALUES (%s,%s,%s,%s)', data_list)
            print(cursor.statusmessage)
    except Exception as e:
        print(f"Process {os.getpid()} error: {e}")
    end = time.time()
    print(f"Process {os.getpid()} took {end - start:.2f}s")
    return True

if __name__ == "__main__":
    # 父进程不需要初始化连接池，交给子进程自己处理
    skiper = [1,2,3,4,5,6,7,8]
    with Pool(processes=4, initializer=init_child_process) as pool:
        pool.map(clean, skiper)

方法2：改用多线程替代多进程 #

如果你的任务是IO密集型（比如数据库写入），多线程其实比多进程更合适：

• ThreadedConnectionPool本身就是线程安全的，完美适配多线程场景
• 多线程不需要复制内存空间，资源开销更小
• Python的GIL在IO等待时会自动释放，不会影响多线程的并发效率

修改后的代码示例： #

from concurrent.futures import ThreadPoolExecutor

# 父进程初始化连接池
Database.initialize(*db_args_passed_here*)

def clean(skiprows=None):
    df_flags = pd.read_csv('filename.csv')
    data_list = [tuple(row) for row in df_flags.itertuples(index=False)]
    start = time.time()
    with CursorFromPool() as cursor:
        psycopg2.extras.execute_batch(cursor,'INSERT INTO db_name.table_name VALUES (%s,%s,%s,%s)', data_list)
        print(cursor.statusmessage)
    end = time.time()
    print(f"Thread took {end - start:.2f}s")
    return True

if __name__ == "__main__":
    skiper = [1,2,3,4,5,6,7,8]
    with ThreadPoolExecutor(max_workers=4) as executor:
        executor.map(clean, skiper)
    # 最后关闭连接池
    if not Database.connection_closed():
        Database.close_all_connections()

方法3：每个进程使用独立连接（不使用连接池） #

如果必须用多进程且不想维护多个连接池，可以直接在每个任务里创建独立的数据库连接，用完就关闭：

def clean(skiprows=None):
    df_flags = pd.read_csv('filename.csv')
    data_list = [tuple(row) for row in df_flags.itertuples(index=False)]
    start = time.time()
    conn = None
    cursor = None
    try:
        # 每个任务自己创建连接
        conn = psycopg2.connect(**db_args_passed_here*)
        cursor = conn.cursor()
        psycopg2.extras.execute_batch(cursor,'INSERT INTO db_name.table_name VALUES (%s,%s,%s,%s)', data_list)
        conn.commit()
        print(cursor.statusmessage)
    except Exception as e:
        if conn:
            conn.rollback()
        print(f"Error: {e}")
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()
    end = time.time()
    print(f"Process took {end - start:.2f}s")
    return True

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