FastAPI中外部网站共享单会话管理及多Worker适配问题

原方案的局限性 #

你当前用threading.Lock的方案仅在**单Worker（单进程）**环境下有效。因为gunicorn的每个Worker都是独立的Python进程，进程间内存完全隔离，每个Worker都会初始化自己的OldAppSession实例，各自维护会话——不仅无法实现"全局单活跃会话"的需求，线程锁也只能限制当前进程内的线程，跨进程完全无效。

适配多Worker的可行方案 #

方案1：基于Redis的分布式会话+分布式锁 #

这是多进程/多实例环境下共享会话的通用方案，核心思路是把会话凭证（Cookie、Token等）存在共享存储（Redis）中，并用分布式锁保证同一时间只有一个进程执行登录刷新操作。

实现思路：

• 将requests.Session的CookieJar序列化为可存储的格式（如JSON），存入Redis并设置与外部网站会话一致的过期时间。
• 用Redis的SETNX命令实现分布式锁：尝试获取锁的进程中，只有拿到锁的才能执行login()，刷新会话后更新Redis中的凭证。
• 其他进程遇到401/302时，先尝试获取锁，拿不到就等待一段时间，再从Redis拉取最新凭证重试请求。

简化代码示例：

import redis
import json
import requests
import time
import logging
from requests.cookies import RequestsCookieJar
from fastapi import HTTPException, status

logger = logging.getLogger(__name__)

class RedisOldAppSession:
    def __init__(self, email, password, redis_host="localhost", redis_port=6379):
        self.email = email
        self.password = password
        self.redis_client = redis.Redis(host=redis_host, port=redis_port, db=0)
        self.session_key = "old_app_session_cookies"
        self.lock_key = "old_app_login_lock"
        self.lock_timeout = 10  # 登录操作最长超时，防止锁死

        # 初始化时检查Redis中是否有有效会话，无则登录
        if not self._get_cookies():
            self.login()

    def _get_cookies(self):
        """从Redis获取会话Cookie"""
        cookie_str = self.redis_client.get(self.session_key)
        if not cookie_str:
            return None
        cookie_dict = json.loads(cookie_str)
        jar = RequestsCookieJar()
        for k, v in cookie_dict.items():
            jar.set(k, v)
        return jar

    def _set_cookies(self, cookie_jar):
        """将会话Cookie存储到Redis"""
        cookie_dict = {c.name: c.value for c in cookie_jar}
        # 设置过期时间，与外部网站会话超时一致（示例为3600秒）
        self.redis_client.setex(self.session_key, 3600, json.dumps(cookie_dict))

    def _acquire_lock(self):
        """获取分布式锁"""
        return self.redis_client.set(self.lock_key, "locked", ex=self.lock_timeout, nx=True)

    def _release_lock(self):
        """释放分布式锁"""
        self.redis_client.delete(self.lock_key)

    def login(self):
        """执行登录，更新Redis中的会话凭证"""
        if not self._acquire_lock():
            return False
        
        try:
            # 执行实际登录逻辑
            session = requests.Session()
            login_response = session.post(
                "https://external-site/login",
                data={"email": self.email, "password": self.password}
            )
            login_response.raise_for_status()
            
            # 登录成功后存储Cookie到Redis
            self._set_cookies(session.cookies)
            return True
        except Exception as e:
            logger.error(f"Login failed: {str(e)}")
            return False
        finally:
            # 无论成功失败，都释放锁
            self._release_lock()

    def _request(self, method, url, data=None):
        session = requests.Session()
        cookies = self._get_cookies()
        if cookies:
            session.cookies.update(cookies)
        
        response = getattr(session, method)(url, data=data)
        if response.status_code in [401, 302]:
            # 尝试登录刷新会话
            login_success = self.login()
            if not login_success:
                # 等待其他进程完成登录
                for _ in range(3):
                    time.sleep(1)
                    cookies = self._get_cookies()
                    if cookies:
                        session.cookies.update(cookies)
                        response = getattr(session, method)(url, data=data)
                        if response.status_code not in [401, 302]:
                            return response
                logger.error("Failed to refresh old app session")
                raise HTTPException(
                    status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                    detail="Failed to connect to old app"
                )
            # 用新凭证重试请求
            session.cookies.update(self._get_cookies())
            return getattr(session, method)(url, data=data)
        return response

    def get(self, url):
        return self._request("get", url)

    def post(self, url, data):
        return self._request("post", url, data=data)

# 初始化实例，所有Worker共享Redis中的会话
old_app_session = RedisOldAppSession("test@example.com", "12345")

方案2：独立会话代理服务 #

把会话管理逻辑抽成一个独立的微服务（比如用FastAPI/Flask实现），所有FastAPI Worker都通过调用这个代理服务来获取外部网站的数据，由代理服务维护唯一的会话。

优势：

• 完全避免多进程共享状态的问题，代理服务本身用单进程多线程模式即可，用原来的threading.Lock就能保证单会话。
• 后续扩展方便，比如添加会话缓存、限流等逻辑，只需修改代理服务。

简化流程：

1. 启动会话代理服务，内部维护OldAppSession实例。
2. FastAPI Worker需要调用外部网站时，向代理服务发送请求（比如GET /proxy?url=https://external-site/data）。
3. 代理服务用自身维护的会话请求外部网站，将结果返回给FastAPI Worker。

关键注意事项 #

• 分布式锁超时设置：必须设置合理的锁超时时间，防止登录过程中进程崩溃导致锁一直被占用。
• 会话过期同步：Redis中存储的会话凭证过期时间要和外部网站的会话超时保持一致，或提前1-2分钟刷新，避免请求时刚好过期。
• 异常降级处理：登录失败时要返回明确错误，避免无限重试拖垮服务。
• 敏感数据加密：若存储的Cookie/Token包含敏感信息，需加密后再存入Redis，防止数据泄露。
• 锁的原子性：使用Redis的SETNX（或set nx ex）命令保证锁的原子性，避免多个进程同时获取锁。

内容的提问来源于stack exchange，提问作者Ruuza