10.1 并发与队列：实例池与速率协调

10.1 并发与队列：实例池与速率协调

本节摘要：本节把第 7 章的会话基建放大成并发系统。先立观点：并发是基建问题不是循环问题——每个并发单元里跑的还是第 2.1 节那个循环，要新建的是循环外面的三件东西：任务队列（提交与消费解耦，优先级字段把主流程任务与异常工单分层编排）、浏览器实例池（池上限保护资源、预热消除冷启动、回收防泄漏；context 是任务的分配单元也是失败隔离的边界——一任务一 context，毒丸任务隔离不扩散）、速率闸（令牌桶限速，全局与按域名两档——全局档护自己基础设施，域名档兑现第 8 章的速率礼仪）。失败隔离与第 6.2 节回退链衔接：单任务救援有预算（局部），池有熔断（全局），两层保险。人机协同的形态在这里落成「异常工单队列」：L2 转出的任务带证据进队列，人处理完结论回写。双语言实现一个最小但结构完整的「队列加池加速度闸」骨架。

学习目标

• 说出并发三件基建（队列、实例池、速率闸）各自的职责。
• 实现带失败隔离与优雅回收的池化执行（双语）。
• 写出令牌桶限速器并解释两档（全局、按域名）的动机。
• 区分单任务救援预算与全局熔断两层保险。
• 设计异常工单队列的人机协同闭环。

一、并发是基建问题

把第 2 章的 Agent 循环直接跑一百份，不是并发系统，是事故预案。每个循环自己起浏览器、自己抢网络、自己失败自己重试，会发生三件事：资源被先到者占光（没有秩序）、一个任务的失控拖垮全体（没有隔离）、对目标站点的请求在毫秒级叠加（没有礼仪）。所以并发的正确起点是承认：循环不需要改，循环外面需要一套基建：

 提交 ──▶ 队列（优先级、积压可见） ──▶ 调度 ──▶ 实例池（上限、预热、回收）
                                          │
                                          ▼
                                    浏览器 context（一任务一个）
                                          │
                                    速率闸（全局 + 按域名）
                                          │
                                          ▼
                                       目标站点
​

三个部件的分工：队列负责秩序（谁先跑、积压多少看得见），池负责资源（同时多少个、坏了怎么换），速率闸负责节奏（多快算合适）。第 6.2 节的回退链住在 context 里面——并发系统不改变单任务的救援逻辑，只给它加全局约束（救援预算总占比）。

二、实例池：参数与生命周期

生命周期的关键动作是回收要干净：实例退出前确认浏览器进程结束、临时目录清理、会话状态按第 7.1 节策略销毁——泄漏的浏览器进程会在几小时后把「莫名变慢」变成停摆。分配单元是 context 而不是浏览器：一个实例上顺序跑多个任务、每个任务一个全新 context，既省启动成本又保住隔离边界。

Python 骨架（队列加池，asyncio 原生）：

# pool_queue.py —— 队列加实例池的最小骨架（写法示意，以官方文档为准）
import asyncio

POOL_SIZE = 3           # 池上限（示意）
IDLE_TIMEOUT_S = 30


async def worker(name: str, queue: asyncio.Queue) -> None:
    while True:
        task = await queue.get()
        try:
            print(f"[{name}] 领取: {task}")
            await asyncio.sleep(0.2)          # 任务执行（示意，含第 6.2 节回退链）
            print(f"[{name}] 完成: {task}")
        except Exception as exc:              # 失败隔离：异常不出循环
            print(f"[{name}] 隔离失败任务: {exc}")
        finally:
            queue.task_done()                 # 无论成败都归还名额


async def main() -> None:
    queue: asyncio.Queue = asyncio.Queue()
    for i in range(10):
        await queue.put(f"task-{i}")
    workers = [asyncio.create_task(worker(f"pool-{i}", queue)) for i in range(POOL_SIZE)]
    await queue.join()
    for w in workers:
        w.cancel()


asyncio.run(main())
​

JS 骨架（池上限加独立 context）：

// pool_queue.mjs —— 池上限与 context 级隔离（写法示意，以官方文档为准）
import { chromium } from "playwright";

const POOL_SIZE = 3; // 池上限（示意）

