如何在Flask 3.0中共享requests.Session并在服务关闭时优雅关闭？

核心问题分析 #

你遇到的请求缓慢和套接字泄漏，本质是每次请求新建requests.Session导致无法复用TCP连接池；直接用全局Session出现资源警告，是因为没有在应用退出时主动关闭Session，导致资源未被正确回收。

解决方案1：单进程/多进程环境下的全局Session（推荐生产多进程部署） #

如果你的WSGI服务器（如Gunicorn）采用多进程模式，每个进程独立维护一个全局Session是安全的，且能最大化复用连接池。代码实现如下：

from flask import Flask, jsonify
import requests
import atexit

app = Flask(__name__)
# 全局初始化Session
session = requests.Session()

# 注册进程退出时的清理函数，确保Session被关闭
atexit.register(session.close)

@app.get("/info")
def info():
    r = session.get("https://api.example.com/info")
    return jsonify(r.json())

关键说明

• atexit.register(session.close)会在Python进程退出时主动调用Session的关闭方法，释放所有持有的套接字连接，彻底解决资源警告和泄漏问题。
• 多进程部署时，每个进程会创建独立的全局Session，进程间资源隔离，不存在线程安全风险。

解决方案2：多线程环境下的线程本地Session（适配Flask开发服务器） #

如果服务运行在多线程环境（如Flask默认开发服务器），直接共享全局Session会有线程安全问题（requests.Session的连接池并非线程安全），此时用线程本地存储为每个线程单独维护Session：

from flask import Flask, jsonify
import requests
import threading
import atexit

app = Flask(__name__)
# 线程本地存储，每个线程拥有独立的Session实例
local_session = threading.local()

def get_session():
    """获取当前线程的Session，不存在则创建"""
    if not hasattr(local_session, "instance"):
        local_session.instance = requests.Session()
    return local_session.instance

# 进程退出时清理所有线程的Session（可选，进程退出后系统会自动回收资源）
def cleanup_sessions():
    if hasattr(local_session, "instance"):
        local_session.instance.close()

atexit.register(cleanup_sessions)

@app.get("/info")
def info():
    session = get_session()
    r = session.get("https://api.example.com/info")
    return jsonify(r.json())

关键说明

• threading.local()保证每个线程只能访问自己的Session实例，避免多线程下连接池的状态混乱。
• 线程内的所有请求复用同一个Session，保留了连接池复用的优势，有效提升请求速度。

为什么之前的全局变量方案会有警告？ #

直接定义全局session = requests.Session()但未主动关闭时，Python退出时虽会尝试调用Session的析构函数，但可能因循环引用、GC时机等问题导致资源未被及时释放，从而触发资源警告。通过atexit主动注册关闭逻辑，能确保Session在进程退出时被正确清理。

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