This post is also available in English.
在高并发场景里，"发请求"这件事没有看起来那么简单。
你可能写了一个爬虫，或者要批量调用第三方 API。一开始跑得很顺畅，但请求量一上去，问题就来了：接口开始 429，事件循环卡住了，或者线程池直接把对面服务打挂。
这篇文章从最基础的信号量讲起，一步步走到 PID 控制器、令牌桶、自适应并发调整。每种方案都有完整的 Python 代码，拿来就能跑。
先看全局：并发控制的核心思路
不管你用的是 asyncio 还是 threading，并发控制的核心问题都是一样的：怎么在「尽可能快」和「不把系统打崩」之间找到平衡。
下面这张图展示了一个通用的并发控制流程：
graph LR
A[请求队列] --> B{并发控制器}
B -->|获取令牌/信号量| C[Worker 池]
C --> D[目标服务]
D -->|响应| E{反馈收集}
E -->|成功率/延迟| B
E --> F[结果聚合]
style A fill:#EFF6FF,stroke:#2563EB,stroke-width:2px,color:#1E3A5F
style B fill:#FFFBEB,stroke:#D97706,stroke-width:2px,color:#92400E
style C fill:#ECFDF5,stroke:#059669,stroke-width:2px,color:#065F46
style D fill:#FEF2F2,stroke:#DC2626,stroke-width:2px,color:#991B1B
style E fill:#FFFBEB,stroke:#D97706,stroke-width:2px,color:#92400E
style F fill:#EFF6FF,stroke:#2563EB,stroke-width:2px,color:#1E3A5F关键在那个反馈回路：把请求的成功率、响应时间作为信号，反过来调整并发量。这就是从「固定并发」到「动态并发」的核心区别。
1. 信号量：最简单的并发限制
信号量（Semaphore）是并发控制的起点。思路很直接：设一个上限，同时只允许 N 个任务运行，多的排队等着。
异步信号量
在 asyncio 里，信号量用 asyncio.Semaphore：
import asyncio
import httpx
async def fetch_url(semaphore: asyncio.Semaphore, client: httpx.AsyncClient, url: str):
async with semaphore: # 进入时 -1，退出时 +1
response = await client.get(url)
return response.status_code
async def main():
urls = [f"https://httpbin.org/delay/1" for _ in range(50)]
semaphore = asyncio.Semaphore(10) # 最多 10 个请求同时飞
async with httpx.AsyncClient(timeout=30) as client:
tasks = [fetch_url(semaphore, client, url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
success = sum(1 for r in results if isinstance(r, int) and r == 200)
print(f"完成: {success}/{len(urls)} 成功")
asyncio.run(main())
这段代码的关键：async with semaphore 是一个上下文管理器，进入时信号量计数减 1，退出时加 1。当计数为 0 时，新的 async with 会挂起（不是阻塞线程），等到有任务完成释放信号量后才继续。
线程信号量
在多线程里，原理一样，API 略有不同：
import concurrent.futures
import threading
import time
import random
def fetch_with_semaphore(semaphore: threading.Semaphore, task_id: int):
with semaphore:
print(f"[Task {task_id}] 开始执行")
time.sleep(random.uniform(0.1, 0.5)) # 模拟网络请求
print(f"[Task {task_id}] 完成")
return task_id
def main():
semaphore = threading.Semaphore(5) # 最多 5 个线程同时执行
with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
futures = [
executor.submit(fetch_with_semaphore, semaphore, i)
for i in range(30)
]
results = [f.result() for f in concurrent.futures.as_completed(futures)]
print(f"完成 {len(results)} 个任务")
main()
信号量的局限
信号量好用，但有一个明显的问题：并发上限是写死的。
你设 10，那就永远是 10。但现实中：
- 凌晨 3 点服务器空闲，10 太保守
- 下午 3 点流量高峰，10 可能太激进
- 目标服务刚好在发布，承载能力临时下降
怎么办？得让并发量「活」起来。
2. PID 控制器：用反馈回路动态调参
PID 控制器（比例-积分-微分控制器）是工业控制里的经典算法。它的思路是：观察系统的实际状态和目标状态之间的差距（误差），然后根据误差来调整输出。
用在并发控制里：
- 目标：把失败率控制在某个阈值以下（比如 5%）
- 输出：并发量
- 反馈：实际的请求失败率
PID 三个分量
| 分量 | 作用 | 并发控制中的含义 |
|---|---|---|
| P（比例） | 根据当前误差调整 | 失败率高了，立刻降并发 |
| I（积分） | 累计历史误差，消除稳态偏差 | 失败率一直略高，持续微调 |
| D（微分） | 根据误差变化速度调整 | 失败率突然飙升，快速响应 |
完整实现
import asyncio
import httpx
import time
import random
class PIDController:
"""PID 控制器，用于动态调整并发量"""
def __init__(self, kp: float, ki: float, kd: float,
target: float, min_output: int, max_output: int):
self.kp = kp # 比例系数
self.ki = ki # 积分系数
self.kd = kd # 微分系数
self.target = target # 目标失败率
self.min_output = min_output
self.max_output = max_output
self._integral = 0.0
self._prev_error = 0.0
def compute(self, current_value: float) -> int:
"""根据当前失败率计算新的并发量"""
error = self.target - current_value # 正值 = 还有余量可以加速
# P: 当前误差
p_term = self.kp * error
# I: 累积误差（加 anti-windup 限制）
self._integral += error
self._integral = max(-50, min(50, self._integral)) # 防止积分饱和
i_term = self.ki * self._integral
# D: 误差变化率
d_term = self.kd * (error - self._prev_error)
self._prev_error = error
output = p_term + i_term + d_term
return int(max(self.min_output, min(self.max_output, output)))
async def fetch_with_pid(url: str, client: httpx.AsyncClient, semaphore: asyncio.Semaphore):
"""带信号量的异步请求"""
async with semaphore:
try:
resp = await client.get(url, timeout=10)
# 模拟：随机失败来演示 PID 调整
if random.random() < 0.15:
return False
return resp.status_code == 200
except Exception:
return False
async def main():
pid = PIDController(
kp=20.0, ki=0.5, kd=5.0,
target=0.05, # 目标失败率 5%
min_output=2, # 最少 2 并发
max_output=50 # 最多 50 并发
)
current_concurrency = 10
url = "https://httpbin.org/delay/0"
async with httpx.AsyncClient() as client:
for batch in range(10):
semaphore = asyncio.Semaphore(current_concurrency)
tasks = [fetch_with_pid(url, client, semaphore) for _ in range(20)]
results = await asyncio.gather(*tasks)
failure_rate = sum(1 for r in results if not r) / len(results)
new_concurrency = pid.compute(failure_rate)
print(f"Batch {batch+1}: "
f"并发={current_concurrency}, "
f"失败率={failure_rate:.1%}, "
f"调整后并发={new_concurrency}")
current_concurrency = new_concurrency
asyncio.run(main())
PID 调参建议
调 PID 参数是一门手艺。几个经验值：
- 先只用 P：把
ki和kd设为 0，调kp让系统大致能跟上 - 加 I 消除稳态误差：如果发现失败率稳定在目标值附近但就是到不了，加一点
ki - 加 D 抑制震荡：如果并发量来回剧烈波动，加
kd - 一定要加 anti-windup：限制积分项的累积范围，防止积分饱和后系统失控
3. 令牌桶：控制请求速率
信号量控制的是「同时有多少请求在飞」，令牌桶控制的是「每秒能发多少请求」。这是两个不同的维度。
令牌桶 vs 信号量
举个例子：信号量设 10，意味着最多 10 个请求同时进行。但如果每个请求只要 10ms 就返回，那一秒钟你就发了 1000 个请求。信号量管不了这个。
令牌桶的思路：
- 桶里有令牌，以固定速率生成（比如每秒 100 个）
- 每个请求消耗一个令牌
- 桶满了就不再生成（有上限）
- 没有令牌就等着
线程安全的令牌桶实现
import threading
import time
class TokenBucket:
"""线程安全的令牌桶"""
def __init__(self, rate: float, capacity: int):
self.rate = rate # 令牌生成速率（个/秒）
self.capacity = capacity # 桶的最大容量
self._tokens = float(capacity)
self._last_refill = time.monotonic()
self._lock = threading.Lock()
def _refill(self):
"""按时间补充令牌"""
now = time.monotonic()
elapsed = now - self._last_refill
new_tokens = elapsed * self.rate
self._tokens = min(self.capacity, self._tokens + new_tokens)
self._last_refill = now
def acquire(self, timeout: float = None) -> bool:
"""获取一个令牌。成功返回 True，超时返回 False"""
deadline = time.monotonic() + timeout if timeout else None
while True:
with self._lock:
self._refill()
if self._tokens >= 1:
self._tokens -= 1
return True
# 没有令牌，等一下再试
if deadline and time.monotonic() >= deadline:
return False
time.sleep(0.01) # 10ms 间隔重试
class RateLimitedExecutor:
"""带速率限制的任务执行器"""
def __init__(self, rate: float, capacity: int, max_workers: int):
self.bucket = TokenBucket(rate, capacity)
self.max_workers = max_workers
def execute(self, task_fn, task_args_list):
"""执行一批任务"""
import concurrent.futures
def wrapped(args):
# 先拿令牌，再执行
if self.bucket.acquire(timeout=30):
return task_fn(*args)
else:
raise TimeoutError("获取令牌超时")
with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as pool:
futures = [pool.submit(wrapped, args) for args in task_args_list]
return [f.result() for f in concurrent.futures.as_completed(futures)]
# 使用示例
def call_api(task_id: int):
print(f"[{time.strftime('%H:%M:%S')}] Task {task_id} 执行中")
time.sleep(0.1)
return f"Task {task_id} 完成"
executor = RateLimitedExecutor(
rate=10, # 每秒 10 个请求
capacity=20, # 最多攒 20 个令牌（允许短暂突发）
max_workers=5
)
results = executor.execute(call_api, [(i,) for i in range(30)])
print(f"\n完成 {len(results)} 个任务")
asyncio 版令牌桶
如果你用的是异步模型，令牌桶可以用 asyncio.Event 实现，避免 busy-wait：
import asyncio
import time
class AsyncTokenBucket:
"""异步令牌桶"""
def __init__(self, rate: float, capacity: int):
self.rate = rate
self.capacity = capacity
self._tokens = float(capacity)
self._last_refill = time.monotonic()
self._event = asyncio.Event()
self._event.set()
async def acquire(self):
"""获取一个令牌（协程会挂起，不阻塞线程）"""
while True:
now = time.monotonic()
elapsed = now - self._last_refill
self._tokens = min(self.capacity, self._tokens + elapsed * self.rate)
self._last_refill = now
if self._tokens >= 1:
self._tokens -= 1
return True
# 计算下一个令牌生成的时间
wait_time = (1 - self._tokens) / self.rate
await asyncio.sleep(wait_time)
4. 自适应并发调整：从 1 开始探测上限
前面几种方案都需要你提前知道一些参数（并发上限、目标失败率、令牌速率）。但如果你完全不知道目标服务能承受多少呢？
自适应并发调整的思路很简单：从低并发开始，逐步加速，直到开始出错，然后退回来。类似于 TCP 的拥塞控制。
核心逻辑
并发量 = 1
每个批次结束后：
如果 失败率 < 阈值 → 并发量 += 步长（加速探测）
如果 失败率 > 阈值 → 并发量 = 并发量 / 2（快速退避）
并发量 = clamp(并发量, 最小值, 最大值)
完整实现
import asyncio
import httpx
import random
class AdaptiveConcurrencyController:
"""自适应并发控制器 — 自动探测最佳并发量"""
def __init__(self, min_concurrency: int = 1, max_concurrency: int = 100,
failure_threshold: float = 0.1, step: int = 2):
self.min_concurrency = min_concurrency
self.max_concurrency = max_concurrency
self.failure_threshold = failure_threshold # 允许的最大失败率
self.step = step # 每次增加的并发量
self.current_concurrency = min_concurrency
self._history = [] # 记录每轮的 (并发量, 失败率)
def adjust(self, failure_rate: float):
"""根据失败率调整并发量"""
old = self.current_concurrency
if failure_rate <= self.failure_threshold:
# 还有余量，加速
self.current_concurrency = min(
self.max_concurrency,
self.current_concurrency + self.step
)
else:
# 超过阈值，快速退避（减半）
self.current_concurrency = max(
self.min_concurrency,
self.current_concurrency // 2
)
self._history.append((old, failure_rate, self.current_concurrency))
return self.current_concurrency
def report(self):
"""打印调整历史"""
print("\n--- 并发调整历史 ---")
print(f"{'轮次':>4} | {'并发量':>6} | {'失败率':>6} | {'调整后':>6}")
print("-" * 40)
for i, (old, rate, new) in enumerate(self._history, 1):
arrow = "↑" if new > old else ("↓" if new < old else "→")
print(f"{i:>4} | {old:>6} | {rate:>5.1%} | {new:>5} {arrow}")
async def fetch(client: httpx.AsyncClient, semaphore: asyncio.Semaphore, url: str):
async with semaphore:
try:
resp = await client.get(url, timeout=10)
# 模拟：并发越高，失败率越高
simulated_failure = random.random() < (semaphore._value / 100)
return not simulated_failure and resp.status_code == 200
except Exception:
return False
async def main():
controller = AdaptiveConcurrencyController(
min_concurrency=1,
max_concurrency=50,
failure_threshold=0.15, # 15% 失败率阈值
step=3
)
url = "https://httpbin.org/delay/0"
batch_size = 20
async with httpx.AsyncClient() as client:
for round_num in range(15):
concurrency = controller.current_concurrency
semaphore = asyncio.Semaphore(concurrency)
tasks = [fetch(client, semaphore, url) for _ in range(batch_size)]
results = await asyncio.gather(*tasks)
failure_rate = sum(1 for r in results if not r) / len(results)
controller.adjust(failure_rate)
controller.report()
asyncio.run(main())
运行后你会看到并发量先逐步上升，在失败率超过阈值后快速回退，最终在某个范围内稳定——这就是自适应的效果。
5. 实战选型：哪种方案适合你？
| 场景 | 推荐方案 | 理由 |
|---|---|---|
| 对接有明确 rate limit 的 API | 令牌桶 | 直接按 API 限制设速率 |
| 并发量大致已知，追求简单 | 固定信号量 | 三行代码搞定 |
| 目标服务负载不可预测 | 自适应并发 | 自动探测，不用猜参数 |
| 需要精确控制失败率 | PID 控制器 | 反馈回路最精细 |
| 需要同时控制并发和速率 | 信号量 + 令牌桶 | 两者组合使用 |
几个实战建议
- 先用固定信号量。大多数情况下，设一个保守的并发上限就够了。不要过度工程化。
- 监控先行。不管用哪种方案，先把成功率、响应时间、并发量这些指标记录下来。有数据才能调参。
- 令牌桶和信号量可以叠加。信号量控制同时在飞的请求数，令牌桶控制每秒的请求速率，两者不矛盾。
- 重试要带退避。并发控制解决的是「发多快」的问题，重试策略解决的是「失败了怎么办」的问题。两者要配合，记得用指数退避（exponential backoff）。
- 测试环境和生产环境差异很大。本地测试跑得好的参数，上了生产可能完全不同。留好动态调整的能力。
最后
并发控制说到底就一个问题：怎么在「跑得快」和「别把对面打崩」之间找平衡。
信号量是最简单的答案，能覆盖大多数场景；要控速率就上令牌桶；想精细控制失败率，PID 的反馈回路最细；完全不知道对面底细，就让自适应自己去探上限。
我的建议还是老一套：从简单的开始，遇到问题再升级。别一上来就堆最复杂的方案，那本身就跟「控制复杂度」拧着来了。