如果你写过异步爬虫大概率遇到过这样的情况用asyncio.gather一次性丢出去几百个协程结果目标网站开始返回超时错误、服务端直接断开连接或者更惨——你的 IP 被临时封禁。这不是代码逻辑出错而是你没给并发加闸门。Python 异步编程里的信号量asyncio.Semaphore就是专门用来限制同时执行协程数量的那个闸门。把它用好了既能享受异步的高吞吐又不会因为并发失控把下游系统打趴下。这篇文章我会从信号量的本质讲起结合真实的限流场景、连接池保护、批量任务调度等案例把asyncio.Semaphore的用法、底层机制、常见坑点和进阶变形整个过一遍。内容面向已经会写基础async/await代码、但还没系统处理过并发上限的开发者如果你刚接触异步不久我也尽量用通俗类比把原理说清楚掌握后可以直接用到项目里。1. 为什么异步程序需要信号量并发失控的典型场景1.1 协程不是免费的无限制并发的代价很多人对异步编程有个误解觉得协程很轻量于是可以无限制地创建。这只说对了一半。单个协程确实很轻创建成本在微秒级别但协程执行过程中往往会占用文件描述符、持有数据库连接、发起外部 HTTP 请求、等待锁资源。这些资源是有限的不会因为你用异步就变多。举个最典型的例子。假设你写了一个数据采集脚本需要请求某个开放接口每次请求大约耗时 200 毫秒。用同步requests循环跑 1000 次总耗时约 200 秒太慢。于是你改成异步import asyncio import aiohttp async def fetch(session, idx): async with session.get(fhttps://api.example.com/data/{idx}) as resp: return await resp.json() async def main(): async with aiohttp.ClientSession() as session: results await asyncio.gather( *(fetch(session, i) for i in range(1000)) ) print(len(results)) asyncio.run(main())这段代码写起来很干净但一跑就出事。1000 个协程同时发出去服务端承接不住于是大量请求超时、连接被重置脚本最终可能只拿到 300 条有效数据还得反复重试。更麻烦的是如果这个接口是第三方付费服务它可能会直接限制你账号的调用额度一晚上的任务量用几分钟额度就烧光了。无限制的并发不是自由而是失控。做系统设计时有一个基本常识任何下游系统都有容量上限。你可以在架构层面做负载均衡、加缓存但作为调用方最直接能控制的就是我最多同时发起多少请求。这正是信号量的职责。1.2 信号量的本质一个带计数的闸门信号量Semaphore是计算机科学里一种经典同步原语最早用于多进程、多线程互斥和资源控制。它的核心是一个整数计数器配合两个操作acquire获取和release释放。打个比方。想象一个停车场入口处有一个电子屏显示剩余车位 N。每进一辆车N 减 1每出一辆车N 加 1。当 N 变成 0 时后来的车必须排队等待直到有车离开释放一个车位。信号量就是这个电子屏和排队系统的抽象。在asyncio里Semaphore不是用来同步多个线程的那需要配合线程锁而是用来在同一个事件循环内限制协程的并发执行数量。它的计数方式和上面停车场的逻辑完全一致。import asyncio async def demo(): sem asyncio.Semaphore(3) # 允许 3 个协程同时通过 async def worker(i): async with sem: await asyncio.sleep(1) print(fworker {i} 完成) await asyncio.gather(*(worker(i) for i in range(10))) asyncio.run(demo())上面这个例子里虽然有 10 个worker协程会被创建但同时执行async with sem内部代码的只有 3 个。其余 7 个会在sem.acquire()处挂起等待直到有人release()。这一点很关键和你直接用asyncio.gather任意并发完全不同——协程还是全部创建但真正进入临界区的只有有限数量。这也引出一个常见误区信号量不能减少协程创建开销它限制的是同一时刻活跃的协程数。1.3 线程里的信号量 vs 异步里的信号量如果你用过threading.Semaphore可能会觉得概念很相似但两者有一个重大区别threading.Semaphore是阻塞式的acquire()调用会让当前线程睡眠而asyncio.Semaphore是非阻塞的acquire()是一个协程它会把当前任务挂起让出事件循环给其他协程运行。所以异步信号量在等待时不会浪费 CPU甚至可以同时等好几个信号量。这一点在实际代码中感受深刻——同样限制最大 10 个并发同步线程方案需要调整线程池大小、处理线程生命周期而异步方案只需要一个信号量对象代码结构几乎不变流畅很多。2. asyncio.Semaphore 核心 API 与底层运作机制2.1 构造参数value的语义与选择asyncio.Semaphore(valueN)里的value就是初始计数器的值代表同一时间最多允许多少个协程进入临界区。这个值的选择直接决定系统的吞吐和保护程度设小了浪费资源设大了起不到保护作用。判断合理值的思路有两条根据下游容量推算。例如对一个接口做限流你观察到它能稳定支撑大约 20 路并发再多就开始超时那value就设在 15 到 20 之间留出缓冲。根据资源占用推算。如果每个协程持有一个数据库连接而数据库连接池最大为 10信号量值直接设 10避免协程拿到信号量却拿不到连接白白阻塞在那。值得提醒的是value的数值只能是非负整数。设为 0 时所有acquire()都会永久等待这在某些启动前先关闭入口的场景中有特殊用途但一般不建议新手这么做容易把程序挂死设为大于 1 的数时多个协程可以同时进入不需要互斥——这是信号量和锁的关键区别锁本质是value1的信号量被称为互斥信号量Mutex。2.2acquire()/release()/async with的写法差异信号量的核心操作有四种写法我逐个说下适用场景。最基础的调用方式是手动acquire()和release()async def worker(sem): await sem.acquire() try: # 实际的并发受限资源操作 await asyncio.sleep(1) finally: sem.release()注意那个finally因为协程内部一旦抛异常而没走到release()这个信号量就永远少一个许可证最终程序会卡死在acquire()上。手动方式容易出这种漏释放问题所以我建议默认用上下文管理器async def worker(sem): async with sem: await asyncio.sleep(1)async with sem会在进入时自动执行acquire()退出时自动执行release()包括异常退出时也会释放。这就避免了漏释放的坑。第三种方式是提前拿好许可证再抛任务适用于生产者消费者模式后面实战部分会展开。第四种是配合asyncio.timeout()或asyncio.wait_for()做超时控制这个在第 4 章详细讲。2.3 事件循环内的 waiters 队列信号量是怎么调度的看源码是理解信号量最直接的办法。asyncio.Semaphore内部维护两个核心字段_value计数器和_waiters双向队列。当调用acquire()时如果_value 0直接把_value减 1返回如果_value 0当前协程会创建一个 Future 加入_waiters然后await这个 Future挂起自己。当某个协程调用release()时如果_waiters非空就set_result()唤醒队首那个 Future被唤醒的协程重新获得执行机会并再次尝试acquire()如果_waiters为空则_value 1。所以实际上信号量的等待队列是先进先出的谁先等待谁先被唤醒这保证了公平性。后续会提到这种公平调度和后面要说的超时取消机制配合时有一些需要特别注意的细节。3. 信号量的三大典型实战场景3.1 场景一对外部 HTTP 接口的并发限流回到开头的爬虫例子。用信号量改造后代码结构几乎不变只是加入一个sem对象import asyncio import aiohttp async def fetch_with_sem(session, url, sem): async with sem: async with session.get(url, timeoutaiohttp.ClientTimeout(total10)) as resp: resp.raise_for_status() return await resp.read() async def crawl(urls, max_concurrency10): sem asyncio.Semaphore(max_concurrency) async with aiohttp.ClientSession() as session: tasks [fetch_with_sem(session, url, sem) for url in urls] results await asyncio.gather(*tasks, return_exceptionsTrue) return results这里有三个细节值得圈出来。第一max_concurrency10是限制同一时刻飞行中的请求数而不是总请求数。gather依然会创建全部协程但只有 10 个能真正发出请求其余协程都在sem.acquire()等待。这避免了大量 socket 连接同时打开降低了对目标服务器的压力也降低了本地文件描述符的占用。第二return_exceptionsTrue必须保留。在最外层gather中如果某个任务因为网络抖动抛异常其他任务不会受影响而且异常会作为结果返回方便统一统计失败率。不加这个参数一个任务异常会导致整个gather取消所有协程。第三async with sem的位置很关键。如果你写成async with session.get(url) as resp: ... async with sem: ...效果完全变了请求本身不受信号量保护只有响应处理被限流。但连接已经发出去了目标服务器该承受的压力一点没少。信号量必须包裹真正需要保护的资源操作这是使用中最容易犯的定位错误。3.2 场景二数据库连接池和复用保护另一个常见场景是异步数据库连接。数据库并发连接数通常有硬上限比如 PostgreSQL 默认最大连接数为 100。如果你的异步任务并发超过连接池规模会直接触发too many clients错误。常见的处理思路是连接池内部自己有限流机制但如果你用的客户端库不支持或你没配置可以在业务侧做保护。import asyncpg class DatabaseGuard: def __init__(self, dsn, pool_size20): self.dsn dsn self.pool_size pool_size self.sem asyncio.Semaphore(pool_size) async def execute(self, query, *args): async with self.sem: conn await asyncpg.connect(self.dsn) try: return await conn.fetch(query, *args) finally: await conn.close()这段代码里信号量保证同一个时间最多有pool_size个协程同时尝试建立连接和执行查询。当然更规范的做法是用连接池库如asyncpg.create_pool但信号量方案在一些轻量脚本里足够有效而且你能精确控制查询中的并发而不仅仅是连接建立中的并发。这里要给一个经验判断如果你发现信号量值设得很大但数据库负载还是上不去说明瓶颈不在连接数而在查询本身的执行计划或磁盘 IO 上。信号量只是闸门不是引擎它不会提升单次操作的效率。3.3 场景三批量任务的并发调度与渐进式退出第三种典型场景是生产者-消费者模式。有一批任务需要处理每个任务内部还包含若干子任务而这些子任务共享同一个外部资源额度。信号量很适合做这种多层级的控制。import asyncio from collections import deque class TaskScheduler: def __init__(self, max_workers5): self.max_workers max_workers self.sem asyncio.Semaphore(max_workers) self.pending deque() def submit(self, coro): self.pending.append(coro) async def _run_one(self, coro): async with self.sem: return await coro async def run(self): workers [ asyncio.create_task(self._worker()) for _ in range(self.max_workers) ] await asyncio.gather(*workers) async def _worker(self): while self.pending: coro self.pending.popleft() await self._run_one(coro)这种实现的好处是有固定数量的 worker 协程每个 worker 处理完一个任务后立即从队列取下一个信号量确保即使实际并发任务超过 worker 数也不会失控。配合asyncio.Queue可以做得更规范但核心思想是把并发控制从具体业务逻辑中剥离出来放在一个可复用的调度器里。再补充一个实际场景。某个数据同步任务每天凌晨要更新 10 万条记录每条记录需要调用一次远程校验接口接口限制每秒最多 100 个请求。这时信号量的作用就很明显sem asyncio.Semaphore(100) async def process_one(record): async with sem: result await remote_check(record) return result async def main(): records load_records() results await asyncio.gather( *(process_one(r) for r in records), return_exceptionsTrue )从结果上看这段时间的请求速率大致稳定在每秒 100 左右——刚好踩在接口容量线上。如果你还需要更严格的时间窗口限流比如每秒 100 个、每分钟 3000 个那就得配合令牌桶算法了信号量擅长的是控制同时并发数而不是单位时间频率。这两者的区别很多人在项目里容易混淆后面我再具体讲。4. 使用信号量时最容易踩的坑从排查到解决4.1 漏释放导致的死锁主要原因与排查链路我见过不少异步程序运行到一半突然卡死最后定位到信号量漏释放。最常见的原因有三个手动acquire()后忘记release()代码写得复杂异常路径又多漏掉了finally。在async with sem内部把信号量对象当普通值传给了子协程子协程内部自行acquire()但释放时用的是同一个信号量逻辑上造成双重获取。gather的任务在执行过程中被取消但取消时没有触发上下文管理器退出某些写法下会。排查卡死问题可以按这几步走在可疑协程的入口和出口增加日志确定卡在哪一行。用asyncio.all_tasks()获取当前所有任务看哪些任务处于PENDING状态。如果卡在acquire()打印信号量的_value和_waiters长度内部属性调试可用如果_value为 0 且_waiters不为空说明有人持有许可证但没释放。import asyncio async def debug_semaphore(sem): print(fvalue{sem._value}, waiters{len(sem._waiters)})需要说明的是访问下划线私有属性只适合调试正式代码不要依赖它们。真正规范的解决方案很简单永远用async with sem能不用手动acquire/release就不用。这是最简单也很难被打破的纪律。4.2 带超时的acquire()防止任务无限等待有些场景里你不希望协程无限期等待许可证比如定时任务超时后应该放弃本次执行而不是一直排队。asyncio.wait_for()可以给acquire()加超时async def worker(sem, job): try: async with asyncio.timeout(5): async with sem: return await process(job) except asyncio.TimeoutError: log_warning(fjob {job} 等待超时跳过) return None不过要注意一个细节如果超时发生在sem.acquire()的过程中信号量内部的 Future 会被取消但_waiters队列中的这个 Future 仍然存在只是其结果为CancelledError。好在asyncio.Semaphore内部已经处理了取消清理逻辑不会造成队列堆积但我们自己要保证外层异常处理被正确触发。从 Python 3.11 开始asyncio.timeout()返回的是一个异步上下文管理器能同时处理多层超时比asyncio.wait_for()的写法更干净。如果项目 Python 版本还停留在 3.8 或 3.9就继续用asyncio.wait_for(..., timeout5)。4.3 信号量的生命周期全局复用还是每任务创建这是一个非常隐蔽的问题。很多人在每次请求时新建信号量async def bad_worker(): sem asyncio.Semaphore(5) async with sem: await do_work()这等于没限流——因为每个协程都有自己的信号量互相之间根本不知道对方的存在。信号量必须是一个多个协程共享的对象要么定义在模块级要么作为参数传入协程要么挂在某个App容器对象上。反过来也有问题在asyncio.run()顶层创建的信号量被传递到别的线程或别的事件循环中可能无法正常工作。因为asyncio.Semaphore和事件循环绑定它的内部 Future 在哪个循环创建就需要在那个循环里 await。标准做法是每个事件循环里单独创建信号量不要跨循环传递。4.4 公平性困局信号量不是精确速率控制器我还想强调前面提到的那个误区信号量控制的是同时并发数不是每秒请求频率。比如Semaphore(100)不代表平均每秒只有 100 个请求。设想 10 个请求同时进入10 毫秒内全部完成并释放许可证下一批 10 个请求紧接着进入——网络中仍然只有 10 个请求但每秒频率可能已经达到 1000 次。如果你要精确限速到每秒 N 次就需要令牌桶算法。最简单的实现是配合一个定时器class RateLimiter: def __init__(self, rate: float): self.rate rate self.sem asyncio.Semaphore(rate) async def acquire(self): async with self.sem: # 每获得一次许可证就启动一个协程在 1 秒后释放 asyncio.create_task(self._release_after(1.0)) async def _release_after(self, delay): await asyncio.sleep(delay) self.sem.release()这个实现很粗糙但思路清楚信号量不按计数-1释放而是按时间-1释放。实际生产中可以借助asyncio.Queue或三方限流库做更严格的平滑限流但理解这个差异对你选型很重要。5. 信号量的进阶变形BoundedSemaphore 与相关原语对比5.1 BoundedSemaphore 的边界保护asyncio.BoundedSemaphore继承自Semaphore唯一区别是它不允许release()次数超过初始值。也就是说正常信号量如果某个协程多调了一次release()计数值会超过初始值而不报错导致后续并发峰值超过预期非常危险BoundedSemaphore则会在这种情况下抛出ValueError把错误暴露在运行时。import asyncio sem asyncio.Semaphore(2) sem.release() print(sem._value) # 3超卖了 bounded asyncio.BoundedSemaphore(2) bounded.release() # 抛出 ValueError: Semaphore released too many times这个特性在调试多补充release()逻辑时极有价值。如果你从写threading代码转过来习惯手动管理锁的释放建议直接用BoundedSemaphore让不规范代码尽早现形。5.2 与 Lock、Event、Condition 的适用边界对比在一个异步程序里并发控制原语不止信号量一种选错了会把代码绕得很复杂。我做了一个简单对比表原语核心语义典型用途和信号量的区别Lock互斥最多一个持有者保护共享变量的读写本质是value1的信号量但语义更明确Semaphore固定数量许可证限流、连接池保护允许多个协程同时进入BoundedSemaphore带边界检查的许可证需要警惕超卖的场景比普通信号量多一层正确性检查Event事件发生/未发生的标志通知其他协程可以开始了不限制并发数只做信号通知Condition条件满足后才继续多个协程等待同一个条件的多阶段任务通常需要配锁语义更复杂在实际项目中我见过有人用信号量去实现互斥比如设置Semaphore(1)来保护一个全局字典的写入。这当然能跑但读代码的人需要理解这个信号量为什么值是 1不如直接用asyncio.Lock()表达意图清晰。原语的选择不只是功能问题也是代码可读性问题。5.3 用 Python 原生实现一个迷你信号量加深理解如果你对信号量的内部机制还有模糊试着不借助asyncio自带类用asyncio.Future自己实现一个class MiniSemaphore: def __init__(self, permits1): self.permits permits self.waiters [] async def acquire(self): if self.permits 0: self.permits - 1 return fut asyncio.get_running_loop().create_future() self.waiters.append(fut) await fut def release(self): if self.waiters: waiter self.waiters.pop(0) waiter.set_result(None) else: self.permits 1这段代码和官方实现的核心逻辑几乎一样只是省去了取消处理、边界检查等细节。当你自己能写出一个微型信号量再回头用asyncio.Semaphore那些_value、_waiters的概念就彻底不神秘了。而且这种自己动手实现原语的过程能帮你更稳地排掉那些看起来莫名其妙的并发问题。6. 一个完整的限流爬虫案例与实测心得6.1 完整示例带超时、重试与统计的信号量限流爬虫把前面几章的东西串起来写一个相对完整的示例并发限制 10、单请求超时 8 秒、失败自动重试 2 次、最终统计成功数和失败数。import asyncio import aiohttp from aiohttp import ClientTimeout async def fetch_with_retry(session, url, sem, max_retries2): for attempt in range(max_retries 1): try: async with sem: async with session.get(url, timeoutClientTimeout(total8)) as resp: if resp.status ! 200: raise RuntimeError(fHTTP {resp.status}) return await resp.read() except (aiohttp.ClientError, asyncio.TimeoutError, RuntimeError) as e: if attempt max_retries: raise await asyncio.sleep(2 ** attempt) # 指数退避 return None async def run(urls, concurrency10): sem asyncio.Semaphore(concurrency) async with aiohttp.ClientSession() as session: results await asyncio.gather( *(fetch_with_retry(session, u, sem) for u in urls), return_exceptionsTrue ) success [r for r in results if isinstance(r, bytes)] failed [r for r in results if isinstance(r, Exception)] print(f成功: {len(success)}, 失败: {len(failed)}) return results这段代码里信号量的位置在session.get外层确保网络请求本身受控重试逻辑在async with sem的外层这意味着重试时同样会重新获取许可证不会让失败的请求绕过限流。6.2 实测对比不设限流 vs 设置信号量限流我在本地模拟环境里用 1000 个 URL 做了简单对比结果很有参考价值。不设信号量时全部协程同时发出请求前 2 秒大量超时报错约四成请求失败设Semaphore(20)后没有超时总耗时大约是不限流时的 3 倍但成功率接近 100%。这个3 倍耗时其实很正常——你把并发从 1000 压到 20理论耗时必然上升。但换来的稳定性是值得的下游服务没有被冲垮脚本不需要反复重试从整体效率看反而更高。系统设计追求的永远不是单次请求的最快速度而是整体任务的稳定完成率。6.3 调优建议与个人经验最后给几条我实际使用后沉淀下来的经验信号量值不要拍脑袋定。先压测下游接口的极限并发取 60% 到 80% 作为信号量值留出余量。async with sem包裹的范围宁大勿小。如果你不确定这段代码内部还会不会再发出子请求直接包外面全局统一控制。先用BoundedSemaphore调试。尤其在多人协作的项目里它可以快速暴露有人多释放许可证的问题等确认稳定后再换回普通信号量提升性能性能差异其实很小主要是语义问题。不要和线程池混用。asyncio.Semaphore只能作用于当前事件循环里的协程。如果你用asyncio.to_thread或loop.run_in_executor把阻塞任务丢到线程池它是管不住的那种场景需要给线程池本身设置max_workers或用线程版信号量。回调到最初的场景——如果你正在写的异步程序已经出现了并发一高就出错、调低又嫌慢的问题先别急着上消息队列、分布式限流这些重武器一个asyncio.Semaphore往往就能解决绝大多数问题。它简单、可控、容易验证是异步编程工具箱里性价比最高的一个组件之一。