Python多进程池初始化数据库连接池后如何复用？

我正尝试搭建一个多进程池，让每个worker进程拥有一组专属的数据库连接池，避免任务函数每次调用都重复创建数据库连接。目前已通过_init_worker为每个worker初始化了连接池，但不知道如何在任务函数中复用这些已初始化的连接池。

当前代码存在一个关键问题：Pool初始化时，initializer参数传入的是_init_worker的调用结果，而非函数本身，这会导致初始化逻辑在主进程执行，而非每个worker进程。

核心思路是利用进程全局变量：每个worker进程拥有独立的内存空间，在_init_worker中把连接池实例绑定到worker进程的全局变量，任务函数即可直接访问这些全局变量复用连接池。

步骤1：修正进程池的初始化方式 #

Pool的initializer需要传入函数对象，初始化参数通过initargs传递，而非直接调用函数：

self.pool = Pool(
    processes=multiprocessing.cpu_count(),
    initializer=self._init_worker,
    initargs=(local_ip, linked_ip, server_ip)  # 传递初始化参数
)

步骤2：在worker初始化函数中绑定全局变量 #

修改_init_worker，将创建的DatabaseManager实例赋值给进程全局变量：

def _init_worker(self, local_ip, linked_ip, server_ip):
    """为每个worker进程初始化专属的数据库连接池，并绑定到全局变量"""
    global local_db_pool, linked_db_pool, server_db_pool
    
    local_db_pool = DatabaseManager(
        host_ip=local_ip,
        host_username='something',
        db_password='something',
        db_name='something',
        db_ip=local_ip,
        db_port=3306,
        db_pool_name='local_db_pool',
        db_pool_size=5
    )
    linked_db_pool = DatabaseManager(
        host_ip=linked_ip,
        host_username='something',
        db_password='something',
        db_name='something',
        db_ip=linked_ip,
        db_port=3306,
        db_pool_name='linked_db_pool',
        db_pool_size=5
    )
    server_db_pool = DatabaseManager(
        host_ip=server_ip,
        host_username='something',
        db_password='something',
        db_name='something',
        db_ip=server_ip,
        db_port=3306,
        db_pool_name='server_db_pool',
        db_pool_size=5
    )

步骤3：在任务函数中复用全局变量的连接池 #

任务函数直接引用worker进程的全局变量，从连接池中获取连接执行操作：

def task_function(some_param):
    """示例任务函数，复用已初始化的连接池"""
    global local_db_pool, linked_db_pool, server_db_pool
    
    # 从本地连接池获取连接
    conn = local_db_pool.db_connection.get_connection()
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM some_table WHERE id = %s", (some_param,))
        result = cursor.fetchall()
        # 处理结果
    finally:
        conn.close()  # 归还连接到池
        
    # 同理可使用linked_db_pool或server_db_pool

完整修正后的代码示例 #

import multiprocessing
from multiprocessing import Pool
import mysql.connector
import socket

class DatabaseManager:
    def __init__(self, host_ip, host_username, db_password, db_name, db_ip, db_port, db_pool_name, db_pool_size):
        print('DatabaseManager instantiation...')
        self.host_ip = host_ip
        self.host_username = host_username
        self.db_password = db_password
        self.db_name = db_name
        self.db_ip = db_ip
        self.db_port = db_port
        # 注：此处的socket连接非必需，可根据实际需求保留或移除
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.socket.connect((self.db_ip, self.db_port))

        self.db_connection = mysql.connector.pooling.MySQLConnectionPool(
            pool_name=db_pool_name,
            pool_size=db_pool_size,
            pool_reset_session=True,
            host=self.host_ip,
            user=self.host_username,
            password=self.db_password,
            database=self.db_name)

class MultiprocessingManager:
    _instance = None  # 单例实例存储

    def __new__(cls, local_ip, linked_ip, server_ip):
        if not cls._instance:
            cls._instance = super(MultiprocessingManager, cls).__new__(cls)
            cls._instance._initialized = False
        return cls._instance

    def __init__(self, local_ip, linked_ip, server_ip):
        if not self._initialized:
            print("Initializing MultiprocessingManager instance...")
            self.pool = Pool(
                processes=multiprocessing.cpu_count(),
                initializer=self._init_worker,
                initargs=(local_ip, linked_ip, server_ip)
            )
            self._initialized = True
            print("MultiprocessingManager initialized.")

    def _init_worker(self, local_ip, linked_ip, server_ip):
        global local_db_pool, linked_db_pool, server_db_pool
        
        local_db_pool = DatabaseManager(
            host_ip=local_ip,
            host_username='something',
            db_password='something',
            db_name='something',
            db_ip=local_ip,
            db_port=3306,
            db_pool_name='local_db_pool',
            db_pool_size=5
        )
        linked_db_pool = DatabaseManager(
            host_ip=linked_ip,
            host_username='something',
            db_password='something',
            db_name='something',
            db_ip=linked_ip,
            db_port=3306,
            db_pool_name='linked_db_pool',
            db_pool_size=5
        )
        server_db_pool = DatabaseManager(
            host_ip=server_ip,
            host_username='something',
            db_password='something',
            db_name='something',
            db_ip=server_ip,
            db_port=3306,
            db_pool_name='server_db_pool',
            db_pool_size=5
        )

# 示例任务函数
def sample_task(task_id):
    global local_db_pool
    conn = local_db_pool.db_connection.get_connection()
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT NOW()")
        current_time = cursor.fetchone()[0]
        print(f"Task {task_id} executed at {current_time} (worker PID: {multiprocessing.current_process().pid})")
    finally:
        conn.close()

# 使用示例
if __name__ == "__main__":
    manager = MultiprocessingManager("127.0.0.1", "192.168.1.100", "192.168.1.200")
    manager.pool.map(sample_task, range(10))
    manager.pool.close()
    manager.pool.join()

注意事项 #

• 每个worker进程的全局变量是独立的，主进程无法直接访问worker的全局变量，避免跨进程共享连接池（数据库连接无法在进程间安全共享）。
• 任务函数中使用连接后务必调用close()归还连接到池，避免连接耗尽。

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