运行中微服务进程的PostgreSQL数据库连接问题

问题背景 #

你当前的Flask应用在Gunicorn多Worker模式下运行，使用全局的PostgreSQL长连接，服务运行10-12小时后出现连接失效（stale），无法获取数据。相关配置如下：

应用代码 #

import psycopg2, json
from flask_cors import CORS
from flask import Flask, Response, request

app = Flask(__name__)
CORS(app)
# 全局初始化数据库长连接
db_connection = psycopg2.connect(host="some_host_address", port="some_port", database="some_db_name",
                                 user="some_user_name", password="some_password")

@app.get("/user")
def get_user_data():
    user = request.args.get('user')
    # 注意：此处存在SQL注入风险
    query = "Select * from user.data where user = '{}'".format(user)
    cursor = db_connection.cursor()
    cursor.execute(query)
    user_data = cursor.fetchall()
    return Response(json.dumps(user_data), mimetype="application/json")

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=4000)

Gunicorn部署命令 #

python3 -m gunicorn -b :4000 --limit-request-line 0 --timeout 300 -w 4 example.flask_app:app

你的核心疑问：

• 为什么每次请求都需要打开和关闭数据库连接？
• PostgreSQL基于TCP，为什么不能长期保持这个TCP连接？
• 原本以为长连接能避免性能损耗，现在反而出问题，该怎么理解？

疑问解答 #

1. 为什么长连接会失效？ #

PostgreSQL的TCP连接并非能永久保持：

• 数据库端超时机制：PostgreSQL默认有idle_in_transaction_session_timeout、tcp_keepalives_idle等参数，长时间空闲的连接会被数据库主动断开。
• 网络设备清理：网关、防火墙等中间网络设备会自动清理长时间无数据传输的TCP连接，这是通用的网络层策略，和数据库本身无关。
• 多Worker的隐性问题：你用了4个Gunicorn Worker，每个Worker是独立进程，启动时都会初始化自己的连接，但单个Worker的连接一旦因空闲被断开，代码没有重连逻辑，后续请求直接失败。

2. 为什么不能依赖全局长连接？ #

全局长连接在多进程/多线程Web服务中存在两个致命问题：

• 失效无自动恢复：连接被断开后，代码不会检测也不会重建连接，后续所有请求都会报错。
• 线程/进程安全隐患：psycopg2的连接和游标并非线程安全，即使单Worker模式下，若后续扩展多线程，全局连接会导致数据混乱、游标异常等问题。

3. 每次请求开闭连接真的会有性能损耗吗？ #

单次TCP连接的建立开销（三次握手）在Web请求的整体耗时中占比极低，远小于数据库查询的耗时。实际生产环境的最佳实践是用连接池平衡性能与可靠性：连接池维护一组活跃连接，请求来时直接取用，结束后放回池内，既避免了频繁建连的开销，又能自动处理失效连接的重连。

优化方案 #

方案1：使用连接池（推荐） #

用psycopg2.pool实现连接池，自动管理连接生命周期：

import psycopg2, json
from psycopg2 import pool
from flask_cors import CORS
from flask import Flask, Response, request

app = Flask(__name__)
CORS(app)
# 初始化连接池，设置最小/最大连接数
db_pool = pool.SimpleConnectionPool(
    minconn=1,
    maxconn=10,
    host="some_host_address",
    port="some_port",
    database="some_db_name",
    user="some_user_name",
    password="some_password"
)

@app.get("/user")
def get_user_data():
    user = request.args.get('user')
    # 参数化查询避免SQL注入
    query = "SELECT * FROM user.data WHERE user = %s"
    
    conn = None
    cursor = None
    try:
        # 从连接池获取连接
        conn = db_pool.getconn()
        cursor = conn.cursor()
        cursor.execute(query, (user,))
        user_data = cursor.fetchall()
        return Response(json.dumps(user_data), mimetype="application/json")
    finally:
        # 关闭游标，归还连接到池
        if cursor:
            cursor.close()
        if conn:
            db_pool.putconn(conn)

@app.teardown_appcontext
def close_pool(exception):
    # 应用关闭时销毁连接池
    db_pool.closeall()

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=4000)

方案2：请求级连接管理 #

如果暂时不想用连接池，可在每个请求内建立/关闭连接：

@app.get("/user")
def get_user_data():
    user = request.args.get('user')
    query = "SELECT * FROM user.data WHERE user = %s"
    
    conn = psycopg2.connect(host="some_host_address", port="some_port", database="some_db_name",
                            user="some_user_name", password="some_password")
    cursor = conn.cursor()
    try:
        cursor.execute(query, (user,))
        user_data = cursor.fetchall()
        return Response(json.dumps(user_data), mimetype="application/json")
    finally:
        cursor.close()
        conn.close()

额外提醒：修复SQL注入风险 #

原代码用format拼接SQL语句存在严重注入漏洞，必须改用参数化查询（如示例中的%s占位符），psycopg2会自动处理参数转义，避免安全问题。

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