来源：网络

📖 什么是Requests库？

为什么需要学习？

在爬虫开发中，Requests库扮演着至关重要的角色：

• 效率提升
  ：合理使用可以大幅提高爬取速度
• 稳定性增强
  ：避免常见的错误和异常情况
• 代码优化
  ：让爬虫程序更加简洁易维护

基本概念

了解基本语法、常用方法和最佳实践是学习的基础。

🛠️ 环境准备

在开始之前，请确保你已经安装了必要的依赖：

pip install requests beautifulsoup4 lxml

🚀 快速上手

让我们通过一个简单的例子来理解基本用法： 

import requests

示例代码 - 基础用法演示

def basicexample(): """使用 Requests 库发送 HTTP 请求"""

headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)" }

session = requests.Session()

try: response = session.get( "https://httpbin.org/get", headers=headers, timeout=30 )

if response.statuscode == 200: data = response.json() if "json" in response.headers.get("content-type", "") else response.text return data else: print(f"请求失败: 状态码={response.statuscode}") return None

except Exception as e: print(f"发生错误: {e}") return None

if name == "main": result = basicexample() if result is not None: print("数据获取成功！")

⚡ 进阶技巧

掌握了基础之后，来看看一些高级技巧：

技巧1：异常处理

技巧2：并发控制

技巧3：日志记录

💻 实战案例

下面是一个完整的实战项目示例，展示了如何在实际场景中应用所学知识。

import asyncio import aiohttp from typing import List, Dict, Optional from dataclasses import dataclass from datetime import datetime 
@dataclass class ArticleData: """文章数据结构""" title: str content: str url: str publishtime: Optional[str] = None
class AsyncCrawler: """异步爬虫类 - 高效采集工具"""
def init(self, maxconcurrent: int = 5, timeout: int = 30, retrytimes: int = 3): self.maxconcurrent = maxconcurrent self.timeout = aiohttp.ClientTimeout(total=timeout) self.retrytimes = retrytimes self.results: List[ArticleData] = [] self.failedurls: List[str] = []
async def fetchsingle(self, session: aiohttp.ClientSession, url: str) -> Optional[Dict]: """抓取单个页面""" for attempt in range(self.retrytimes): try: async with session.get(url, timeout=self.timeout) as resp: if resp.status == 200: html = await resp.text() return {{"url": url, "html": html}} elif resp.status == 429: wait = 2 ** attempt print(f"限流中，等待{{wait}}秒...") await asyncio.sleep(wait) else: print(f"HTTP {{resp.status}}: {{url}}") break except Exception as e: print(f"尝试 {{attempt + 1}} 失败: {{e}}")
self.failedurls.append(url) return None
async def parsearticle(self, rawdata: Dict) -> Optional[ArticleData]: """解析文章内容（需根据目标网站定制）""" from bs4 import BeautifulSoup html = rawdata["html"] soup = BeautifulSoup(html, "html.parser")
titletag = soup.find("h1") or soup.find("title") title = titletag.gettext(strip=True) if titletag else "未知标题"
contenttag = soup.find("div", class="content") or soup.find("article") content = contenttag.gettext(strip=True) if contenttag else ""
return ArticleData( title=title, content=content[:500], url=rawdata["url"] )
async def crawl(self, urls: List[str]) -> List[ArticleData]: """并发抓取多个URL""" semaphore = asyncio.Semaphore(self.maxconcurrent)
async def limitedfetch(session, url): async with semaphore: return await self.fetchsingle(session, url)
connector = aiohttp.TCPConnector(limit=self.maxconcurrent) async with aiohttp.ClientSession(connector=connector) as session: tasks = [limitedfetch(session, url) for url in urls] rawresults = await asyncio.gather(*tasks)
for raw in rawresults: if raw: article = await self.parsearticle(raw) if article: self.results.append(article)
return self.results
def exporttojson(self, filepath: str): """导出结果为JSON""" import json data = [ {{"title": a.title, "url": a.url, "time": a.publishtime}} for a in self.results ] with open(filepath, "w", encoding="utf-8") as f: json.dump(data, f, ensureascii=False, indent=2)
async def main(): """使用示例""" crawler = AsyncCrawler(maxconcurrent=3) urls = ["https://example.com/page/{{i}}" for i in range(1, 6)]
articles = await crawler.crawl(urls)
print(f"成功采集: {{len(articles)}} 篇") print(f"失败 URL: {{len(crawler.failedurls)}} 个")
crawler.exporttojson("articles_{{datetime.now():%Y%m%d}}.json")
if name == "main": asyncio.run(main())

🚀 性能优化

当数据量增大时，性能问题会变得突出。这里分享几个优化策略：

⚠️ 避坑指南

根据我的实战经验，以下是一些常见的坑和解决方案：

✨ 写在最后

Requests库的学习需要循序渐进，不要急于求成。

长按或扫描下方二维码，免费获取 Python公开课和大佬打包整理的几百G的学习资料，内容包含但不限于Python电子书、教程、项目接单、源码等等
▲扫描二维码-免费领取