Python多进程+PostgreSQL连接池报错：无法pickle psycopg2连接对象

错误原因 #

cannot pickle 'psycopg2.extensions.connection' object 本质是因为Python多进程间传递数据必须经过序列化（pickle），但psycopg2的数据库连接绑定到当前进程的操作系统套接字资源，这类资源无法被序列化和跨进程传递。你之前在主进程创建db_manager并传给子进程的ClientsHandler，本质是试图把主进程的数据库连接传递给子进程，必然触发这个错误。

另外你提到注释里的代码能运行，那是因为写法错误：executor.submit(run(group_of_clients_handlers, not_attempted_clients_domains)) 是先在主进程同步执行run函数，再把函数返回值传给submit，根本没用到多进程，自然不会触发序列化问题，这不是真正的多进程调用。

解决方案（不破坏现有架构） #

核心思路：每个子进程独立初始化自己的数据库连接池，而非从主进程传递。因为数据库连接/连接池无法跨进程共享，每个进程维护自己的连接池才是正确做法，同时也能实现连接复用的目标。

具体修改步骤 #

1. 调整主进程的group_of_clients_handlers结构
   不再在主进程创建ClientsHandler实例，只保留客户端ID和名称的映射：

   # 原代码
   # group_of_clients_handlers[client_id] = [ClientsHandler(db_manager), client_name]
   # 修改后
   group_of_clients_handlers[client_id] = client_name
   

2. 在run函数内初始化子进程专属的db_manager和ClientsHandler
   每个子进程启动后，自行创建连接池和处理器实例：

   def run(group_of_clients_handlers, domain):
       # 每个子进程独立初始化数据库连接池
       db_manager = DatabaseManager()
       # 基于客户端ID创建对应的ClientsHandler（绑定当前进程的连接池）
       clients_handlers_map = {
           client_id: ClientsHandler(db_manager) 
           for client_id, client_name in group_of_clients_handlers.items()
       }
       
       # 后续原有业务逻辑，比如使用clients_handlers_map处理domain
       # ...
   

3. 修正多进程调用写法（如果用submit）
   若要使用submit，必须传递函数对象和参数，而非函数执行结果：

   # 原错误写法
   # futures = executor.submit(run(group_of_clients_handlers, not_attempted_clients_domains))
   # 修正后
   futures = [executor.submit(run, group_of_clients_handlers, domain) 
              for domain in list_of_not_attempted_clients_domains]
   for future in concurrent.futures.as_completed(futures):
       pretty_print("Loading futures....")
       print(future.result())
   

额外优化建议 #

• 可以给DatabaseManager添加连接池参数（比如最大连接数），避免子进程连接池过大。
• 若想进一步优化，可以用ProcessPoolExecutor的initializer参数，在每个子进程启动时统一初始化db_manager并存在全局变量中，这样run函数可以直接复用：

  def init_process():
      global db_manager
      db_manager = DatabaseManager()
  
  # 初始化进程池时指定初始化函数
  with ProcessPoolExecutor(cores, initializer=init_process) as executor:
      # ... 后续逻辑
  

  然后在run函数里直接使用全局的db_manager创建ClientsHandler。

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