本文介绍一种解耦、可靠且生产友好的方案：将令牌管理与爬虫逻辑分离，通过外部定时任务更新 redis 中的 token，并在 scrapy 请求中动态读取，避免阻塞式续期导致的并发或超时问题。
本文介绍一种解耦、可靠且生产友好的方案：将令牌管理与爬虫逻辑分离，通过外部定时任务更新 redis 中的 token，并在 scrapy 请求中动态读取，避免阻塞式续期导致的并发或超时问题。
在使用 Scrapy 调用需认证的 API（如 JWT 令牌有效期为 60 分钟）时，不建议在 Spider 内部同步等待或轮询续期——Scrapy 是异步框架，start_requests 和 parse 方法无法自然挂起整个爬虫来等待令牌刷新，强行阻塞会破坏调度逻辑、引发超时或重复登录。
✅ 推荐架构：Token 管理与爬虫职责分离
- ✅ Spider 只负责“读取”和“使用” token：每次请求前从 Redis 获取最新有效 token；
- ✅ 独立服务负责“获取”和“刷新” token：通过定时任务（如 cron 或 APScheduler）每 55 分钟调用认证接口，写入 Redis 并设置过期时间（如 EX 3600），确保始终可用；
- ✅ 高可用 & 容错：即使 token 暂时失效，Spider 可结合重试 + fallback 逻辑（如捕获 401 后触发手动刷新钩子）。
? 示例：Spider 中安全获取 token
import redis
import json
import scrapy
class ApiSpider(scrapy.Spider):
name = 'api_spider'
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# 复用 Redis 连接（推荐使用连接池）
self.redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
def get_access_token(self):
"""从 Redis 安全读取 token；若不存在则返回 None，由上层处理"""
return self.redis_client.get('workable:access_token')
def start_requests(self):
# 首次请求前确保有 token（可选预检）
token = self.get_access_token()
if not token:
self.logger.warning("No access token found in Redis — check renewal service.")
# 示例目标 API 请求
yield scrapy.Request(
url="https://api.example.com/data",
headers={
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
},
callback=self.parse_data,
errback=self.errback
)
def parse_data(self, response):
if response.status == 401:
self.logger.error("Token expired or invalid — trigger manual refresh or alert.")
# 可扩展：触发告警、降级逻辑，或调用内部刷新接口（不推荐阻塞）
else:
yield {"data": json.loads(response.text)}
⚙️ 独立 Token 刷新脚本（refresh_token.py）
Web Accessibility (WCAG 2 AA))
下载
将 WCAG 2 AA 无障碍标准应用于设计系统、UI 组件和网页——通过可运行的验证器进行对比度计算，采用设计 token 模式使颜色天生可访问，设定排版和焦点状态规则，并提供语义检查清单。适用于构建或审查 UI。
import redis
import requests
import json
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
REDIS_KEY = "workable:access_token"
REDIS_TTL = 3600 # 60 分钟，与 JWT 有效期一致
def fetch_new_token():
auth_url = "https://api.example.com/authenticate"
payload = {"username": "username", "password": "password"}
headers = {"Content-Type": "application/json"}
try:
resp = requests.post(auth_url, json=payload, headers=headers, timeout=10)
resp.raise_for_status()
token = resp.json()["jwt"]
r = redis.Redis(host='localhost', port=6379, db=0)
r.setex(REDIS_KEY, REDIS_TTL, token) # SET + EXPIRE 原子操作
logger.info("✅ New access token refreshed and stored in Redis.")
except Exception as e:
logger.error(f"❌ Failed to refresh token: {e}")
if __name__ == "__main__":
fetch_new_token()
? 定时执行（Linux cron 示例）：
# 每 55 分钟执行一次，预留 5 分钟缓冲 */55 * * * * /usr/bin/python3 /path/to/refresh_token.py >> /var/log/token_refresh.log 2>&1
? 注意事项与最佳实践：
- ❗ 不要在 start_requests 或 parse 中直接调用 requests.post(...) 获取新 token —— 这会阻塞 Twisted reactor，导致性能下降甚至死锁；
- ✅ 使用 redis-py 的连接池（ConnectionPool）提升并发稳定性；
- ✅ 在生产环境启用 Redis 密码认证与 TLS，并限制 redis-cli 权限；
- ✅ 对 401 Unauthorized 响应做统一中间件处理（如自定义 DownloaderMiddleware），实现透明重试或 token 重载；
- ✅ 日志中记录 token 刷新时间与状态，便于审计与故障排查。
该方案将认证生命周期交由专用服务管理，Scrapy 专注高效抓取，兼顾可维护性、可观测性与横向扩展能力。
