本文介绍如何通过分离耗时计算与轻量 I/O，利用 concurrent.futures 安全、鲁棒地并行处理大规模字符串列表，并将结果可靠写入 CSV 文件——避免共享 csv.writer 导致的序列化失败、线程竞争与进程挂起问题。

本文介绍如何通过分离耗时计算与轻量 i/o，利用 `concurrent.futures` 安全、鲁棒地并行处理大规模字符串列表，并将结果可靠写入 csv 文件——避免共享 `csv.writer` 导致的序列化失败、线程竞争与进程挂起问题。

在 Python 中对大批量文本（如医学术语、用户查询、日志条目）进行 CPU 或 I/O 密集型处理时，直接使用多线程/多进程写 CSV 常引发严重问题：csv.writer 对象不可被 pickle（导致 ProcessPoolExecutor 报 TypeError），且多线程并发调用 writerow() 会因共享文件句柄引发竞态条件、数据错乱甚至死锁。你遇到的超时后进程挂起、线程无法回收、批量中断难恢复等问题，根源正在于将计算逻辑与 I/O 操作耦合在同一函数中，并试图跨进程/线程共享状态。

正确的解法是遵循“计算与输出分离”原则：
✅ 让并行任务只负责纯计算——输入字符串，返回结构化结果（如 list 或 dict），不触碰任何文件或全局状态；
✅ 将所有 I/O（CSV 写入、错误记录）移至主线程单线程串行执行，确保原子性与可预测性；
✅ 利用 concurrent.futures 的异常捕获机制实现容错：单个任务失败不影响整体流程，错误可单独记录。

以下是经过生产验证的完整实现：

import csv
import logging
import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

# 配置结构化日志（便于排查超时/异常）
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-8s | %(message)s",
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler("processing.log", mode="a")
    ]
)

# 示例原始数据（实际中可能来自文件或数据库）
term_list = [
    "Dementia", "HER2-positive Breast Cancer", "Stroke", "Hemiplegia",
    "Type 1 Diabetes", "IBD", "Lung Cancer", "Psoriasis", "Healthy",
    "Asthma", "Obesity", "Schizophrenia", "Cancer", "COVID-19"
]

def run_mappers(individual_string: str, other_args) -> list:
    """
    核心处理函数：仅执行业务逻辑，严格禁止 I/O 或修改全局状态。
    返回值必须是可序列化的（如 list/tuple/dict），供主进程收集。
    """
    try:
        # ✅ 模拟真实耗时操作（网络请求、NLP 分析、规则匹配等）
        time.sleep(0.1 + len(individual_string) * 0.005)  # 避免过快掩盖并发效果

        # 示例处理：标准化术语 + 添加元信息
        normalized = individual_string.strip().title()
        processed_result = [normalized, other_args, len(normalized), int(time.time() % 1000)]

        logging.debug("✓ Processed: %r → %s", individual_string, processed_result)
        return processed_result

    except Exception as e:
        # ❌ 不在此处处理异常，由主流程统一捕获
        logging.error("✗ Failed on %r: %s", individual_string, e)
        raise  # 让 future.exception() 可检测

