大数据

ThreadPoolExecutor + 后台轮询线程协作指南

Posted by

在实际项目中，我们常常需要同时处理两类并发任务：一类是 CPU/IO 密集型的并行执行任务（如批量文档处理），另一类是需要在后台持续轮询、监控主任务进度的守护任务（如检测任务队列状态、上报进度等）。这两类任务如何协作，往往比单独使用线程池或单独使用后台线程更复杂。

本文通过一个简化的数据处理流水线 Demo，展示如何将 ThreadPoolExecutor 和 threading.Thread 结合使用。主线程使用 ThreadPoolExecutor 并发执行一批模拟的文档处理任务，同时启动一个独立的 threading.Thread 后台线程，轮询检测任务进度并实时打印状态。主线程通过 as_completed 阻塞等待所有任务完成后，再通过 Event 通知后台线程优雅退出，最后基于所有任务结果执行汇总操作。

import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime


def process_document(doc_id: int) -> dict:
    """模拟文档处理任务（耗时操作）"""
    duration = 1 + (doc_id % 3)  # 1~3 秒
    print(f"[任务 {doc_id}] 开始处理，预计耗时 {duration}s")
    time.sleep(duration)
    print(f"[任务 {doc_id}] 处理完成")
    return {"doc_id": doc_id, "status": "completed", "duration": duration}


def background_monitor(total_tasks: int, stop_event: threading.Event, check_interval: float = 0.5):
    """
    后台轮询线程：监控任务进度

    参数：
        total_tasks: 总任务数
        stop_event: 停止信号，主线程通知后台线程退出
        check_interval: 轮询间隔（秒）
    """
    completed = 0
    print(f"[监控] 启动，总任务数: {total_tasks}，轮询间隔: {check_interval}s")

    while not stop_event.is_set():
        time.sleep(check_interval)
        # 这里模拟读取进度（实际项目中可从共享状态/数据库中读取）
        # 为了演示，我们用随机数模拟进度
        import random
        if random.random() < 0.3:  # 模拟进度变化
            completed = min(completed + 1, total_tasks)

        progress = (completed / total_tasks) * 100 if total_tasks > 0 else 0
        print(f"[监控] 进度: {completed}/{total_tasks} ({progress:.1f}%)")

    print("[监控] 收到停止信号，退出")


def main():
    print("=" * 60)
    print("文档流水线模拟开始")
    print("=" * 60)

    # 1. 定义任务列表
    doc_ids = [1, 2, 3, 4, 5, 6]
    total_tasks = len(doc_ids)

    # 2. 创建停止事件（用于通知后台线程退出）
    stop_event = threading.Event()

    # 3. 启动后台监控线程
    monitor_thread = threading.Thread(
        target=background_monitor,
        args=(total_tasks, stop_event, 0.5),
        daemon=False,  # 非守护线程，主线程会等待它结束
        name="BackgroundMonitor"
    )
    monitor_thread.start()
    print("[主线程] 后台监控线程已启动")

    # 4. 主线程：使用 ThreadPoolExecutor 并发执行任务
    results = []
    print(f"[主线程] 提交 {total_tasks} 个任务到线程池...")
    start_time = time.time()

    with ThreadPoolExecutor(max_workers=3) as executor:
        # 提交所有任务
        future_to_doc = {
            executor.submit(process_document, doc_id): doc_id
            for doc_id in doc_ids
        }

        # 等待所有任务完成（as_completed 阻塞直到有任务完成）
        # 谁先完成先处理谁，实时获取结果
        print("[主线程] 等待所有任务完成...")
        for future in as_completed(future_to_doc):
            doc_id = future_to_doc[future]
            try:
                result = future.result()
                results.append(result)
                print(f"[主线程] 任务 {doc_id} 的结果已收集")
            except Exception as e:
                print(f"[主线程] 任务 {doc_id} 执行失败: {e}")

    elapsed = time.time() - start_time
    print(f"[主线程] 所有任务执行完成，耗时: {elapsed:.2f}s")

    # 5. 所有任务完成后，通知后台线程退出
    print("[主线程] 通知后台监控线程退出...")
    stop_event.set()

    # 6. 等待后台线程真正结束
    monitor_thread.join()
    print("[主线程] 后台监控线程已退出")

    # 7. 执行后续操作（基于所有任务的结果）
    print("\n" + "=" * 60)
    print("📊 最终结果汇总")
    print("=" * 60)
    print(f"成功处理文档数: {len(results)}")
    for r in results:
        print(f"  - 文档 {r['doc_id']}: {r['status']}，耗时 {r['duration']}s")
    print(f"总耗时: {elapsed:.2f}s")
    print("流水线结束 ✅")


if __name__ == "__main__":
    main()
步骤操作主线程状态监控线程状态
第3步monitor_thread.start()立即继续执行开始工作（轮询）
第4步提交任务到线程池，as_completed 等待所有任务完成阻塞等待，但不会被监控线程卡住继续轮询，打印进度
第5步stop_event.set()发送停止信号收到信号，退出循环
第6步monitor_thread.join()等待监控线程真正结束已经退出，立即响应

上一篇 数据治理pipeline-demo