当前博文未提及该问题，以下是基于C知道的知识内容为您回答 要并行执行文件中的1000个SQL语句，可以使用Python的`multiprocessing`模块创建进程池，结合数据库连接池实现高效的并发执行。以下是完整的实现方案： ## 一、技术架构设计 ### 1.1 核心组件选择 | 组件 | 选择 | 说明 | |------|------|------| | 并发框架 | `multiprocessing.Pool` | 进程级并行，避免GIL限制 | | 数据库驱动 | `oracledb` (推荐) 或 `cx_Oracle` | Oracle官方Python驱动 | | 任务分配 | 分块(chunk)分配 | 均衡负载，减少进程间通信 | | 错误处理 | 异常捕获 + 重试机制 | 提高任务执行成功率 | | 进度监控 | `tqdm` 或自定义日志 | 实时查看执行进度 | ### 1.2 执行流程 ``` 读取SQL文件 → 分割为16个任务块 → 创建进程池 → 并行执行 → 收集结果 → 生成报告 ``` ## 二、完整实现代码 ```python import oracledb import logging from multiprocessing import Pool, cpu_count, Manager from typing import List, Tuple, Dict, Any import time from datetime import datetime import sys import re # 配置日志系统 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(processName)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(f'sql_executor_{datetime.now().strftime("%Y%m%d_%H%M%S")}.log'), logging.StreamHandler(sys.stdout) ] ) logger = logging.getLogger(__name__) class SQLParallelExecutor: def __init__(self, config: Dict[str, Any], max_workers: int = 16): """ 初始化并行SQL执行器 :param config: 数据库连接配置 :param max_workers: 最大并行进程数 """ self.db_config = config self.max_workers = min(max_workers, cpu_count() * 2) # 不超过CPU核心数的2倍 self.results = Manager().list() # 进程安全的共享列表 self.failures = Manager().list() # 存储失败的任务 def read_sql_file(self, file_path: str) -> List[str]: """ 读取SQL文件并分割为独立的SQL语句 [ref_2] :param file_path: SQL文件路径 :return: SQL语句列表 """ sql_statements = [] try: with open(file_path, 'r', encoding='utf-8') as f: content = f.read() # 使用分号分割SQL语句，并过滤空语句 raw_statements = [stmt.strip() for stmt in content.split(';') if stmt.strip()] for stmt in raw_statements: # 移除注释 [ref_4] stmt = re.sub(r'--.*$', '', stmt, flags=re.MULTILINE) # 移除单行注释 stmt = re.sub(r'/\*.*?\*/', '', stmt, flags=re.DOTALL) # 移除多行注释 stmt = stmt.strip() if stmt and not stmt.lower().startswith('--'): sql_statements.append(stmt) logger.info(f"从文件 {file_path} 读取到 {len(sql_statements)} 条有效SQL语句") return sql_statements except FileNotFoundError: logger.error(f"文件未找到: {file_path}") raise except Exception as e: logger.error(f"读取SQL文件时出错: {e}") raise def split_into_chunks(self, sql_list: List[str], num_chunks: int) -> List[List[str]]: """ 将SQL列表分割为指定数量的块 :param sql_list: SQL语句列表 :param num_chunks: 块数量 :return: 分割后的块列表 """ avg = len(sql_list) // num_chunks remainder = len(sql_list) % num_chunks chunks = [] start = 0 for i in range(num_chunks): end = start + avg + (1 if i < remainder else 0) if start < len(sql_list): chunks.append(sql_list[start:end]) start = end logger.info(f"将 {len(sql_list)} 条SQL分割为 {len(chunks)} 个任务块") return chunks def execute_sql_chunk(self, chunk_id: int, sql_chunk: List[str]) -> Dict[str, Any]: """ 单个工作进程执行的任务：执行一个SQL块 :param chunk_id: 块ID :param sql_chunk: SQL语句块 :return: 执行结果统计 """ process_name = f"Worker-{chunk_id:02d}" logger.info(f"{process_name}: 开始执行 {len(sql_chunk)} 条SQL语句") # 每个进程创建独立的数据库连接 connection = None results = { 'chunk_id': chunk_id, 'total': len(sql_chunk), 'success': 0, 'failed': 0, 'failed_details': [], 'execution_time': 0 } try: start_time = time.time() # 创建数据库连接 connection = oracledb.connect( user=self.db_config['user'], password=self.db_config['password'], dsn=self.db_config['dsn'] ) cursor = connection.cursor() # 逐条执行SQL语句 for i, sql in enumerate(sql_chunk, 1): try: # 根据SQL类型选择执行方式 [ref_3] sql_lower = sql.strip().lower() if sql_lower.startswith('select'): # 查询语句 cursor.execute(sql) # 对于查询，可以获取部分结果或仅执行 rows = cursor.fetchmany(100) # 限制返回行数，避免内存溢出 logger.debug(f"{process_name}: 查询语句执行成功，返回 {len(rows)} 行") else: # DML/DDL语句 cursor.execute(sql) if not sql_lower.startswith('commit') and not sql_lower.startswith('rollback'): connection.commit() # 自动提交非事务控制语句 # 获取影响行数（如果适用） try: rowcount = cursor.rowcount logger.debug(f"{process_name}: 执行成功，影响 {rowcount} 行") except: logger.debug(f"{process_name}: 执行成功") results['success'] += 1 # 进度提示（每50条输出一次） if i % 50 == 0: logger.info(f"{process_name}: 已处理 {i}/{len(sql_chunk)} 条SQL") except oracledb.Error as db_err: error_msg = f"Oracle错误: {db_err}" logger.error(f"{process_name}: SQL执行失败 - {sql[:100]}... - {error_msg}") results['failed'] += 1 results['failed_details'].append({ 'sql': sql[:500], # 截断长SQL 'error': str(db_err) }) # 失败后继续执行下一条SQL if connection: try: connection.rollback() except: pass continue except Exception as e: error_msg = f"未知错误: {e}" logger.error(f"{process_name}: SQL执行失败 - {sql[:100]}... - {error_msg}") results['failed'] += 1 results['failed_details'].append({ 'sql': sql[:500], 'error': str(e) }) continue results['execution_time'] = time.time() - start_time except oracledb.Error as e: logger.error(f"{process_name}: 数据库连接失败 - {e}") results['failed'] = len(sql_chunk) results['failed_details'].append({ 'sql': '连接失败', 'error': str(e) }) except Exception as e: logger.error(f"{process_name}: 进程执行异常 - {e}") results['failed'] = len(sql_chunk) finally: if connection: try: connection.close() except: pass logger.info(f"{process_name}: 执行完成，成功 {results['success']}/{results['total']}，耗时 {results['execution_time']:.2f}秒") return results def execute_parallel(self, sql_file_path: str) -> Dict[str, Any]: """ 主执行函数：并行执行SQL文件 :param sql_file_path: SQL文件路径 :return: 执行结果汇总 """ logger.info("=" * 60) logger.info(f"开始并行执行SQL文件: {sql_file_path}") logger.info(f"最大并行进程数: {self.max_workers}") logger.info("=" * 60) # 1. 读取SQL文件 try: sql_statements = self.read_sql_file(sql_file_path) if not sql_statements: logger.warning("SQL文件为空，无需执行") return {'total': 0, 'success': 0, 'failed': 0} except Exception as e: logger.error(f"读取SQL文件失败: {e}") return {'total': 0, 'success': 0, 'failed': 0, 'error': str(e)} # 2. 分割任务块 chunks = self.split_into_chunks(sql_statements, self.max_workers) # 3. 创建参数列表（为每个块分配ID） task_args = [(i, chunk) for i, chunk in enumerate(chunks) if chunk] # 4. 使用进程池并行执行 start_time = time.time() all_results = [] with Pool(processes=self.max_workers) as pool: # 使用starmap传递多个参数 results = pool.starmap(self.execute_sql_chunk, task_args) all_results.extend(results) total_time = time.time() - start_time # 5. 汇总结果 summary = self._summarize_results(all_results, total_time) # 6. 生成报告 self._generate_report(summary, all_results) return summary def _summarize_results(self, results: List[Dict], total_time: float) -> Dict[str, Any]: """汇总所有进程的执行结果""" total_sql = 0 total_success = 0 total_failed = 0 all_failures = [] for result in results: total_sql += result['total'] total_success += result['success'] total_failed += result['failed'] all_failures.extend(result['failed_details']) summary = { 'total_sql_statements': total_sql, 'successful_executions': total_success, 'failed_executions': total_failed, 'success_rate': (total_success / total_sql * 100) if total_sql > 0 else 0, 'total_execution_time': total_time, 'avg_time_per_sql': total_time / total_sql if total_sql > 0 else 0, 'failures': all_failures[:100], # 只保留前100个失败详情 'total_failures_count': len(all_failures) } return summary def _generate_report(self, summary: Dict, detailed_results: List[Dict]): """生成执行报告""" logger.info("\n" + "=" * 60) logger.info("SQL并行执行完成报告") logger.info("=" * 60) logger.info(f"总SQL语句数: {summary['total_sql_statements']}") logger.info(f"成功执行: {summary['successful_executions']}") logger.info(f"执行失败: {summary['failed_executions']}") logger.info(f"成功率: {summary['success_rate']:.2f}%") logger.info(f"总执行时间: {summary['total_execution_time']:.2f}秒") logger.info(f"平均每条SQL耗时: {summary['avg_time_per_sql']:.4f}秒") if summary['failed_executions'] > 0: logger.warning(f"\n失败详情（共 {summary['total_failures_count']} 条失败）:") for i, failure in enumerate(summary['failures'][:10], 1): logger.warning(f"{i}. SQL: {failure['sql'][:200]}...") logger.warning(f" 错误: {failure['error']}") if summary['total_failures_count'] > 10: logger.warning(f"... 以及另外 {summary['total_failures_count'] - 10} 条失败记录") # 各进程详细统计 logger.info("\n各进程执行详情:") logger.info("-" * 60) for result in detailed_results: logger.info(f"进程 {result['chunk_id']:2d}: " f"成功 {result['success']:4d}/" f"失败 {result['failed']:4d} " f"耗时 {result['execution_time']:7.2f}秒") # 使用示例 if __name__ == '__main__': # 数据库配置 DB_CONFIG = { 'user': 'your_username', 'password': 'your_password', 'dsn': 'localhost:1521/orcl', # 根据实际情况修改 'encoding': 'UTF-8' } # SQL文件路径 SQL_FILE = 'your_sql_file.sql' # 替换为你的SQL文件路径 # 创建执行器并执行 try: executor = SQLParallelExecutor(DB_CONFIG, max_workers=16) results = executor.execute_parallel(SQL_FILE) # 输出最终摘要 print("\n" + "=" * 60) print("执行摘要:") print(f"总语句数: {results['total_sql_statements']}") print(f"成功率: {results['success_rate']:.1f}%") print(f"总耗时: {results['total_execution_time']:.1f}秒") if results['failed_executions'] > 0: print(f"\n警告: 有 {results['failed_executions']} 条语句执行失败") print("详细错误信息请查看日志文件") except Exception as e: logger.error(f"执行过程中发生错误: {e}") sys.exit(1) ``` ## 三、高级功能扩展 ### 3.1 支持事务控制 ```python def execute_with_transaction(self, sql_chunk: List[str], connection): """支持事务的批量执行""" try: cursor = connection.cursor() # 按语句类型分组执行 ddl_statements = [] dml_statements = [] for sql in sql_chunk: sql_lower = sql.strip().lower() if any(sql_lower.startswith(keyword) for keyword in ['create ', 'alter ', 'drop ', 'truncate ']): ddl_statements.append(sql) else: dml_statements.append(sql) # 执行DDL（自动提交） for ddl in ddl_statements: cursor.execute(ddl) # 执行DML（批量提交） for dml in dml_statements: cursor.execute(dml) connection.commit() return True except Exception as e: connection.rollback() raise ``` ### 3.2 性能优化建议 1. **连接池配置**： ```python # 使用连接池提高性能 pool = oracledb.create_pool( user=DB_CONFIG['user'], password=DB_CONFIG['password'], dsn=DB_CONFIG['dsn'], min=5, max=20, increment=2 ) ``` 2. **批量执行优化**： ```python # 对于INSERT/UPDATE语句，使用executemany进行批量操作 def batch_execute(self, sql_template: str, data_list: List[tuple]): """批量执行相同模板的SQL""" cursor = connection.cursor() cursor.executemany(sql_template, data_list) connection.commit() ``` ## 四、最佳实践建议 ### 4.1 执行前的准备工作 1. **SQL验证**：执行前验证SQL语法是否正确 2. **权限检查**：确保数据库用户有执行所有SQL的权限 3. **备份数据**：对于UPDATE/DELETE操作，先备份相关数据 4. **测试环境验证**：先在测试环境执行验证 ### 4.2 监控与调优 | 监控指标 | 建议值 | 调整策略 | |----------|--------|----------| | 数据库连接数 | < 最大连接数的80% | 调整进程数 | | CPU使用率 | < 70% | 控制并发度 | | 内存使用 | < 80% | 分批执行 | | 执行时间 | 监控异常长SQL | 优化SQL或设置超时 | ### 4.3 错误处理策略 1. **重试机制**：对网络超时等临时错误进行重试 2. **跳过机制**：对某些可跳过的错误继续执行后续SQL 3. **断点续传**：记录执行进度，支持从中断点继续执行 ## 五、执行示例 假设SQL文件内容如下（`test.sql`）： ```sql -- 创建测试表 CREATE TABLE test_table1 (id NUMBER, name VARCHAR2(50)); -- 插入测试数据 INSERT INTO test_table1 VALUES (1, 'Alice'); INSERT INTO test_table1 VALUES (2, 'Bob'); INSERT INTO test_table1 VALUES (3, 'Charlie'); -- 查询数据 SELECT * FROM test_table1 WHERE id > 1; -- 更新数据 UPDATE test_table1 SET name = 'David' WHERE id = 1; -- 创建索引