背压把消费者的有限容量与生产者行为联系起来。通道已满时，生产者可能等待、拒绝新任务，或执行经过明确选择的丢弃策略。关键是容量限制真正改变上游怎么做。队列不断增长，或仪表盘显示积压，只能说明供需失衡，本身不自动形成控制接入的机制。需要从观察到过载，继续追到生产者是否确实减速或停止提交。

坏掉的样子

无界队列可以吸收短暂突发，却也会隐藏持续过载。如果到达速度长期超过完成速度，未完成任务和等待时间就持续增长。把积压挪到待处理线程、重试任务或另一个无界缓冲，并没有消除压力。容量边界必须覆盖任务真正被保存的位置，不能只限制一个容易观察的队列对象，却允许周围继续无限保留工作。

有界队列本身也没有定义完整过载策略。无限阻塞可能耗尽生产者线程或让关闭流程卡住；立即拒绝时，调用方若误认为成功，就可能丢失工作；每次拒绝后立即重试，又可能加重负载。应依据应用契约明确选择：在截止时间内等待、拒绝并稍后重试，还是有意丢弃。不同选择改变交付语义，不能在实现中含糊处理。

队列容量与处理完成是两件事。消费者取出项目可以释放槽位，但此时可能仍在处理这个项目。等待容量为二，也可能同时还有额外执行中的工作。生产者成功入队，仅说明队列接受了项目，不证明已经执行、持久保存或完成下游副作用。若把所有这些状态都叫作“成功”，过载和失败时就无法判断责任归属。

快速验证

将脚本保存为 check-backpressure.py，运行 python3 check-backpressure.py。它只使用 Python 标准库，在 Python 3.14.4 上执行。线程通过明确事件和有界等待协调，不依赖远端消息代理、数据库或刻意拖慢的外部服务。所有队列和任务都只存在当前进程中，结果用于观察接入与完成的语义，不代表分布式队列的持久化保证：

from queue import Queue,Full
from threading import Event,Thread
import json

unbounded=Queue()
for item in range(1000):unbounded.put_nowait(item)
assert unbounded.qsize()==1000
channel=Queue(maxsize=2)
channel.put('a');channel.put('b')
try:
    channel.put('deadline-rejected',timeout=.02)
    raise AssertionError('a full bounded queue accepted another item')
except Full:rejected=True
assert channel.qsize()==2
entered,delivered=Event(),Event()
def produce():
    entered.set();channel.put('c',timeout=2);delivered.set()
producer=Thread(target=produce,daemon=True);producer.start();assert entered.wait(1)
assert not delivered.wait(.05) and channel.qsize()==2
first=channel.get(timeout=1);assert first=='a'
assert delivered.wait(1);producer.join(timeout=1);assert not producer.is_alive()
assert channel.qsize()==2
rest=[channel.get(timeout=1),channel.get(timeout=1)];assert rest==['b','c']
assert channel.empty()
completed=Event()
def join():channel.join();completed.set()
waiter=Thread(target=join,daemon=True);waiter.start();assert not completed.wait(.05)
for item in [first,*rest]:channel.task_done()
assert completed.wait(1);waiter.join(timeout=1);assert not waiter.is_alive()
print(json.dumps({'unbounded_backlog':unbounded.qsize(),'bounded_capacity':2,'deadline_rejected':rejected,
    'blocked_producer_resumed_after_get':delivered.is_set(),'delivered_items':[first,*rest],
    'empty_queue_did_not_mean_processing_complete':True,'join_finished_after_task_done':completed.is_set()}))

第一个队列使用默认无界容量。没有消费者运行时，一千次 put_nowait 全部成功，留下同样数量的积压。这是小规模行为演示，不是耗尽内存测试。它说明接入成功本身无法证明下游已有进展，也不能证明到达速度可持续。监控若只统计提交成功率，就可能在处理能力不足时仍显示表面正常。

第二个队列容量为二，先放入 a 和 b。再次 put 使用短超时，随后抛出 Full，已有项目保持不变。失败项目没有被暗中排队等待以后执行。调用方因此获得明确接入结果，可以按契约决定重试或拒绝，而不是把无限等待包装成已经交付。脚本核对队列状态，避免仅凭异常出现就忽略是否发生了部分接收。

随后生产者线程尝试放入 c。两个槽位占满时，测试确认它仍被阻塞。取出 a 后空出一个位置，生产者完成 put，队列留下 b 和 c。程序验证三个已接受项目最终按顺序被取出。这演示压力确实传递到了生产者，又被具体容量变化释放，而不是仅记录一个队列长度，却让生产者继续无节制提交。

取出其余项目后，队列已经为空，但另一个等待 join 的线程仍未完成。只有对三个已接受项目分别调用 task_done，join 才返回。取出动作先释放容量，任务确认之后才表示被跟踪工作完成。测试刻意分开这些事件，避免把队列清空误认为业务处理结束。生产系统也需要对“拿到任务”和“完成任务”保留对应区分。

让接入契约贯穿整个处理链

应使用队列操作的真实结果，不要根据 qsize 或 full 先检查再行动。其他线程可以随时改变队列，这些观察在并发环境下只具有近似性质。本例通过事件和暂停消费者，使断言对应受控观察；生产调用方应通过操作本身处理 Full 或成功 put，避免在检查与提交之间发生竞争，或将过期容量判断当成保证。

容量应对应真正保留的资源。项目大小不受限时，按个数限制不能直接封顶内存；等待队列限制也不包括所有活跃工作线程或外部请求。评估过载时，需要一起考虑处理并发和上游重试。一个有界组件周围如果仍存在无限待处理工作，整个系统依然可能无界，不能因局部出现 maxsize 就宣布过载闭环已经完成。

最后要定义完成的含义。Python task_done 更新队列记账，不是持久事务，也不证明外部副作用成功。工作线程失败、进程崩溃，或响应不明确之后的重试，都需要应用对应的交付与幂等契约。背压控制接入和节奏，正确性仍依赖明确处理已接受、被拒绝、执行中与已完成任务。两类职责配合，才能在压力变化时保持可解释行为。

延伸阅读

来源