def parallel_process_and_write(
    term_list: list,
    other_args,
    output_csv: str = "results.csv",
    error_log: str = "errors.txt",
    max_workers: int = 4,
    executor_type: str = "thread"  # "thread" or "process"
):
    """
    主协调函数：并行计算 → 收集结果 → 单线程写入
    """
    logging.info("Starting parallel processing of %d terms...", len(term_list))

    # Step 1: 并行提交所有任务（无状态、无共享）
    ExecutorClass = ThreadPoolExecutor if executor_type == "thread" else ProcessPoolExecutor
    with ExecutorClass(max_workers=max_workers) as executor:
        # 提交任务：每个 term 独立运行，返回 Future 对象
        futures = [
            executor.submit(run_mappers, term, other_args)
            for term in term_list
        ]

        # Step 2: 同步等待全部完成（支持 timeout，但不中断已启动任务）
        # 注意：as_completed 是可选优化，此处用 list comprehension 确保顺序无关性
        results = []
        errors = []
        for i, future in enumerate(futures):
            try:
                result = future.result(timeout=180)  # 单任务超时 3 分钟
                results.append(result)
                logging.debug("Task %d completed successfully.", i+1)
            except Exception as exc:
                error_msg = f"[Term {i+1}: {term_list[i]}] {type(exc).__name__}: {exc}"
                errors.append(error_msg)
                logging.warning("Task %d failed: %s", i+1, error_msg)

    # Step 3: 主线程单次写入 CSV（绝对安全）
    logging.info("Writing %d successful results to %s...", len(results), output_csv)
    with open(output_csv, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        # 写入表头（按需调整）
        writer.writerow(["Term", "OtherArgs", "Length", "Timestamp"])
        writer.writerows(results)

    # Step 4: 记录所有失败项（便于重试或审计）
    if errors:
        logging.warning("Encountered %d errors. Details written to %s", len(errors), error_log)
        with open(error_log, "w", encoding="utf-8") as f:
            f.write("Failed Processing Log\n" + "="*50 + "\n")
            f.write("\n".join(errors))

    logging.info("Processing completed. %d success, %d failed.", len(results), len(errors))

# 使用示例
if __name__ == "__main__":
    # ✅ 推荐：I/O 密集型（如含 HTTP 请求）用 ThreadPoolExecutor
    parallel_process_and_write(
        term_list=term_list,
        other_args="v2.1-annotation",
        output_csv="terms_output.csv",
        error_log="term_errors.log",
        max_workers=6,
        executor_type="thread"
    )

    # ⚠️ 备选：纯 CPU 密集型（如复杂 NLP 模型推理）可尝试 ProcessPoolExecutor
    # parallel_process_and_write(..., executor_type="process")

关键设计说明与注意事项

• 为什么不用 ThreadPoolExecutor 共享 csv.writer？
  csv.writer 包含内部缓冲区和文件句柄，多线程并发调用 writerow() 会导致写入交错、换行丢失、部分行被截断。即使加锁，也会严重抵消并发收益。

• 超时处理为何更健壮？
  future.result(timeout=180) 仅限制单个任务等待时间，超时后抛出 concurrent.futures.TimeoutError，主线程立即记录错误并继续处理下一个 future，不会阻塞或挂起整个池。这比手动管理 threading.Event 或 terminate_flag 更简洁可靠。

• ThreadPoolExecutor vs ProcessPoolExecutor 怎么选？

  • ✅ ThreadPoolExecutor：适用于 I/O 密集型任务（HTTP 调用、数据库查询、文件读取）。Python 的 GIL 在 I/O 时自动释放，线程能真正并发。
  • ✅ ProcessPoolExecutor：适用于 CPU 密集型任务（正则爆炸、数值计算、模型推理）。绕过 GIL，但有进程启动开销和对象序列化成本。建议先用线程池测试，若 CPU 使用率低且耗时长，再切换为进程池并 benchmark。

• 内存与批量控制（呼应你的需求）
  若 term_list 极大（千万级），可分批处理：

  batch_size = 1000
  for i in range(0, len(term_list), batch_size):
      batch = term_list[i:i+batch_size]
      parallel_process_and_write(batch, other_args, 
                               output_csv=f"batch_{i//batch_size}.csv")

  每批独立执行，失败批次可单独重试，彻底规避单次长任务失控风险。

• 生产环境增强建议

  • 添加 tqdm 进度条：for future in tqdm(as_completed(futures), total=len(futures))
  • 结果去重/校验：在写入前对 results 做 set(tuple(r) for r in results)
  • CSV 编码容错：open(..., encoding="utf-8-sig") 防止 Excel 打开乱码
  • 错误重试：对 TimeoutError 或特定异常，可封装 retrying 逻辑（需确保 run_mappers 幂等）

通过此方案，你将获得：
? 高吞吐：并行压榨 CPU/I/O 资源；
? 强鲁棒：单任务崩溃不阻塞全局，超时自动跳过；
? 易维护：计算与 I/O 解耦，逻辑清晰，调试简单；
? 可扩展：无缝支持分批、重试、监控与分布式迁移（如改用 dask 或 Celery）。

现在，只需将你真实的 run_mappers() 业务逻辑填入函数体，保持其「纯计算、无副作用」特性，即可安全投入生产。