# 限制并发量：固定工作者与有界等待队列

学习目标：区分同时执行的任务数、等待队列长度和结果占用，让批量下载入口真正接受背压。

前置知识：循环、错误返回或异常、异步等待；先读并发等待。本篇选择 Go 的阻塞通道与 Python 的异步队列；两者都能仅用标准库表达固定工作者。下载器是本机延迟后返回内容的确定性模拟，不发公网请求，不创建下载文件，不声称测量了网络性能。

# 问题与边界

输入是有限整数编号序列和正整数工作者数 limit。非负编号下载成功，内容为编号的两倍；负编号模拟下载失败。成功结果按输入位置排列，空输入返回空结果。limit <= 0 在创建工作者前失败。

最多 limit 个下载同时执行，待处理队列容量固定为 1。队列满时生产者必须等待，而不是继续创建任务。个别下载失败后继续处理本批其余项，最后报告输入位置最靠前的错误，不返回部分结果；这是一种明确的“收齐后失败”策略，不是吞掉错误。取消中的批次另见超时与取消。

这里保留输入和结果，因此总内存仍为 O(n)；有界的是排队消息与工作者数量。真正的大文件下载还应逐块写入、限制响应体大小，并把结果元数据流式输出。

# 关键差异

信号量只限制进入临界区的任务数；如果仍为每个输入创建一个等待信号量的任务，任务对象数量并没有上限。本例从源头只创建固定数量的工作者。

# 各语言实现

# Go

有缓冲通道在满时阻塞发送，见 Go 通道规范；WaitGroup 只等待结束，不自动传递错误，所以结果消息包含 err。

Gopackage main

import (
    "errors"
    "fmt"
    "sync"
    "time"
)

type job struct { index, id int }
type result struct { index, value int; err error }

func download(id int) (int, error) {
    time.Sleep(time.Millisecond) // 本地模拟 I/O 等待，不是网络访问。
    if id < 0 { return 0, errors.New("download failed") }
    return id * 2, nil
}

func batch(ids []int, limit int) ([]int, error) {
    if limit <= 0 { return nil, errors.New("invalid limit") }
    jobs := make(chan job, 1)
    results := make(chan result, 1)
    var workers sync.WaitGroup
    workers.Add(limit)
    for i := 0; i < limit; i++ {
        go func() {
            defer workers.Done()
            for item := range jobs {
                value, err := download(item.id)
                results <- result{item.index, value, err}
            }
        }()
    }
    go func() {
        for index, id := range ids { jobs <- job{index, id} }
        close(jobs)
        workers.Wait()
        close(results)
    }()
    values := make([]int, len(ids))
    firstIndex := len(ids)
    var firstError error
    for item := range results {
        if item.err != nil {
            if item.index < firstIndex {
                firstIndex, firstError = item.index, item.err
            }
        } else {
            values[item.index] = item.value
        }
    }
    if firstError != nil { return nil, firstError }
    return values, nil
}

func main() {
    empty, err := batch(nil, 2)
    if err != nil || len(empty) != 0 { panic("empty batch") }
    values, err := batch([]int{0, 1, 2, 3}, 2)
    if err != nil || fmt.Sprint(values) != "[0 2 4 6]" { panic("bad values") }
    fmt.Println("0,2,4,6")
    for _, test := range []struct { ids []int; limit int; message string }{
        {[]int{0, -1, 2}, 2, "download failed"},
        {nil, 0, "invalid limit"},
    } {
        if _, err := batch(test.ids, test.limit); err == nil || err.Error() != test.message {
            panic("failure was lost")
        } else { fmt.Println(err) }
    }
}

结果通道也有界，主任务持续接收，避免工作者卡在结果发送、生产者又卡在输入发送的循环等待。只有生产者关闭输入，只有等待全部工作者结束的协调者关闭输出。预期下载错误通过消息传播；程序错误导致的 panic 不被伪装成普通下载失败。

# Python

asyncio.Queue 的 maxsize 限制排队项，满队列的 put 会等待；TaskGroup 对非预期异常会取消并等待兄弟任务。

Pythonimport asyncio

async def download(identifier):
    await asyncio.sleep(0)  # 本地模拟一次可挂起的 I/O 完成。
    if identifier < 0:
        raise ValueError("download failed")
    return identifier * 2

async def batch(ids, limit):
    if limit <= 0:
        raise ValueError("invalid limit")
    queue = asyncio.Queue(maxsize=1)
    values = [None] * len(ids)
    failures = {}
    active = 0
    peak = 0

    async def produce():
        for index, identifier in enumerate(ids):
            await queue.put((index, identifier))
        for _ in range(limit):
            await queue.put(None)

    async def consume():
        nonlocal active, peak
        while True:
            item = await queue.get()
            if item is None:
                return
            index, identifier = item
            active += 1
            peak = max(peak, active)
            try:
                values[index] = await download(identifier)
            except ValueError as error:
                failures[index] = error
            finally:
                active -= 1

    async with asyncio.TaskGroup() as group:
        group.create_task(produce())
        for _ in range(limit):
            group.create_task(consume())
    assert active == 0 and peak <= limit
    if failures:
        raise failures[min(failures)]
    return values

async def main():
    assert await batch([], 2) == []
    values = await batch([0, 1, 2, 3], 2)
    assert values == [0, 2, 4, 6]
    print(",".join(map(str, values)))
    for ids, limit, message in [
        ([0, -1, 2], 2, "download failed"),
        ([], 0, "invalid limit"),
    ]:
        try:
            await batch(ids, limit)
        except ValueError as error:
            assert str(error) == message
            print(error)
        else:
            raise AssertionError("failure was lost")

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

这里没有 queue.join()，完成条件是所有消费者收到哨兵并由 TaskGroup 等待结束，因此不需要 task_done() 计数。不要在加入 join() 后仍省略 task_done()。active 的更新没有跨越 await，只在同一事件循环线程执行；它不是可直接移植到多线程的共享计数器。

# 迁移误区与适用限制

• 无界队列加两个工作者：执行数有界，但上游仍能把全部任务压入内存，背压不成立。
• 工作者一遇到失败就直接退出：若生产者继续向满队列发送，会永久等待。本例正常失败记入结果并继续消费；非预期异常在 Python 由 TaskGroup 整体取消。
• 提前返回而不收结果：Go 工作者可能永远阻塞在结果发送。本例直到结果通道关闭才返回错误。
• limit 并非越大越好：数据库连接池、远端速率限制和每项响应体大小才决定可接受上限；本例只拒绝非正数，不替部署环境制定配额。

# 运行与验证

两个程序预期输出均为：

纯文本0,2,4,6
download failed
invalid limit

覆盖空输入、工作者数少于任务数、编号 0、批中失败和零并发限制。时延只用于模拟等待，不以运行时间判断成功，也不声称获得了固定加速比。语言版本见示例基线，实际执行情况见验证记录。

# 相关阅读

• 任务间传递数据：生产者与消费者通道：生产者退出、消费者失败与队列完成信号。
• 超时与取消：把背压等待也纳入取消协议。