Python中的异步任务并发限制与信号量（Semaphore）机制

描述：
在异步编程中，并发任务数量可能超过系统资源承受能力，导致性能下降或错误。信号量（Semaphore）是一种同步原语，用于限制同时运行的并发任务数量，防止资源过载。在Python的asyncio中，Semaphore通过计数器控制对共享资源的访问，确保同一时刻只有指定数量的协程能执行关键代码段。

解题过程循序渐进讲解：

1. 问题场景引入
假设需要异步下载1000个网页，如果同时发起1000个网络请求，可能造成网络阻塞、服务器拒绝或内存耗尽。理想情况是限制同时进行的请求数，例如最多10个。这就是并发限制的需求。

2. 信号量的基本概念
信号量维护一个内部计数器：

• 计数器的初始值表示允许的最大并发数。
• 当协程获取信号量时，计数器减1；如果计数器为0，则协程等待，直到有其他协程释放信号量。
• 当协程释放信号量时，计数器加1，并唤醒等待的协程。

在asyncio中，asyncio.Semaphore类实现了异步版本的信号量。

3. 创建与使用信号量
示例：限制最多3个并发任务。

import asyncio

async def worker(semaphore, task_id):
    async with semaphore:  # 进入时获取信号量，退出时自动释放
        print(f"任务 {task_id} 开始")
        await asyncio.sleep(1)  # 模拟I/O操作
        print(f"任务 {task_id} 结束")

async def main():
    semaphore = asyncio.Semaphore(3)  # 最多3个并发
    tasks = [asyncio.create_task(worker(semaphore, i)) for i in range(10)]
    await asyncio.gather(*tasks)

• 创建信号量semaphore = asyncio.Semaphore(3)，初始计数器为3。
• async with semaphore：上下文管理器自动调用semaphore.acquire()和semaphore.release()。

4. 信号量的底层实现机制
asyncio.Semaphore内部依赖asyncio.Lock和计数器：

• _value：当前计数器值。
• _waiters：一个队列，存储等待获取信号量的协程。
• acquire()方法：如果_value > 0，则_value -= 1并立即返回；否则将当前协程加入_waiters，并等待通知。
• release()方法：_value += 1，如果有等待的协程，则唤醒其中一个（FIFO顺序）。

关键代码简化逻辑：

class Semaphore:
    def __init__(self, value=1):
        self._value = value
        self._waiters = collections.deque()
    
    async def acquire(self):
        while self._value <= 0:
            # 创建Future对象，让当前协程等待
            fut = asyncio.get_event_loop().create_future()
            self._waiters.append(fut)
            await fut  # 等待被release()唤醒
        self._value -= 1
    
    def release(self):
        self._value += 1
        if self._waiters:
            fut = self._waiters.popleft()
            fut.set_result(None)  # 唤醒一个等待的协程

5. 信号量的应用场景

• 限制数据库连接数、网络请求并发数。
• 控制对共享资源的访问（如文件写入）。
• 实现生产者-消费者模型中的缓冲区大小限制。

6. 信号量与BoundedSemaphore的区别

• Semaphore的计数器可以超过初始值（多次调用release()可能导致资源计数异常增多）。
• BoundedSemaphore是Semaphore的子类，在release()时检查计数器是否超过初始值，如果超过则抛出ValueError，避免编程错误。

7. 实际例子：限制并发HTTP请求

import aiohttp
import asyncio

async def fetch(url, semaphore):
    async with semaphore:  # 控制并发数
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as resp:
                return await resp.text()

async def main():
    urls = ["http://example.com"] * 100
    semaphore = asyncio.Semaphore(10)  # 最多10个并发请求
    tasks = [fetch(url, semaphore) for url in urls]
    await asyncio.gather(*tasks)

8. 注意事项

• 信号量必须在同一事件循环中使用。
• 确保每个acquire()都有对应的release()，否则可能导致死锁（使用async with可避免）。
• 信号量适用于控制“可同时执行的任务数”，而不是资源池管理（如数据库连接池需结合其他机制）。

总结：
信号量是异步编程中控制并发度的核心工具。通过计数器与等待队列，它确保系统资源不被过度占用。在asyncio中使用async with semaphore能简洁实现并发限制，提升程序的稳定性和效率。