第 22 章 / 共 57 章 · 约 7 分钟
并发题 2：限速执行器
题目
实现一个执行器：调用方 submit(task) 提交任务，执行器保证每秒最多启动 R 个任务，同时最多 K 个任务并发运行。典型场景是调用有配额的第三方 API（比如每秒 10 次、最多 5 个并发连接），以及批量抓取、批量发送通知。
这题把阻塞队列、信号量和一个新组件——令牌桶——组合在一起，很考设计取舍。
先澄清
| 问题 | 不同答案带来的设计差异 |
|---|---|
| 限的是启动速率、并发数，还是都限？ | 速率用令牌桶，并发用信号量 |
| 允许突发吗？ | 令牌桶的容量 B |
| 超速时阻塞提交者、排队，还是拒绝？ | submit 是否阻塞，队列是否有界 |
| 需要拿到结果吗？ | submit 返回 Future |
| 单进程还是多机？ | 多机需要集中计数，本题先做单进程 |
下面按“都限、允许小突发、队列有界、满了阻塞提交者、返回 Future”来设计。
限速算法对比
| 算法 | 思路 | 优点 | 缺点 |
|---|---|---|---|
| 固定窗口 | 每秒一个计数器 | 最简单 | 窗口边界附近可能瞬间通过约 2R 个 |
| 滑动窗口日志 | 记录每个请求的时间戳 | 精确 | 内存 O(R)，随限额线性增长 |
| 漏桶 | 以恒定速率流出 | 输出平滑 | 不允许突发 |
| 令牌桶 | 以速率 R 生成令牌，桶上限 B | 允许有限突发，实现简单 | 需要选择 B |
执行器场景选令牌桶：既限制长期平均速率，又允许短时间内用掉攒下的令牌。
整体结构
查看图表源码
flowchart LR
accTitle: 限速执行器的结构
submit["submit()"] --> q["有界任务队列"] --> disp["调度线程"]
disp -->|"1"| permit["拿并发许可（Semaphore K）"]
permit -->|"2"| token["拿速率令牌（TokenBucket R, B）"]
token -->|"3"| worker["交给工作线程立即执行"]
worker -->|"任务结束"| release["归还并发许可"]
为什么先拿并发许可，再拿令牌？ 这是这道题最容易被忽略的细节。如果先拿令牌、再等许可，令牌会在手里“放旧”：几个任务先后拿到令牌，然后都卡在等许可；某一刻许可同时空出来，它们就在同一瞬间启动，瞬时速率突破 R。反过来，先拿许可再拿令牌，拿到令牌立刻启动，启动时间严格受令牌桶约束。
这个顺序也有代价：拿到许可之后、等令牌的这段时间里，这个并发名额是空着的，没有任务在跑。速率是瓶颈时（每秒 R 个远小于 K 个并发能消化的量），空着的名额本来也用不上，这个代价可以接受；如果并发才是瓶颈，就要重新权衡。面试中说明这一点即可。
为什么还要单独的信号量，不能只靠线程池大小 K？ 标准线程池（Java 的 newFixedThreadPool、Python 的 ThreadPoolExecutor）内部有一个无界队列。调度线程拿到令牌后把任务丢进去，如果 K 个线程都在忙，任务会在池子里排队，之后才真正启动——令牌在任务实际开始前就被消耗了，速率控制和实际启动时间脱节。信号量保证“交给线程池时一定有空闲线程”。
令牌桶：不需要后台线程
常见误区是起一个后台线程，每 1/R 秒往桶里加一个令牌。更好的做法是惰性补充：每次取令牌时，按距离上次的时间差一次性补上。
import threading, time
class TokenBucket:
def __init__(self, rate, burst, clock=time.monotonic):
self.rate, self.burst, self.clock = rate, burst, clock
self.tokens, self.last = float(burst), clock()
self.lock = threading.Lock()
def acquire(self):
with self.lock:
now = self.clock()
self.tokens = min(self.burst, self.tokens + (now - self.last) * self.rate)
self.last = now
self.tokens -= 1 # 可以变负：预约未来的令牌
wait = -self.tokens / self.rate if self.tokens < 0 else 0
if wait > 0:
time.sleep(wait) # 在锁外睡
两个关键点：
- 允许令牌为负，表示预约了未来的令牌，并据此算出要等多久。多个调用方会自动排好队：后来的人看到的欠额更大，等得更久。代价是：调用方预约之后如果被取消（线程被中断、执行器关闭），已经扣掉的额度不会自动还回来，这段配额就浪费了。在意的话，可以在取消时把令牌加回去，只是已经在睡的调用方不会因此提前醒来。
- 在锁外 sleep。如果在锁里睡，其他调用方全被堵在锁上，也没法及时响应关闭。
执行器主体
import queue
from concurrent.futures import Future, ThreadPoolExecutor
class RateLimitedExecutor:
def __init__(self, rate, burst, max_concurrency, queue_size):
self.bucket = TokenBucket(rate, burst)
self.slots = threading.Semaphore(max_concurrency)
self.pool = ThreadPoolExecutor(max_concurrency)
self.tasks = queue.Queue(queue_size) # 有界：满了 put 会阻塞，形成背压
threading.Thread(target=self._dispatch, daemon=True).start()
def submit(self, fn):
fut = Future() # 示意；Java 中对应 CompletableFuture
self.tasks.put((fn, fut))
return fut
def _dispatch(self):
while True:
item = self.tasks.get()
if item is None: # 关闭哨兵
break
fn, fut = item
self.slots.acquire() # 1. 并发许可
self.bucket.acquire() # 2. 速率令牌
self.pool.submit(self._run, fn, fut) # 3. 立即启动
def _run(self, fn, fut):
try:
fut.set_result(fn())
except Exception as e:
fut.set_exception(e) # 任务异常不能拖垮执行器
finally:
self.slots.release()
值得主动讨论的点
背压：任务队列有界，满了让 submit 阻塞，上游自然慢下来。也可以改成满了立即拒绝，取决于调用方能否承受阻塞。
关闭语义：
shutdown()：不再接收新任务，往队列放一个哨兵；调度线程处理完前面的任务后退出，再关闭线程池。shutdown_now()：清空队列，把还没启动的任务的 Future 标记为取消。
异常隔离：任务抛出的异常必须被捕获并放进 Future，finally 里归还许可。否则一个出错的任务会永久占掉一个并发名额，K 次之后执行器彻底停摆。
时钟：令牌桶用单调时钟 time.monotonic()，不要用会被 NTP 调整的 time.time()；时钟可注入，测试时用假时钟验证“突发 B 个之后，第 B+1 个需要等 1/R 秒”。
单个调度线程会成为瓶颈吗？ 它只做“拿许可、拿令牌、投递”，非常轻；而且速率本来就被限制在 R，通常不是瓶颈。
多租户：每个租户一个令牌桶，调度线程轮询各租户的队列，防止一个大租户饿死其他人。
分布式：多台机器共享配额时，令牌桶的状态要放到集中存储（比如 Redis），用原子脚本完成“补充 + 扣减”；或者把总配额按机器数静态切分，简单但不够灵活。
常见失分点
- 用后台线程定时补令牌，还在锁里 sleep。
- 只靠线程池大小控制并发，忽略了池内部的无界队列。
- 先拿令牌再等并发许可，导致瞬时速率超限。
- 任务异常时没有归还许可。
- 完全没有关闭逻辑，测试进程退不出去。