用 asyncio.semaphore(n) 包住需要限并发的操作，同一时间最多有 n 个任务进入这段代码。使用 async with 可在退出时释放许可。

模拟十项 I/O 任务

将脚本保存为 limit_tasks.py。示例用 sleep 模拟等待，限制并发为 3。

import asyncio

async def main():
    semaphore = asyncio.Semaphore(3)
    active = peak = 0

    async def run(number):
        nonlocal active, peak
        async with semaphore:
            active += 1
            peak = max(peak, active)
            try:
                await asyncio.sleep(0.01)
                return number * 2
            finally:
                active -= 1

    values = await asyncio.gather(*(run(i) for i in range(10)))
    print(peak, values)

if __name__ == "__main__":
    asyncio.run(main())

限制只适用于 async with 内的操作。gather 返回值仍按传入顺序排列。

核对峰值和返回数量

执行脚本并对照预期。

Biopython

全面的分子生物学工具集。适用于序列操作、文件解析（FASTA/GenBank/PDB）、系统发育分析，以及通过 Bio.Entrez 以编程方式访问 NCBI/

下载

python3 limit_tasks.py
# 3 [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

调小许可数后，峰值也应随之降低。

任务很多时再限制等待队列

若输入有数十万项，逐步放入有容量的队列，并用固定数量的 worker 消费。

queue = asyncio.Queue(maxsize=100)
# 在生产者协程中
await queue.put(item)

这个片段只演示队列背压。实际使用还需配套消费者和退出信号。

Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！