asyncio.semaphore 初始化后不能直接修改 value，因其内部计数器为只读属性，强行修改不触发等待队列重排；应使用 dynamicsemaphore 封装 condition 实现动态限流，并注意取消任务时的泄漏问题。
asyncio.Semaphore 初始化后不能直接修改 value
很多人试过直接改 semaphore._value 或给 semaphore.value 赋值，发现没用——因为 asyncio.Semaphore 的内部计数器是只读属性，且所有 acquire/release 操作都基于当前值做原子判断，强行改底层字段不会触发等待队列重排，已阻塞的协程也不会被唤醒。
真正可行的方式是「替换信号量实例」，但必须确保：所有任务都在使用同一个引用，且切换时无竞态。常见错误是多个地方各自创建 asyncio.Semaphore(5)，后续换一个新实例，老任务仍卡在旧实例上。
- 统一管理信号量实例，通过全局变量或依赖注入传递（不推荐全局变量，优先用类属性或 contextvar）
- 切换前需确保没有协程正在执行
acquire()中间态（即不在await semaphore.acquire()的挂起点上） - 若已有大量任务在等待，替换后它们会继续等新信号量，所以需配合取消逻辑或超时控制
用 asyncio.Condition 实现安全的动态并发调节
想让运行中的任务感知并发上限变化，得靠通知机制。asyncio.Condition 是最轻量的选择：它能阻塞等待，并在并发数调整后唤醒部分协程去重试 acquire。
核心思路是封装一层 DynamicSemaphore，内部维护当前 value 和一个 Condition，每次 acquire() 先检查是否满足条件，否则等待通知；set_limit(n) 则更新值并 notify_all。
class DynamicSemaphore:
def __init__(self, initial_value: int = 1):
self._value = initial_value
self._condition = asyncio.Condition()
self._waiting = 0
<pre class="brush:python;toolbar:false;">async def acquire(self):
async with self._condition:
while self._value old:
# 只有变大才唤醒，避免惊群
for _ in range(min(self._waiting, n - old)):
self._condition.notify()
注意：release() 不是协程，调用方必须保证线程/协程安全；set_limit() 是同步的，但 notify 是异步生效的。
配合 aiohttp.ClientSession 控制请求并发的实际写法
真实场景中，比如用 aiohttp 爬一批 URL，希望根据响应延迟或错误率动态收紧并发数。这时不能只靠信号量，还得采集指标。
- 每个请求包装进
async def fetch(url, sem)，在try/except后统计成功/失败/耗时 - 另起一个监控协程，每 5 秒计算最近 20 次请求的平均延迟和错误率，调用
sem.set_limit(...) -
ClientSession本身也有connector限制，记得设为 None 或足够大，否则会和信号量叠加限流
示例关键片段：
sem = DynamicSemaphore(10)
<p>async def fetch(url):
async with sem: # 这里会走 DynamicSemaphore.acquire()
start = time.time()
try:
async with session.get(url, timeout=10) as resp:
await resp.text()
return {"ok": True, "latency": time.time() - start}
except Exception as e:
return {"ok": False, "latency": time.time() - start}</p><p>async def monitor():
history = deque(maxlen=20)
while True:
await asyncio.sleep(5)
if len(history) >= 10:
avg_lat = sum(h["latency"] for h in history) / len(history)
err_rate = sum(1 for h in history if not h["ok"]) / len(history)
new_limit = max(1, min(50, int(30 / (avg_lat + 0.1)) if avg_lat > 0.5 else 30))
if err_rate > 0.2:
new_limit = max(1, new_limit // 2)
sem.set_limit(new_limit)</p>
容易忽略的边界：取消任务时的信号量泄漏
如果某个任务在 await sem.acquire() 时被 cancel，DynamicSemaphore 的当前实现不会自动 release，导致信号量永久卡死——因为 __aexit__ 没触发，而我们又没在 acquire 里加 cancel handler。
正确做法是在 acquire 内部用 asyncio.shield() 包住 wait，并捕获 CancelledError 做清理：
async def acquire(self):
async with self._condition:
while self._value <p>更稳妥的方案是把 acquire 改成支持 <code>__aenter__/__aexit__</code> 的上下文管理器，并在 <code>__aexit__</code> 里做 release 判断——但这就要求你始终用 <code>async with</code>，不能裸调 acquire/release。</p><p>实际项目里，动态信号量最难的从来不是怎么调大，而是调小之后，如何让正在跑的任务优雅退场、等待中的任务及时感知、被取消的任务不拖垮整个池子。</p>Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