/**
 * 按池上限调度任务，每个任务一个独立 context
 * @param {string[]} names 任务名列表 @returns {Promise<string[]>} 结果清单
 */
async function runWithPool(names) {
  const browser = await chromium.launch();
  const results = [];
  const executing = new Set();
  for (const name of names) {
    const job = (async () => {
      const context = await browser.newContext(); // 一任务一 context：隔离边界
      const page = await context.newPage();
      try {
        await page.goto("http://127.0.0.1:8080/demo_todo.html");
        results.push(`${name}: ${await page.title()}`);
      } finally {
        await context.close(); // 用完即回收，泄漏防线
      }
    })();
    executing.add(job);
    job.finally(() => executing.delete(job));
    if (executing.size >= POOL_SIZE) {
      await Promise.race(executing); // 等最早完成的腾出名额
    }
  }
  await Promise.all(executing);
  await browser.close();
  return results;
}

console.log(await runWithPool(["a", "b", "c", "d", "e"]));
​

两个骨架共同的要点：异常被锁死在任务边界内（worker 的 try 与 job 的 finally），一个任务毒死不了池子；份额的归还发生在 finally——成败都归还，这是池不枯竭的保证。

三、速率闸：令牌桶与两档限速

并发的节奏控制用令牌桶：桶以固定速率补充令牌，动作先取令牌再执行，取不到就等。

# rate_gate.py —— 令牌桶限速器（写法示意）
import asyncio
import time


class TokenBucket:
    def __init__(self, rate: float, capacity: int) -> None:
        self.rate = rate            # 每秒补充的令牌数
        self.capacity = capacity
        self.tokens = float(capacity)
        self.updated = time.monotonic()

    async def take(self) -> None:
        while True:
            now = time.monotonic()
            self.tokens = min(self.capacity, self.tokens + (now - self.updated) * self.rate)
            self.updated = now
            if self.tokens >= 1:
                self.tokens -= 1
                return
            await asyncio.sleep((1 - self.tokens) / self.rate)


async def main() -> None:
    global_gate = TokenBucket(rate=5, capacity=10)     # 全局档：护自己的基础设施
    domain_gate = TokenBucket(rate=1, capacity=3)      # 域名档：对目标站点的礼仪
    for i in range(6):
        await global_gate.take()
        await domain_gate.take()
        print(f"放行请求 {i}")


asyncio.run(main())
​

两档的动机不同：全局档保护自己的出口带宽、模型接口配额、DNS 与连接池——超载的受害者是自己；域名档兑现第 8 章速率礼仪——对每个目标域名的请求频率控制在该方可接受的范围内（具体数值来自服务条款与站点方的说明，写进配置而非代码）。域名档要按域名各配一只桶，全局一只总桶，请求先过全局再过域名。

四、失败隔离与人机协同

失败隔离的三层结构，从内到外：

熔断的阈值与退避（暂停多久、半开试探）接第 11.2 节告警分级。毒丸任务的处置：连续让多个 context 异常退出的任务（数据异常或指令病态）应被隔离进 quarantine 队列而非反复重派——池的吞吐是公共财产，不该为一个毒丸无限买单。

人机协同落在异常工单队列：第 6.2 节 L2 转出的任务（带第 8.3 节证据三件套）进低优先级人工队列，处理结论回写知识库（「该站点改版，入口换了」「该账号需要重新授权」），有价值的结论转成新规则或新固化候选。这个闭环让并发系统的「失败」不是损耗而是养料——第 11.2 节的错误分类统计，原料正来自这里。

五、常见坑

本节要点回顾

• 并发三件基建：队列管秩序（优先级、积压可见）、池管资源（上限、预热、干净回收）、速率闸管节奏（全局加按域名两档）。
• 分配与隔离单元是 context：一任务一 context，成败都归还份额；毒丸进 quarantine。
• 令牌桶两档动机：全局档护自己，域名档兑现第 8 章礼仪；数值进配置不进代码。
• 失败隔离三层：任务内回退链、池内 context 销毁、全局熔断接告警。
• 异常工单闭环：证据进队列、结论回写知识库——失败变养料。

秩序立好了，账单来了：几十个并发任务跑一周，token 费用会自己找到你。下一章最后一站之前，先把成本这件事解剖清楚——感知、决策、重试三个成分各占什么量级，省钱三招各自砍哪里。