在Python使用requests库循环抓取数据时，频繁请求可能导致服务器返回401未授权错误。本文将详细介绍如何通过引入重试机制、设置请求延迟以及利用多线程并发处理来构建一个健壮的网络爬虫，有效应对此类问题，确保数据稳定获取，同时优化抓取效率。
理解HTTP 401未授权错误及其成因
HTTP状态码401（Unauthorized）通常表示客户端的请求缺乏有效的身份验证凭证，或者凭证已过期、无效。在循环抓取大量URL时，即使初始请求成功（返回200），后续请求却收到401，这可能由以下原因导致：
会话过期或令牌失效： 服务器可能对会话或认证令牌设置了有效期，长时间或大量请求后，令牌可能失效。IP限制或速率限制： 服务器可能将连续来自同一IP地址的大量请求识别为异常行为，并暂时拒绝服务，有时会以401或其他错误码（如429 Too Many Requests）响应。缺少必要的请求头： 某些API需要特定的User-Agent、Referer或其他自定义请求头才能正常访问，这些在循环中可能被忽略或设置不当。服务器端实现问题： 尽管401有其特定含义，但少数情况下，服务器后端可能错误地使用401来表示其他类型的拒绝服务。
当面对这种问题时，简单的循环请求往往无法满足需求，我们需要一个更智能、更健壮的策略。
优化请求策略：重试、延迟与并发
为了解决循环请求中出现的401错误并提高数据抓取的稳定性，我们可以结合以下三种核心策略：
重试机制： 当请求失败时，不立即放弃，而是等待一段时间后再次尝试。请求延迟： 在连续请求之间加入短暂的暂停，以模拟人类行为，降低被服务器识别为爬虫的风险，并遵守服务器的速率限制。多线程并发： 利用多线程同时处理多个请求，在不牺牲总运行时间的前提下，允许单个请求有足够的延迟。
下面我们将通过一个具体的Python示例来演示如何实现这些策略。
立即学习“Python免费学习笔记（深入）”；
示例代码：构建健壮的数据抓取器
假设我们需要从NHL API抓取多个比赛的数据，并处理可能出现的401错误。
from requests import getfrom time import sleepfrom multiprocessing.pool import ThreadPoolimport logging# 配置日志，以便更好地追踪请求状态和错误logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')# 配置常量MAX_RETRIES = 5 # 最大重试次数DELAY = 10 # 每次重试批次之间的延迟时间（秒）THREAD_POOL_SIZE = 5 # 线程池大小，可根据实际情况调整# 示例游戏ID列表，实际应用中可能远大于此gameIds = [ "2022020022", "2022020028", "2022020044", "2022020072", "2022020088", "2022020090", "2022020091", "2022020092", "2022020093", "2022020094", "2022020095", "2022020096", "2022020097", "2022020098", "2022020099"]all_data = [] # 用于存储所有成功获取的数据# 定义一个处理单个游戏ID的函数def process_game_data(gameId: str) -> tuple[dict, str]: """ 尝试获取单个游戏ID的数据。 返回一个元组：(数据字典, 成功获取数据的gameId) 如果失败，返回({}, "") """ rv = {}, "" url = f"https://api-web.nhle.com/v1/gamecenter/{gameId}/play-by-play" # 可以在这里添加自定义的headers和proxies # headers = {'User-Agent': 'YourCustomUserAgent'} # proxies = {'http': 'http://your_proxy:port', 'https': 'https://your_proxy:port'} try: # 使用with语句确保response资源被正确关闭 with get(url) as response: match response.status_code: case 200: data = response.json() rv = data, gameId logging.info(f"成功获取数据: {gameId}") case 401: logging.warning(f"获取数据失败，Game ID: {gameId}，状态码: 401 Unauthorized. 将在下一轮重试。") # 对于401，我们不抛出异常，而是让其进入下一轮重试 case _: # 对于其他非200/401的状态码，抛出HTTPError response.raise_for_status() logging.error(f"获取数据失败，Game ID: {gameId}，状态码: {response.status_code}") except Exception as e: logging.error(f"无法获取Game ID: {gameId} 的数据，原因: {e}") return rv# 主循环：执行多次重试批次remaining_game_ids = gameIds.copy() # 复制原始列表，以便在重试时修改for attempt in range(MAX_RETRIES): if not remaining_game_ids: logging.info("所有数据已成功获取。") break logging.info(f"开始第 {attempt + 1} 次尝试，剩余 {len(remaining_game_ids)} 个游戏ID。") # 使用线程池并发处理当前批次的请求 # imap_unordered可以无序地返回结果，提高效率 with ThreadPool(THREAD_POOL_SIZE) as pool: # 使用list()来强制迭代并收集所有结果，避免在remove时修改迭代器 results = list(pool.imap_unordered(process_game_data, remaining_game_ids)) # 清空当前剩余ID列表，重新构建 current_successful_ids = [] for data, gameId in results: if data: all_data.append(data) current_successful_ids.append(gameId) # 从remaining_game_ids中移除已成功获取的ID remaining_game_ids = [gid for gid in remaining_game_ids if gid not in current_successful_ids] if remaining_game_ids: logging.info(f"本轮尝试后，仍有 {len(remaining_game_ids)} 个游戏ID未获取。将在 {DELAY} 秒后进行下一次尝试。") sleep(DELAY) # 在每次重试批次之间加入延迟 else: logging.info("所有数据已成功获取。") break# 最终检查未获取的数据if remaining_game_ids: logging.warning(f"经过 {MAX_RETRIES} 次尝试后，仍无法获取以下游戏ID的数据: {remaining_game_ids}")else: logging.info("所有指定的游戏数据均已成功获取。")# all_data 列表现在包含了所有成功获取的游戏数据# print(f"总共获取了 {len(all_data)} 条数据。")
代码解析与最佳实践
导入必要的库：
requests.get：用于发送HTTP GET请求。time.sleep：用于引入延迟。multiprocessing.pool.ThreadPool：用于创建线程池，实现并发请求。logging：用于替代print，提供更专业的日志输出，便于调试和追踪。
配置常量：
MAX_RETRIES：定义了最大重试批次。DELAY：定义了每次重试批次之间的等待时间。THREAD_POOL_SIZE：定义了并发执行的线程数量。合理设置线程池大小可以平衡效率和服务器负载。
process_game_data 函数：
这是核心的请求逻辑，负责处理单个gameId。with get(url) as response:：使用with语句确保请求结束后连接资源被正确释放。match response.status_code:：Python 3.10+ 的match语句提供了一种清晰的方式来处理不同的HTTP状态码。case 200:：成功获取数据，解析JSON并返回。case 401:：遇到401错误时，我们不立即抛出异常，而是记录警告并让函数正常返回空数据。这样，这个gameId会在下一轮重试中再次被处理。case _:：对于其他非200/401的状态码，使用response.raise_for_status()抛出HTTPError，这会自动处理4xx和5xx系列错误（除了我们明确处理的401）。try-except Exception as e:：捕获网络连接问题、DNS解析失败等其他异常，提高程序的健壮性。请求头和代理： 示例中注释掉了headers和proxies的设置。在实际应用中，务必根据目标网站的要求设置User-Agent，必要时使用代理IP池来分散请求来源，进一步降低被封禁的风险。
主循环与重试逻辑：
remaining_game_ids = gameIds.copy()：维护一个需要处理的gameId列表。每次成功获取数据后，将对应的gameId从这个列表中移除。for attempt in range(MAX_RETRIES):：外层循环控制重试的批次。with ThreadPool(THREAD_POOL_SIZE) as pool:：在每次重试批次中创建一个线程池。pool.imap_unordered(process_game_data, remaining_game_ids)：将remaining_game_ids中的每个gameId提交给线程池并行处理。imap_unordered可以无序地返回结果，这在处理大量独立任务时效率更高。跟踪成功请求： 遍历线程池返回的结果，将成功获取的数据添加到all_data列表，并将对应的gameId从remaining_game_ids中移除。sleep(DELAY)： 在每次重试批次结束后，如果仍有未处理的gameId，则暂停DELAY秒，以等待服务器状态恢复或避免触发更严格的限制。提前退出： 如果remaining_game_ids为空，说明所有数据都已获取，可以提前跳出重试循环。
注意事项与进阶优化
User-Agent和请求头： 始终建议设置一个合理的User-Agent，模拟浏览器行为。某些网站可能还需要Referer、Cookie等其他请求头。代理IP： 对于大规模或高频率的抓取，使用代理IP池是必不可少的。在process_game_data函数中动态切换代理可以有效规避IP封禁。会话管理： 如果API需要持久化的认证（如基于requests.Session()的会话），应在process_game_data函数外部初始化Session对象，并将其传递给函数，或者在函数内部使用Session对象进行请求。错误日志： 使用logging模块而非简单的print，可以更好地管理日志级别、输出格式和目标（文件、控制台等），方便问题排查。动态延迟： 可以实现指数退避策略，即每次重试的延迟时间逐渐增加，例如DELAY * (2 ** attempt)，以更好地应对服务器的临时性过载。并发数量： THREAD_POOL_SIZE的设置需谨慎。过多的线程可能导致本地资源耗尽、连接池溢出，甚至对目标服务器造成过大压力。建议从小到大逐步测试，找到最佳平衡点。目标网站的Robots协议和服务条款： 在进行任何爬取活动之前，务必查阅目标网站的robots.txt文件和用户服务协议，确保您的行为合法合规。
总结
通过引入重试机制、合理设置请求延迟以及利用多线程并发处理，我们可以显著提高Python requests库在循环抓取数据时的稳定性和成功率，有效应对HTTP 401未授权等常见错误。构建一个健壮的爬虫不仅需要处理各种HTTP状态码，还需要关注请求频率、会话管理和异常处理，从而在保证数据获取效率的同时，降低对目标服务器的冲击。
以上就是解决Python requests循环请求中遇到的401未授权错误的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1373543.html
微信扫一扫 
支付宝扫一扫 
