Python爬虫零基础实战⑧:aiohttp异步高性能爬虫,百倍提速并发请求,彻底解决requests串行爬取慢、效率低难题

pip install aiohttp -i https://pypi.douban.com/simple/

import aiohttpimport asyncio# 全局请求头伪装防爬HEADERS = {    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"}# 异步请求函数async def fetch_single_url(session, url):    try:        # 异步发起GET请求，设置10秒超时        async with session.get(url, headers=HEADERS, timeout=aiohttp.ClientTimeout(total=10)) as resp:            html = await resp.text()            print(f"✅ 请求成功 状态码：{resp.status}")            return html    except Exception as e:        print(f"❌ 请求失败：{str(e)}")        return None# 异步主函数async def main():    # 全局创建异步会话，全程复用（核心！不要重复创建）    async with aiohttp.ClientSession() as session:        result = await fetch_single_url(session, "https://www.baidu.com")        print("页面源码长度：", len(result))# 启动异步程序入口if __name__ == "__main__":    asyncio.run(main())

import aiohttpimport asyncioHEADERS = {    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"}async def fetch_url(session, url):    try:        async with session.get(url, headers=HEADERS, timeout=aiohttp.ClientTimeout(total=10)) as resp:            html = await resp.text()            print(f"✅ 成功爬取：{url} 状态码：{resp.status}")            return {"url": url, "html": html, "status": resp.status}    except Exception as e:        print(f"❌ 爬取失败 {url}：{str(e)}")        return {"url": url, "html": None, "status": 0}async def main():    # 批量请求链接列表    url_list = [        "https://www.baidu.com",        "https://www.qq.com",        "https://www.sogou.com",        "https://www.taobao.com"    ]    async with aiohttp.ClientSession() as session:        # 批量创建异步任务        task_list = [asyncio.create_task(fetch_url(session, url)) for url in url_list]        # 等待所有任务执行完成，收集结果        all_result = await asyncio.gather(*task_list)        # 遍历打印结果        for res in all_result:            print(f"链接：{res['url']} 源码长度：{len(res['html']) if res['html'] else0}")if __name__ == "__main__":    asyncio.run(main())

import aiohttpimport asyncio# 核心配置：限制最大并发数（根据网站反爬强度调整，一般5-10最佳）MAX_CONCURRENT = 5HEADERS = {    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"}async def safe_fetch(session, sem, url):    # 限制并发：同一时间最多执行MAX_CONCURRENT个任务    async with sem:        try:            async with session.get(url, headers=HEADERS, timeout=aiohttp.ClientTimeout(total=15)) as resp:                html = await resp.text()                print(f"✅ 爬取完成：{url}")                return url, html        except Exception as e:            print(f"❌ 爬取异常 {url}：{e}")            return url, Noneasync def main():    # 在main内部创建信号量（正确写法，绑定当前事件循环）    sem = asyncio.Semaphore(MAX_CONCURRENT)    # 批量生成15条测试链接    url_list = [f"https://www.baidu.com?page={i}" for i in range(1, 16)]    async with aiohttp.ClientSession() as session:        tasks = [asyncio.create_task(safe_fetch(session, sem, url)) for url in url_list]        results = await asyncio.gather(*tasks)    print("\n🎉 全部任务执行完成！")    # 遍历结果示例    # for url, html in results:    #     if html:    #         print(f"{url} 页面长度：{len(html)}")if __name__ == "__main__":    asyncio.run(main())

import aiohttpimport asyncioasync def async_post():    url = "https://httpbin.org/post"    headers = {        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",        "Content-Type": "application/json"    }    # 提交JSON参数    post_data = {        "username": "spider_test",        "type": "async_crawl"    }    async with aiohttp.ClientSession() as session:        async with session.post(url, headers=headers, json=post_data) as resp:            # 异步解析JSON数据            res_data = await resp.json()            print("响应数据：", res_data)if __name__ == "__main__":    asyncio.run(async_post())

import aiohttpimport asynciofrom bs4 import BeautifulSoup# 并发限制配置MAX_CONCURRENT = 5HEADERS = {    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"}# 异步请求+数据解析async def crawl_and_parse(session, sem, url):    async with sem:        try:            async with session.get(url, headers=HEADERS, timeout=aiohttp.ClientTimeout(total=10)) as resp:                html = await resp.text()                # 同步解析页面                soup = BeautifulSoup(html, "lxml")                title = soup.title.get_text().strip() if soup.title else "无标题"                print(f"✅ 解析成功：{url} | 标题：{title}")                return {"url": url, "title": title}        except Exception as e:            print(f"❌ 解析失败 {url}：{e}")            return Noneasync def main():    # 信号量必须在main内创建，绑定当前事件循环    sem = asyncio.Semaphore(MAX_CONCURRENT)    url_list = [        "https://www.baidu.com",        "https://www.qq.com",        "https://www.sogou.com"    ]    async with aiohttp.ClientSession() as session:        tasks = [asyncio.create_task(crawl_and_parse(session, sem, url)) for url in url_list]        result_list = await asyncio.gather(*tasks)        # 过滤空数据        valid_data = [res for res in result_list if res]        print(f"\n🎉 批量爬取完成，有效数据{len(valid_data)}条")if __name__ == "__main__":    asyncio.run(main())