如何通过Python脚本并行执行Oracle数据库中的存储过程？

核心注意事项 #

Oracle数据库连接不是线程安全的，因此每个并行任务必须使用独立的数据库连接，不能共用同一个连接对象。

步骤1：封装存储过程执行逻辑 #

先将单个存储过程的执行逻辑封装为独立函数，包含连接创建、执行、异常处理和资源释放：

import logging
import cx_Oracle

def get_db_connection():
    """创建并返回独立的数据库连接"""
    dsn = cx_Oracle.makedsn('你的数据库主机', '端口号', service_name='服务名')
    return cx_Oracle.connect(
        user='用户名',
        password='密码',
        dsn=dsn
    )

def execute_procedure(proc_package):
    """执行指定包中的startProcess存储过程"""
    conn = None
    cursor = None
    try:
        conn = get_db_connection()
        cursor = conn.cursor()
        # 执行存储过程
        cursor.execute(f'BEGIN {proc_package}.startProcess; END;')
        conn.commit()
        logging.info(f"存储过程{proc_package}.startProcess执行成功")
    except Exception as e:
        if conn:
            conn.rollback()
        logging.exception(f"存储过程{proc_package}.startProcess执行失败: {str(e)}")
    finally:
        # 确保资源关闭
        if cursor:
            cursor.close()
        if conn:
            conn.close()

步骤2：并行执行实现 #

方法1：使用threading模块手动管理线程

import threading

def main():
    # 定义需要执行的存储过程包列表
    proc_packages = ['MYPACKAGE1', 'MYPACKAGE2', 'MYPACKAGE3']
    
    threads = []
    for pkg in proc_packages:
        thread = threading.Thread(target=execute_procedure, args=(pkg,))
        threads.append(thread)
        thread.start()
    
    # 等待所有线程执行完成
    for thread in threads:
        thread.join()

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    main()

方法2：使用concurrent.futures.ThreadPoolExecutor（更简洁）

from concurrent.futures import ThreadPoolExecutor

def main():
    proc_packages = ['MYPACKAGE1', 'MYPACKAGE2', 'MYPACKAGE3']
    
    # 用线程池管理并行任务，max_workers控制并发数
    with ThreadPoolExecutor(max_workers=3) as executor:
        executor.map(execute_procedure, proc_packages)

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    main()

额外提示 #

• 控制并发数：max_workers不要设置过大，避免超出数据库的最大连接数限制
• 连接复用：如果需要频繁执行，可以考虑使用连接池（如cx_Oracle的SessionPool）优化连接创建开销
• 日志配置：根据需求调整日志级别和格式，方便问题排查

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