本文介绍如何在 scrapy 项目中可靠地管理短期有效的 api 访问令牌（如 jwt），通过解耦认证逻辑与爬虫主流程，利用外部服务（如 redis）和定时任务实现令牌自动刷新，避免请求因过期而失败。
本文介绍如何在 scrapy 项目中可靠地管理短期有效的 api 访问令牌（如 jwt），通过解耦认证逻辑与爬虫主流程，利用外部服务（如 redis）和定时任务实现令牌自动刷新，避免请求因过期而失败。
在 Scrapy 中直接“在运行时暂停爬取、同步刷新令牌”既违背框架异步非阻塞设计原则，也极易引发竞态、死锁或请求中断等问题。推荐采用职责分离 + 外部状态管理的生产级方案：将令牌生命周期交由独立服务维护，Scrapy 蜘蛛按需获取最新有效令牌。
✅ 推荐架构：Redis + 定时刷新脚本
-
使用 Redis 存储并共享令牌
Redis 提供原子操作、过期时间（EXPIRE）和高并发读写能力，天然适合作为分布式令牌缓存：# settings.py 或 utils.py import redis import json REDIS_CLIENT = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True) TOKEN_KEY = "workable:access_token"
-
编写独立的令牌刷新脚本（refresh_token.py）
该脚本每 55 分钟执行一次（预留 5 分钟缓冲），确保令牌始终有效：# refresh_token.py import json import requests import redis from scrapy.utils.project import get_project_settings settings = get_project_settings() redis_client = redis.Redis( host=settings.get('REDIS_HOST', 'localhost'), port=settings.get('REDIS_PORT', 6379), db=settings.get('REDIS_DB', 0), decode_responses=True ) def fetch_new_token(): url = "https://api.example.com/authenticate" payload = {"username": "username", "password": "password"} headers = {"Content-Type": "application/json"} resp = requests.post(url, json=payload, headers=headers, timeout=10) resp.raise_for_status() token = resp.json()["jwt"] # 设置 Redis key 过期时间为 60 分钟（与令牌实际有效期一致） redis_client.setex(TOKEN_KEY, 3600, token) print(f"[INFO] New access token refreshed: {token[:12]}...") if __name__ == "__main__": fetch_new_token()配合系统定时任务（如 Linux cron）：
# 每 55 分钟刷新一次 */55 * * * * cd /path/to/your/scrapy/project && python refresh_token.py >> /var/log/token_refresh.log 2>&1
-
Scrapy 蜘蛛按需读取令牌（无状态、高可用）
在 start_requests 或中间件中动态获取，无需关心刷新逻辑：import json import redis from scrapy import Spider, Request class WorkableSpider(Spider): name = 'workable' custom_settings = { 'REDIRECT_ENABLED': False, } def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.redis_client = redis.Redis( host='localhost', port=6379, db=0, decode_responses=True ) def start_requests(self): # 每次请求前安全获取当前有效令牌 token = self.redis_client.get("workable:access_token") if not token: raise RuntimeError("Access token missing in Redis — ensure refresh script is running.") headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", } yield Request( url="https://api.example.com/jobs", headers=headers, callback=self.parse_jobs, errback=self.handle_error, ) def parse_jobs(self, response): # 正常解析逻辑 yield {"jobs_count": len(response.json().get("data", []))}
⚠️ 注意事项与最佳实践
- 避免在 parse_access_token 中重复刷新：Scrapy 的回调链不保证执行顺序与时序，无法精确控制“60分钟后触发”，且多请求并发时易造成令牌覆盖冲突。
- Redis 连接池优化：生产环境建议使用连接池（如 redis.ConnectionPool）提升并发性能。
- 令牌安全性：Redis 应启用密码认证与网络隔离；敏感凭证（如账号密码）切勿硬编码，应通过环境变量或密钥管理服务注入。
- 降级与监控：在 start_requests 中增加 token is None 校验，并记录告警；可扩展为自动触发紧急刷新（如首次获取失败时调用 requests.post(...) 同步获取）。
该方案将认证复杂度从爬虫中剥离，显著提升 Spider 的稳定性、可测试性与可扩展性——即使令牌服务短暂不可用，爬虫仍可重试或降级处理，而非整体崩溃。
