并发、子图与结果汇聚

并发是让多个任务在同一段时间内推进。查订单时通常要等服务返回，查退款规则也可能要等资料系统；两项查询互不依赖，可以一起发起。等两项都返回，再整理答复，这个收集结果的步骤叫汇聚，也常称为汇合。

像去食堂打饭：同伴排饮料，你排饭菜，最后在座位会合。排队等待可以重叠，会合时要检查饭菜和饮料是否都拿到了。并发查询也是一样：同时开始，分别记录结果，最后按照需要组合。

先把依赖关系画出来

上一节用条件边选择一条路线；这里的一次任务会分成两条同时推进的分支：

• 订单分支：读取付款记录，再整理付款状态和金额。
• 规则分支：查询退款到账时间。
• 汇总节点：收到两个结果后，生成客服能读懂的结论。

订单分支内部仍有先后顺序。读取订单完成后，才能整理金额。这一组小步骤可以包装成子图，也就是主流程中的一个小流程。外部调用它时，只关心输入和最终返回值。

运行正常与部分失败两种情况

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l4-concurrency" -Force
cd "F:\学习\l4-concurrency"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

新建 F:\学习\l4-concurrency 文件夹，保存下面的 concurrency_demo.py。数据和失败情况已放在程序中，执行一次就能看到两组结果。

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l4-concurrency" -Force
cd "F:\学习\l4-concurrency"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

"""并行查询订单和退款规则，再汇总为客服可读的结论。"""
import asyncio
from dataclasses import dataclass


@dataclass(frozen=True)
class Part:
    name: str
    status: str
    value: str


async def order_query(fail=False):
    await asyncio.sleep(0.02)
    if fail:
        raise ConnectionError("订单服务暂时不可用")
    return {"paid": True, "amount": 80}


def order_summary(order):
    # 子流程的第二步把原始记录整理为主流程需要的字段。
    return f"已付款，金额{order['amount']}元"


async def order_subgraph(fail=False):
    order = await order_query(fail)
    return order_summary(order)


async def policy_query():
    await asyncio.sleep(0.01)
    return "退款通常3至5个工作日到账"


async def guarded(name, operation):
    try:
        value = await operation
        return Part(name, "ok", value)
    except ConnectionError as error:
        # 预期的连接失败保留为该分支的结果，其余分支仍能返回。
        return Part(name, "failed", str(error))


async def collect(fail_order=False):
    parts = await asyncio.gather(
        guarded("订单", order_subgraph(fail_order)),
        guarded("规则", policy_query()),
    )
    # gather 按传入顺序返回；用名称做索引，汇总时更加直观。
    by_name = {part.name: part for part in parts}
    if all(part.status == "ok" for part in parts):
        answer = f"{by_name['订单'].value}；{by_name['规则'].value}"
    else:
        answer = "规则已查到，订单信息待补查"
    return parts, answer


async def main():
    for label, failed in (("正常", False), ("订单连接失败", True)):
        parts, answer = await collect(failed)
        print(label)
        for part in parts:
            print(f"{part.name}：{part.status}；{part.value}")
        print(f"汇总：{answer}")


if __name__ == "__main__":
    asyncio.run(main())

打开 PowerShell：

cd "F:\学习\l4-concurrency"
ls
python -X utf8 concurrency_demo.py

程序输出：

正常
订单：ok；已付款，金额80元
规则：ok；退款通常3至5个工作日到账
汇总：已付款，金额80元；退款通常3至5个工作日到账
订单连接失败
订单：failed；订单服务暂时不可用
规则：ok；退款通常3至5个工作日到账
汇总：规则已查到，订单信息待补查

第一组两项资料都齐全，汇总把它们组合起来。第二组订单查询失败，但规则仍能查询成功，最终结论明确保留“订单信息待补查”。

这里的 asyncio.sleep 模拟等待服务返回。两条查询在一个线程中交替推进，等待可以重叠。例子用短暂等待保持运行方便；实际网络耗时可以变化。

从发起查询到结果汇合

collect 是主流程。它调用 asyncio.gather，把两个可等待的任务一起交给事件循环，再等待它们都产生结果。

订单分支经过两步：

1. order_query 等待后返回字典，里面有付款状态和金额。
2. order_summary 读取字典，返回“已付款，金额80元”。

order_subgraph 把两步接起来，所以它是一个子流程入口。主流程可以把它和单步的 policy_query 一起调度。以后在订单查询后增加字段检查，只需调整这个子流程内部。

gather 的返回列表按传入顺序排列，即使规则查询更早完成，结果列表的第一项仍是订单。接着程序按 name 建立字典，汇总可以直接取“订单”和“规则”，更容易阅读。

