如何用Python优雅实现多阶段数据合并的并发处理

问题背景 #

需要实现多阶段数据合并/流水线的并发处理，替代现有结构复杂的process_concurrent()函数。要求任一阶段出现异常的数据行跳过，输出顺序无关紧要。

现有方案的核心不足 #

• 内存累积问题：当球员隶属于多支球队时，player_age字典会持续增长，无法有效释放内存
• 结果反馈滞后：第4阶段（输出）虽能提前启动，但需等待所有球队处理完成才能获取结果，I/O失败时无法提前终止或重试
• 阶段阻塞限制：所有阶段均存在“任务提交后需等前一阶段全部完成，才能检查本阶段已完成任务结果”的问题

优化方案 #

思路1：单球队全链路并发+缓存控制 #

将单支球队的完整处理流程（阶段1到阶段4）封装为独立任务，利用进程池并发执行，同时对球员年龄查询做缓存并限制大小，既避免重复查询，又控制内存增长；完成一支球队的处理后立即输出结果，解决反馈滞后问题。

from concurrent import futures
from time import sleep
from random import random
from functools import lru_cache

# 原有team_players、player_age数据及phase1-phase4函数保持不变，此处省略

@lru_cache(maxsize=128)  # 限制缓存容量，避免内存无限膨胀
def cached_phase2_lookup_player_age(player):
    return phase2_lookup_player_age(player)

def process_single_team(team):
    """处理单支球队的完整流程，异常时返回None跳过"""
    try:
        # 阶段1：获取球队球员列表
        players = phase1_lookup_team_players(team)
        # 阶段2：并发查询球员年龄
        with futures.ThreadPoolExecutor() as executor:
            ages = list(executor.map(cached_phase2_lookup_player_age, players))
        # 阶段3：合并数据为记录列表
        team_records = phase3_merge(team, players, ages)
        # 阶段4：生成CSV字符串
        return phase4_make_csv_table(team, team_records)
    except Exception:
        return None

def process_concurrent(teams):
    """优化后的并发实现：简洁高效，实时输出结果"""
    with futures.ProcessPoolExecutor() as executor:
        # 提交所有球队处理任务
        future_map = {executor.submit(process_single_team, t): t for t in teams}
        # 实时处理已完成的任务结果
        for future in futures.as_completed(future_map):
            result = future.result()
            if result:
                print(result)

思路2：流水线式并发（基于多进程队列） #

用队列串联各个阶段，每个阶段作为独立工作进程，数据在队列中流动，实现真正的流水线处理——每个阶段一有数据就立即处理，无需等待前一阶段全部完成；数据处理后即被消费，从根源避免内存累积。

from multiprocessing import Process, Queue
from time import sleep
from random import random
from functools import lru_cache

# 原有team_players、player_age数据及phase1-phase4函数保持不变，此处省略

@lru_cache(maxsize=128)
def cached_phase2_lookup_player_age(player):
    return phase2_lookup_player_age(player)

def phase1_worker(input_q, output_q):
    """阶段1工作进程：处理球队，输出(球队, 球员列表)"""
    while True:
        team = input_q.get()
        if team is None:  # 终止信号
            break
        try:
            players = phase1_lookup_team_players(team)
            output_q.put((team, players))
        except Exception:
            continue

def phase2_worker(input_q, output_q):
    """阶段2工作进程：处理球员列表，输出(球队, 球员列表, 年龄列表)"""
    while True:
        item = input_q.get()
        if item is None:
            break
        team, players = item
        try:
            ages = [cached_phase2_lookup_player_age(p) for p in players]
            output_q.put((team, players, ages))
        except Exception:
            continue

def phase3_worker(input_q, output_q):
    """阶段3工作进程：合并数据，输出(球队, 记录列表)"""
    while True:
        item = input_q.get()
        if item is None:
            break
        team, players, ages = item
        try:
            records = phase3_merge(team, players, ages)
            output_q.put((team, records))
        except Exception:
            continue

def phase4_worker(input_q):
    """阶段4工作进程：实时输出CSV结果"""
    while True:
        item = input_q.get()
        if item is None:
            break
        team, records = item
        try:
            print(phase4_make_csv_table(team, records))
        except Exception:
            continue

def process_concurrent(teams):
    """流水线并发实现：各阶段独立工作，数据实时流动"""
    # 创建阶段间的传递队列
    q1 = Queue()
    q2 = Queue()
    q3 = Queue()
    q4 = Queue()

    # 初始化工作进程
    workers = [
        Process(target=phase1_worker, args=(q1, q2)),
        Process(target=phase2_worker, args=(q2, q3)),
        Process(target=phase3_worker, args=(q3, q4)),
        Process(target=phase4_worker, args=(q4,))
    ]

    # 启动所有工作进程
    for worker in workers:
        worker.start()

    # 输入待处理的球队任务
    for team in teams:
        q1.put(team)

    # 发送终止信号给每个阶段
    q1.put(None)
    q2.put(None)
    q3.put(None)
    q4.put(None)

    # 等待所有工作进程完成
    for worker in workers:
        worker.join()

方案优势 #

• 内存可控：通过lru_cache限制缓存大小，或流水线的实时消费机制，避免内存中累积大量未处理数据
• 实时反馈：两种方案均能在单支球队处理完成后立即输出结果，I/O失败可及时触发重试或终止逻辑
• 结构简洁：移除了原有复杂的状态管理字典，逻辑清晰，易于维护和扩展
• 异常容错：每个阶段的异常都会被捕获并跳过出错数据，不影响其他任务的正常执行

内容的提问来源于stack exchange，提问作者Victor Olex