Python 异步爬虫：aiohttp + asyncio 实战

安装依赖

pip install aiohttp

异步请求示例

import aiohttp

import asyncio



async def fetch(session, url):

    async with session.get(url) as response:

        return await response.text()



async def main():

    urls = ["http://httpbin.org/get", "http://httpbin.org/ip"]

    async with aiohttp.ClientSession() as session:

        tasks = [fetch(session, url) for url in urls]

        results = await asyncio.gather(*tasks)

        for result in results:

            print(result[:100])



asyncio.run(main())

并发控制（限流）

使用 asyncio.Semaphore 控制并发数，防止被封 IP。

sem = asyncio.Semaphore(10)



async def bounded_fetch(session, url):

    async with sem:

        return await fetch(session, url)

错误重试机制

async def fetch_with_retry(session, url, retries=3):

    for i in range(retries):

        try:

            return await fetch(session, url)

        except Exception as e:

            print(f"重试 {i+1}: {e}")

            await asyncio.sleep(1)

    raise Exception("所有重试失败")

异步爬虫比多线程更轻量，IO 密集型场景下性能提升 5-10 倍。