为什么每条分支要带状态

Part 有三个字段：

@dataclass(frozen=True) 把这三项整理成结果记录，建立后保持字段不变；主流程汇总这些记录，而不是再改动分支给出的结果。

guarded 捕获这里预先安排的连接错误，将它变成一项失败结果。主流程因此可以同时看到“订单失败”和“规则成功”，决定补查订单。

捕获的范围很具体：只有 ConnectionError 会被转成 failed。字段错误等其他异常继续向外报告，便于定位程序问题。取消任务也继续向上传递，调用方才能停止整项工作。

分支怎样共享数据

示例中的分支各自返回结果，汇总节点统一组合。这种做法容易看出数据归属。若两个分支同时修改同一个字典的 answer 字段，最后写入的内容可能覆盖前一个结果。

更合适的结构是让订单分支返回 order，规则分支返回 policy，再由汇总节点生成 answer。如果多条分支要共同追加列表，就要约定合并规则：是否按输入顺序排列、是否去重、相同名称发生冲突时怎样处理。图框架里常用 reducer 表示这样的合并函数。

共享数据也有成本。把主流程全部聊天历史交给每个子流程，会让它们承担额外信息处理，并重复消耗上下文空间。子流程入口应明确需要哪些字段，返回哪些结果；订单查询只要订单标识和必要权限，通常用不到整段聊天记录。

哪些步骤适合同时执行

并发数量增加时，需要控制上限。例如每次最多查5份资料，其他任务等待名额。查询时间还需要超时上限，整项任务被取消后要释放资源。下一些文章会逐项展开这些控制。

对于已经会产生外部操作的分支，例如扣款和发货，要先明确它们之间的业务顺序、重复执行规则以及部分成功后的处理方式。这些关系写清楚后，再决定是否同时发起。

固定对照与测试

打开 concurrency_demo.py，把 main 中第二个样本的 True 改成 False，保存后运行。两组都得到完整资料，但打印标签仍保留原来的文字。恢复 True 后，第二组回到订单查询失败的情况。触发失败的是传入参数，标签用于展示。

保存测试文件 test_concurrency_demo.py。测试还安排了一个共同等待点：两个分支都开始后才释放，验证它们可以同时推进。

"""检查汇合是否完整、单个分支失败和子流程结果。"""
import asyncio
import unittest
from concurrency_demo import collect, guarded, order_subgraph


class ConcurrencyTests(unittest.IsolatedAsyncioTestCase):
    async def test_success(self):
        parts, answer = await collect()
        self.assertEqual([part.name for part in parts], ["订单", "规则"])
        self.assertEqual([part.status for part in parts], ["ok", "ok"])
        self.assertIn("金额80元", answer)

    async def test_partial_failure(self):
        parts, answer = await collect(True)
        self.assertEqual(parts[0].status, "failed")
        self.assertEqual(parts[1].status, "ok")
        self.assertEqual(answer, "规则已查到，订单信息待补查")

    async def test_subgraph(self):
        self.assertEqual(await order_subgraph(), "已付款，金额80元")

    async def test_concurrent_start(self):
        # 两个分支都到达等待点后才释放，确认它们能共同推进。
        started = 0
        release = asyncio.Event()

        async def branch():
            nonlocal started
            started += 1
            if started == 2:
                release.set()
            await release.wait()
            return "done"

        result = await asyncio.wait_for(
            asyncio.gather(branch(), branch()), timeout=1)
        self.assertEqual(result, ["done", "done"])

    async def test_unexpected_error_is_visible(self):
        async def broken():
            raise ValueError("字段错误")
        with self.assertRaises(ValueError):
            await guarded("订单", broken())

    async def test_cancellation_propagates(self):
        async def cancelled():
            raise asyncio.CancelledError()
        with self.assertRaises(asyncio.CancelledError):
            await guarded("订单", cancelled())


if __name__ == "__main__":
    unittest.main()

cd "F:\学习\l4-concurrency"
ls
python -X utf8 -m unittest -v test_concurrency_demo

应显示6项通过。IsolatedAsyncioTestCase 为每个异步测试准备事件循环，因此测试函数可以直接使用 await。

遇到“订单信息待补查”，先看订单那一项的 status 和 value；如果程序直接报告字段异常，就沿异常位置检查原始返回值。记录分支名称能把故障定位到某一项查询，而非只留一句“汇总失败”。

任务分支现在能共同推进了。下一篇检查点与持久状态解决另一件事：进程关闭后，怎样让这些进度留存并继续。异步任务接口可查阅 Python asyncio 文档。