如果你写过异步爬虫，大概率遇到过这样的情况：用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(f"https://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(f"worker {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(value=N)里的value就是初始计数器的值，代表"同一时间最多允许多少个协程进入临界区"。这个值的选择直接决定系统的吞吐和保护程度，设小了浪费资源，设大了起不到保护作用。

判断合理值的思路有两条：

• 根据下游容量推算。例如对一个接口做限流，你观察到它能稳定支撑大约 20 路并发，再多就开始超时，那value就设在 15 到 20 之间，留出缓冲。
• 根据资源占用推算。如果每个协程持有一个数据库连接，而数据库连接池最大为 10，信号量值直接设 10，避免协程拿到信号量却拿不到连接，白白阻塞在那。

值得提醒的是：value的数值只能是非负整数。设为 0 时，所有acquire()都会永久等待，这在某些"启动前先关闭入口"的场景中有特殊用途，但一般不建议新手这么做，容易把程序挂死；设为大于 1 的数时，多个协程可以同时进入，不需要互斥——这是信号量和锁的关键区别，锁本质是value=1的信号量，被称为互斥信号量（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, timeout=aiohttp.ClientTimeout(total=10)) as resp: resp.raise_for_status() return await resp.read() async def crawl(urls, max_concurrency=10): 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_exceptions=True) return results

这里有三个细节值得圈出来。

第一，max_concurrency=10是限制"同一时刻飞行中的请求数"，而不是总请求数。gather依然会创建全部协程，但只有 10 个能真正发出请求，其余协程都在sem.acquire()等待。这避免了大量 socket 连接同时打开，降低了对目标服务器的压力，也降低了本地文件描述符的占用。

第二，return_exceptions=True必须保留。在最外层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_size=20): 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_workers=5): 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_exceptions=True )

从结果上看，这段时间的请求速率大致稳定在每秒 100 左右——刚好踩在接口容量线上。如果你还需要更严格的时间窗口限流（比如每秒 100 个、每分钟 3000 个），那就得配合令牌桶算法了，信号量擅长的是控制"同时并发数"，而不是"单位时间频率"。这两者的区别很多人在项目里容易混淆，后面我再具体讲。

4. 使用信号量时最容易踩的坑：从排查到解决

4.1 漏释放导致的死锁：主要原因与排查链路

我见过不少异步程序运行到一半突然卡死，最后定位到信号量漏释放。最常见的原因有三个：

1. 手动acquire()后忘记release()，代码写得复杂，异常路径又多，漏掉了finally。
2. 在async with sem内部把信号量对象当普通值传给了子协程，子协程内部自行acquire()但释放时用的是同一个信号量，逻辑上造成双重获取。
3. gather的任务在执行过程中被取消，但取消时没有触发上下文管理器退出（某些写法下会）。

排查卡死问题可以按这几步走：

• 在可疑协程的入口和出口增加日志，确定卡在哪一行。
• 用asyncio.all_tasks()获取当前所有任务，看哪些任务处于PENDING状态。
• 如果卡在acquire()，打印信号量的_value和_waiters长度（内部属性，调试可用），如果_value为 0 且_waiters不为空，说明有人持有许可证但没释放。

import asyncio async def debug_semaphore(sem): print(f"value={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(f"job {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(..., timeout=5)。

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 的适用边界对比

在一个异步程序里，并发控制原语不止信号量一种，选错了会把代码绕得很复杂。我做了一个简单对比表：

在实际项目中，我见过有人用信号量去实现互斥，比如设置Semaphore(1)来保护一个全局字典的写入。这当然能跑，但读代码的人需要理解"这个信号量为什么值是 1"，不如直接用asyncio.Lock()表达意图清晰。原语的选择不只是功能问题，也是代码可读性问题。

5.3 用 Python 原生实现一个迷你信号量，加深理解

如果你对信号量的内部机制还有模糊，试着不借助asyncio自带类，用asyncio.Future自己实现一个：

class MiniSemaphore: def __init__(self, permits=1): 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_retries=2): for attempt in range(max_retries + 1): try: async with sem: async with session.get(url, timeout=ClientTimeout(total=8)) as resp: if resp.status != 200: raise RuntimeError(f"HTTP {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, concurrency=10): 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_exceptions=True ) 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往往就能解决绝大多数问题。它简单、可控、容易验证，是异步编程工具箱里性价比最高的一个组件之一。