使用Python多进程执行DB2查询遇TypeError问题求助

核心错误：TypeError 原因及修复 #

错误根源 #

1. 实例方法无法直接在多进程Pool中调用：multiprocessing.Pool的方法（如starmap/map）无法直接处理类的实例方法，子进程无法正确序列化并传递self实例，导致调用时参数绑定错误。
2. 变量未定义：run方法中使用的filenames是未定义变量，应使用类初始化时的self.files。
3. 参数传递方式错误：starmap适合传递多参数元组，当前每个任务仅需一个参数，使用map更合适。

修复方案 #

将查询执行逻辑重构为独立函数，避免多进程中实例序列化问题；同时修正变量引用错误，采用更适配的map方法传递参数。

其他代码问题及修正 #

1. 全局变量无法跨进程同步：executed_queries作为全局变量，在多进程环境下子进程的修改不会同步到主进程，需用multiprocessing.Manager创建共享列表。
2. 文件路径错误：查询文件存储在MYqueries文件夹下，但原代码遍历当前目录，需拼接完整路径才能正确读取文件。
3. 语法错误：next((f for f in self.files ...)]中括号不匹配，应改为)。
4. 递归调用风险：子进程中递归调用self.execute_query可能导致无限递归或资源泄漏，改为在函数内处理重试逻辑，避免递归。
5. DB连接管理问题：部分版本ibm_db不支持with语句管理连接，需手动在finally块中关闭连接，避免资源泄漏。
6. 无返回值问题：原execute_query无返回值，导致results列表全为None，无法跟踪执行状态，需添加明确返回值。

修正后的完整代码 #

import multiprocessing
import os
import ibm_db
from functools import partial

def execute_query(filename, executed_queries, db_params):
    conn = None
    try:
        # 建立DB2连接
        conn = ibm_db.connect(db_params, "", "")
        if not conn:
            print(f"Failed to connect to DB for file: {filename}")
            return (filename, False, "Connection failed")
        
        # 读取查询文件（拼接文件夹路径）
        full_path = os.path.join("MYqueries", filename)
        with open(full_path, 'r') as f:
            query = f.read().strip()
        
        print(f"Executing query from: {filename}")
        
        # 重试逻辑
        for attempt in range(3):
            try:
                stmt = ibm_db.exec_query(conn, query)
                # 标记查询已执行
                executed_queries.append(filename)
                return (filename, True, f"Success on attempt {attempt+1}")
            except Exception as e:
                print(f"Attempt {attempt+1} failed for {filename}: {str(e)}")
        
        # 3次重试均失败
        return (filename, False, "Failed after 3 retries")
    except Exception as e:
        return (filename, False, f"Unexpected error: {str(e)}")
    finally:
        # 确保关闭DB连接
        if conn:
            ibm_db.close(conn)

def main():
    # 使用Manager创建共享列表，跟踪已执行查询
    with multiprocessing.Manager() as manager:
        executed_queries = manager.list()
        
        # 获取MYqueries文件夹下的查询文件
        query_dir = "MYqueries"
        if not os.path.exists(query_dir):
            print(f"Directory {query_dir} not found")
            return
        
        filenames = [f for f in os.listdir(query_dir) 
                     if f.startswith('MYqueries') and f not in executed_queries]
        
        if not filenames:
            print("No query files found")
            return
        
        # DB2连接参数
        db_params = "DATABASE=sample;HOSTNAME=localhost;PORT=50000;USERNAME=db2admin;PASSWORD=db2admin"
        
        # 绑定共享变量和DB参数到工作函数
        worker_func = partial(execute_query, executed_queries=executed_queries, db_params=db_params)
        
        with multiprocessing.Pool(processes=15) as pool:
            # 使用map传递单个参数，适配当前场景
            results = pool.map(worker_func, filenames)
        
        # 打印执行结果汇总
        print("\n=== Execution Results ===")
        for filename, success, msg in results:
            status = "SUCCESS" if success else "FAILED"
            print(f"{status} - {filename}: {msg}")

if __name__ == "__main__":
    main()

关键改进说明 #

• 多进程安全状态跟踪：通过multiprocessing.Manager().list()创建共享列表，确保子进程对已执行查询的标记能同步到主进程。
• 独立函数适配多进程：将查询执行逻辑改为独立函数，避免实例序列化问题，通过functools.partial传递共享变量和DB参数。
• 正确文件路径处理：拼接MYqueries文件夹路径，确保能正确读取目标查询文件。
• 明确结果返回：函数返回包含文件名、执行状态和消息的元组，便于主进程汇总和展示执行结果。
• 安全资源管理：使用finally块确保DB连接被关闭，避免资源泄漏。

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