京公网安备11010502063487号抓取速度比处理速度快的时候,没有上限的队列会把内存吃光。这篇用 asyncio.Queue 量一量背压:有界和无界队列在内存、总耗时上的差别,上限到底改变什么、不改变什么,task_done 的陷阱,以及多段流水线里积压出现在哪里。
队列 生产者发完用了 队列最长 峰值内存 全部处理完 无界(maxsize=0) 42 ms 2000 242 MiB 11226 ms maxsize=1000 5671 ms 1000 170 MiB 11407 ms maxsize=100 10754 ms 100 47 MiB 11324 ms maxsize=10 11188 ms 10 26 MiB 11250 ms maxsize=1 11251 ms 1 24 MiB 11262 ms (起点内存 24 MiB;2000 个 100 KB 的条目合计约 195 MiB)
消费者每个条目 5 ms、生产者不限速,无界队列让生产者 42 ms 就把 2000 个 100 KB 的内容全放进队列(峰值内存 242 MiB);maxsize=10 时生产者被迫等消费者,峰值内存 26 MiB,几乎没比空进程多多少。总处理时间在 11.2 到 11.4 秒之间,基本不受影响。
场景 maxsize 队列最长 全部处理完 消费者慢(5 ms)、生产者快 0 1000 5662 ms 消费者慢(5 ms)、生产者快 10 10 5635 ms 生产者慢(5 ms)、消费者快(1 ms) 0 1 5487 ms 生产者慢(5 ms)、消费者快(1 ms) 10 1 5449 ms
消费者是瓶颈时,有界队列只是把积压从队列里挪到“生产者在等”;生产者是瓶颈时,队列本来就积不起来。整体速度由慢的一端决定。
maxsize=2,已有 2 个,再 put_nowait:QueueFull await q.put(3) 一直等到被取消:TimeoutError(等了 301 ms) 消费者取完了 3 个条目但没调 task_done:q.join() 等了 501 ms 仍未返回(TimeoutError) 每取一个调一次 task_done:q.join() 在 0.1 ms 内返回
队列满了,put_nowait 抛 QueueFull,await put 则等待。用 join() 等所有条目处理完时,每取出一个条目都必须调一次 task_done(),否则 join() 永远不返回。
每段耗时(抓取/解析/存储) 总耗时 理论 = 100×最慢段 抓取→解析 队列平均长度 解析→存储 队列平均长度 2/2/2 ms 233 ms 200 ms 0.0 0.0 2/10/2 ms 1060 ms 1000 ms 4.5 0.3 2/2/10 ms 1062 ms 1000 ms 4.5 4.5 10/2/2 ms 1071 ms 1000 ms 0.5 0.0
抓取、解析、存储三段各接一个有界队列。最慢的那一段决定总时间(每段 100 个条目,总耗时接近 100 × 最慢段耗时),它前面的队列接近满,更上游的也会逐渐被堵住;最慢的是第一段时,后面的队列几乎是空的。观察队列长度,就能找到瓶颈在哪一段。