解决 Flask 热重载中 ‘不是套接字’ 错误：数据库连接与线程管理

Exception in thread Thread-2 (serve_forever):
Traceback (most recent call last):
  File "C:\...\threading.py", line 1016, in _bootstrap_inner
    self.run()
  File "C:\...\threading.py", line 953, in run
    self._target(*self._args, **self._kwargs)
  File "C:\...\site-packages\werkzeug\serving.py", line 806, in serve_forever
    super().serve_forever(poll_interval=poll_interval)
  File "C:\...\socketserver.py", line 232, in serve_forever
    ready = selector.select(poll_interval)
  File "C:\...\selectors.py", line 324, in select
    r, w, _ = self._select(self._readers, self._writers, [], timeout)
  File "C:\...\selectors.py", line 315, in _select
    r, w, x = select.select(r, w, w, timeout)
OSError: [WinError 10038] An operation was attempted on something that is not a socket

import threading
from flask import Flask
# 假设 MyDBClass 是一个自定义的数据库连接类
# 并且在其初始化时会启动一个独立的线程来管理连接或请求队列

class PostgreDB:
    def __init__(self):
        print("数据库实例被创建，并启动后台线程...")
        self.background_thread = threading.Thread(target=self._run_background_task)
        self.background_thread.daemon = False # 关键：如果不是守护线程，进程不会等待它结束
        self.background_thread.start()

    def _run_background_task(self):
        # 模拟后台任务，如处理请求队列
        while True:
            # ... 实际的数据库操作或队列处理
            pass # 简化，实际会有sleep或事件等待

    def close(self):
        print("数据库实例被关闭，尝试终止后台线程...")
        # 实际中需要更复杂的线程终止逻辑
        pass

app = Flask(__name__)

# 问题根源：在全局作用域创建数据库实例
db = PostgreDB() 

@app.route('/')
def index():
    return "Hello, Flask!"

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5500, debug=True)

from flask import g # 导入 g 对象

def get_db():
    """
    函数将数据库实例插入到 Flask 的全局变量命名空间 `g` 中，
    该命名空间在应用上下文生命周期结束后被销毁。
    """
    if 'db' not in g:
        # 假设 MyDB 是你的数据库连接类，确保其内部线程在关闭时能被正确终止
        g.db = MyDB()  # 创建一个新的数据库连接
    return g.db

from flask import Flask, request, g
# 假设 MyDB 是你的数据库连接类，需要实现 close 方法来清理资源
class MyDB:
    def __init__(self):
        print("数据库实例被创建...")
        # 可以在这里启动线程，但要确保它们是守护线程或能被正确管理
        # 或者，更好的做法是让数据库连接本身不依赖于独立的、非守护线程
        # 如果确实需要后台线程，请确保在 close() 中能可靠地终止它
        # self.background_thread = threading.Thread(target=self._run_task)
        # self.background_thread.daemon = True # 确保是守护线程，随主进程退出
        # self.background_thread.start()

    def get_name(self, query):
        # 模拟数据库操作
        return f"User_{query['id']}"

    def close(self):
        print("数据库实例被关闭...")
        # 确保在这里关闭所有数据库连接和终止所有内部线程
        # if hasattr(self, 'background_thread') and self.background_thread.is_alive():
        #     # 发送信号给线程使其退出
        #     pass

# 创建 Flask 应用的工厂函数
def create_app():
    app = Flask(__name__)

    @app.before_request
    def before_request():
        g.db = get_db()

    # 注册路由
    @app.route('/')
    def index():
        name = g.db.get_name({"id": 123})  # 如何在代码中使用数据库类
        return f"Hello, {name}!"

    # 在应用上下文销毁时关闭数据库连接
    @app.teardown_appcontext
    def teardown_db(exception):
        db_instance: MyDB | None = g.pop('db', None)
        if db_instance is not None:
            db_instance.close()

    return app

if __name__ == '__main__':
    app = create_app()
    app.run(host='0.0.0.0', port=5500, debug=True)