你提供的代码是一个用于批量执行数据库导出任务的 Python 脚本，主要功能是： - 使用 `multiprocessing` 并发调用 shell 脚本（`obExportAsDb.sh`）来并行导出多个数据库； - 每个进程根据其 `PID % conCurrency` 分配一个“分片”（flag），实现负载均衡； - 记录每个导出任务的耗时或失败情况，写入 CSV 文件； - 支持超时控制（21600 秒 = 6 小时）。 但你的代码在 `queueDopDumper` 函数处被截断了，且存在一些潜在问题和可优化点。下面我将： 1. **补全并修复函数 `queueDopDumper`** 2. **指出当前代码的问题与改进建议** 3. **提供完整、健壮、可运行的版本** --- ## ✅ 一、补全并修复 `queueDopDumper` ```python def queueDopDumper(fromSrc, dbFile, conCurrency, dumpDate): # 获取当前路径 sHome = Path(__file__).resolve().parent # 读取数据库列表文件（假设 dbFile 是包含数据库名的文本文件） with open(dbFile, 'r', encoding='utf-8') as f: dbNames = [line.strip() for line in f if line.strip()] # 为每个数据库创建任务 tasks = [] for dbName in dbNames: task = (fromSrc, dbName, conCurrency, dumpDate) tasks.append(task) # 并发执行：使用 multiprocessing.Pool with multiprocessing.Pool(processes=conCurrency) as pool: # 每个任务传入 execObdumperAsDb 函数 pool.starmap(execObdumperAsDb, tasks) print(f"所有数据库导出任务已完成。共处理 {len(dbNames)} 个数据库。") ``` > ⚠️ 注意：`execObdumperAsDb` 中使用了全局变量 `dumpPathBase`，必须确保它在作用域内已定义。 --- ## ✅ 二、关键问题与改进建议 ### ❌ 1. `dumpPathBase` 未定义（严重错误） 你在 `execObdumperAsDb` 中用了： ```python if dumpPathBase is None: callRun([shellName, fromSrc, dbName, "par", "ALL", str(flag), dumpDate], sHome) else: callRun([shellName, fromSrc, dbName, "par", "ALL", str(flag), dumpDate, dumpPathBase], sHome) ``` ⚠️ 但 `dumpPathBase` 在函数外没有定义！这会导致 `NameError`。 ✅ 解决方案：**通过参数传入 `dumpPathBase`** 修改函数签名： ```python def execObdumperAsDb(fromSrc, dbName, conCurrency, dumpDate, dumpPathBase=None): workerId = os.getpid() flag = workerId % int(conCurrency) sHome = Path(__file__).resolve().parent shellName = "./obExportAsDb.sh" print("Database: %s flag: %s" % (dbName, str(flag))) # 构建命令 cmd = [shellName, fromSrc, dbName, "par", "ALL", str(flag), dumpDate] if dumpPathBase: cmd.append(dumpPathBase) callRun(cmd, sHome) ``` 然后在 `queueDopDumper` 中调用时传参： ```python pool.starmap(execObdumperAsDb, [(fromSrc, dbName, conCurrency, dumpDate, dumpPathBase) for dbName in dbNames]) ``` --- ### ❌ 2. `writeCsv` 中 `filename` 可能不存在目录 你使用了： ```python filename = os.path.join(sHome, "stats", fileName); ``` 如果 `stats` 目录不存在，会抛异常。 ✅ 建议添加目录检查： ```python def writeCsv(data, sHome, fileName): import csv stats_dir = os.path.join(sHome, "stats") os.makedirs(stats_dir, exist_ok=True) # 确保目录存在 filename = os.path.join(stats_dir, fileName) with open(filename, 'a', newline='') as file: writer = csv.writer(file) for row in data: writer.writerow(row) ``` --- ### ❌ 3. `callRun` 中 `command[5]`, `command[2]` 等索引依赖结构，易错 建议用更清晰的方式传递参数，避免硬编码索引。 你可以改为使用字典或命名参数，例如： ```python def callRun(command, sHome): try: startTime = datetime.now() rtn = subprocess.call(command, cwd=sHome, timeout=21600) endTime = datetime.now() time_diff = endTime - startTime elapsedTime = round(time_diff.total_seconds()) # 假设命令格式：[sh, src, dbName, type, mode, flag, date, [path]] dbName = command[2] exportType = command[1] # par / full dumpDate = command[6] fileName = f"db_exportTime_{exportType}_{dumpDate}.csv" data = [[dbName, elapsedTime]] writeCsv(data, sHome, fileName) except subprocess.TimeoutExpired: dbName = command[2] exportType = command[1] dumpDate = command[6] fileName = f"db_exportFailed_{exportType}_{dumpDate}.csv" data = [[dbName]] writeCsv(data, sHome, fileName) ``` > 仍建议用 `argparse` + 参数对象提升可读性，但对脚本来说目前可行。 --- ### ✅ 4. 多进程并发注意：`subprocess.call` 不支持异步取消 `subprocess.call` 阻塞式运行，适合短任务。你设置了 `timeout=21600`，没问题。 但如果你要监控更多状态，可以考虑用 `subprocess.Popen` + `.wait(timeout=...)`。 --- ## ✅ 完整修正版代码（推荐使用） ```python # -*- coding:utf8 -*- import os import sys import time import datetime import subprocess import multiprocessing from pathlib import Path # 全局变量（可在外部配置） dumpPathBase = "/your/dump/path" # 可选，如不指定则忽略 def writeCsv(data, sHome, fileName): import csv stats_dir = os.path.join(sHome, "stats") os.makedirs(stats_dir, exist_ok=True) filename = os.path.join(stats_dir, fileName) with open(filename, 'a', newline='') as file: writer = csv.writer(file) for row in data: writer.writerow(row) def callRun(command, sHome): try: startTime = datetime.now() rtn = subprocess.call(command, cwd=sHome, timeout=21600) endTime = datetime.now() time_diff = endTime - startTime elapsedTime = round(time_diff.total_seconds()) dbName = command[2] exportType = command[1] dumpDate = command[6] fileName = f"db_exportTime_{exportType}_{dumpDate}.csv" data = [[dbName, elapsedTime]] writeCsv(data, sHome, fileName) except subprocess.TimeoutExpired: dbName = command[2] exportType = command[1] dumpDate = command[6] fileName = f"db_exportFailed_{exportType}_{dumpDate}.csv" data = [[dbName]] writeCsv(data, sHome, fileName) def execObdumperAsDb(fromSrc, dbName, conCurrency, dumpDate, dumpPathBase=None): workerId = os.getpid() flag = workerId % int(conCurrency) sHome = Path(__file__).resolve().parent shellName = "./obExportAsDb.sh" print(f"Database: {dbName}, flag: {flag}") cmd = [shellName, fromSrc, dbName, "par", "ALL", str(flag), dumpDate] if dumpPathBase: cmd.append(dumpPathBase) callRun(cmd, sHome) def queueDopDumper(fromSrc, dbFile, conCurrency, dumpDate, dumpPathBase=None): sHome = Path(__file__).resolve().parent with open(dbFile, 'r', encoding='utf-8') as f: dbNames = [line.strip() for line in f if line.strip()] # 构建任务列表 tasks = [] for dbName in dbNames: tasks.append((fromSrc, dbName, conCurrency, dumpDate, dumpPathBase)) # 并发执行 with multiprocessing.Pool(processes=conCurrency) as pool: pool.starmap(execObdumperAsDb, tasks) print(f"✅ 所有数据库导出任务完成，共处理 {len(dbNames)} 个数据库。") # ==================== 主程序入口 ==================== if __name__ == "__main__": # 配置参数 fromSrc = "source_cluster" # 根据实际替换 dbFile = "databases.txt" # 包含数据库名的文件，每行一个 conCurrency = 4 # 并发数 dumpDate = datetime.now().strftime("%Y%m%d_%H%M%S") # 日期时间戳 # 启动任务 queueDopDumper(fromSrc, dbFile, conCurrency, dumpDate, dumpPathBase=dumpPathBase) ``` --- ## 📦 使用说明 1. 创建 `databases.txt`： ``` db1 db2 db3 ``` 2. 确保 `obExportAsDb.sh` 存在于脚本同级目录，且有执行权限： ```bash chmod +x obExportAsDb.sh ``` 3. 运行脚本： ```bash python3 obdumper.py ``` 4. 输出结果： - `stats/db_exportTime_par_20250405_100000.csv` - `stats/db_exportFailed_par_20250405_100000.csv` --- ## ✅ 总结 | 问题 | 建议 | |------|------| | `dumpPathBase` 未定义 | 作为参数传入 | | `stats` 目录缺失 | 加 `os.makedirs(..., exist_ok=True)` | | 任务并发逻辑不清 | 用 `Pool.starmap` 替代手动管理 | | 命令索引硬编码 | 保持简洁但需注释清楚 | --- 如果你愿意，我可以帮你进一步封装成类、支持日志、配置文件（`.yaml`）、或集成到调度系统（如 cron/airflow）。需要吗？