FastAPI高负载下触发系统资源不足错误，SQL连接未关闭是否有影响？

错误根源 #

你遇到的OSError: [Errno 24] Too many open files是系统文件描述符耗尽导致的，核心问题集中在代码的资源管理和架构设计上：

• 全局单连接+游标未清理：全局初始化的conn被所有请求复用，但每次请求创建的游标cursor从未关闭，高负载下游标持续累积，占用大量文件描述符；同时单连接无法应对并发请求，导致连接阻塞加剧资源耗尽。
• 异步接口混用同步驱动：接口用async def定义，但pyodbc是同步数据库库，同步代码会阻塞FastAPI的事件循环，高负载下事件循环被占满，无法处理新请求，资源也无法及时释放。
• 严重SQL注入风险：用字符串拼接生成SQL语句（如f"SELECT idu FROM indirect.usr WHERE cn='{signinname}'"），不仅存在安全漏洞，还可能因特殊字符导致语法错误。
• 冗余JSON处理：多次来回执行json.dumps和json.loads，浪费CPU资源，增加请求处理时间，间接放大高负载压力。

解决方案 #

1. 正确管理数据库连接与游标 #

删除全局连接，改为在每个请求内通过上下文管理器获取连接和游标，自动完成资源释放：

# 仅保留连接配置，去掉全局conn初始化
def get_db_connection():
    return pyodbc.connect(conn_string, pooling=True, **pooling_options)

# 在接口内使用上下文管理器
with get_db_connection() as conn:
    with conn.cursor() as cursor:
        # 执行SQL操作
        ...

2. 将异步接口改为同步接口 #

由于pyodbc是同步驱动，FastAPI的def接口会自动放入线程池处理，避免阻塞事件循环：

# 把async def替换为def
@app.post("/user_details")
def read_user_details(user: User):
    ...

3. 修复SQL注入问题，使用参数化查询 #

所有SQL语句用占位符?替代字符串拼接，确保安全且避免语法错误：

# 错误写法
# cursor.execute(f"SELECT idu FROM indirect.usr WHERE cn='{signinname}'")
# 正确参数化写法
cursor.execute("SELECT idu FROM indirect.usr WHERE cn=?", (signinname,))

4. 临时缓解：提升系统文件描述符限制 #

修改Ubuntu系统的文件描述符上限，临时缓解资源耗尽问题（核心还是代码修复）：

• 临时生效（当前会话）：

  ulimit -n 65535
  

• 永久生效：编辑/etc/security/limits.conf，添加以下内容后重启系统：

  * soft nofile 65535
  * hard nofile 65535
  

5. 简化JSON处理逻辑 #

减少不必要的序列化反序列化步骤，直接操作字典提升性能：

# 直接生成字典，避免多次JSON转换
output_data = {"SingleValueAttributes": result1}
# 直接处理查询结果到字典
transformed_data = {}
for sql_query in queries:
    cursor.execute(sql_query[0], (idu,))
    rows = cursor.fetchall()
    if rows:
        row_dict = dict(zip([col[0] for col in cursor.description], rows[0]))
        for key, value in row_dict.items():
            transformed_data[key] = value.split(',') if value else []

修正后的完整代码 #

import uvicorn
import pyodbc
import json
from fastapi import FastAPI, HTTPException, Response
from pydantic import BaseModel

app = FastAPI()

# 数据库连接配置
server = 'fasfasf.privatelink.database.windows.net'
database = 'sfafa-af-wus2-dev-idm'
username = 'fafafs'
password = 'affaaaa$#'
port = '1433'

conn_string = (
    f'DRIVER={{ODBC Driver 18 for SQL Server}};'
    f'SERVER={server},{port};'
    f'DATABASE={database};'
    f'UID={username};'
    f'PWD={password};'
    f'MARS_Connection=yes;'
    f'Encrypt=yes;'
    f'Connection Timeout=30'
)

pooling_options = {
    "enabled": True,
    "min_connections": 5,
    "max_connections": 100,
    "max_idle_time": 300,
}

def get_db_connection():
    return pyodbc.connect(conn_string, pooling=True, **pooling_options)

class User(BaseModel):
    signinname: str

@app.post("/user_details")
def read_user_details(user: User):
    signinname = user.signinname
    print(f"Signinname: {signinname}")

    # 上下文管理器自动管理连接和游标
    with get_db_connection() as conn:
        with conn.cursor() as cursor:
            # 参数化查询获取idu
            cursor.execute("SELECT idu FROM indirect.usr WHERE cn=?", (signinname,))
            row = cursor.fetchone()
            if not row:
                raise HTTPException(status_code=404, detail="No matching idu found for the specified cn.")
            idu = row[0]

            # 获取主用户数据
            cursor.execute("SELECT * FROM indirect.usr WHERE idu=?", (idu,))
            rows1 = cursor.fetchall()
            columns = [col[0] for col in cursor.description]
            result1 = [dict(zip(columns, row)) for row in rows1] if rows1 else [dict(zip(columns, [None]*len(columns)))]
            output_data = {"SingleValueAttributes": result1}

            # 批量查询其他关联数据
            queries = [
                "SELECT STRING_AGG(DGID, ',') as DGID FROM indirect.usr_DGID WHERE idu=? GROUP BY idu",
                "SELECT STRING_AGG(telephoneNumber, ',') as telephoneNumber FROM [indirect].[usr_telephoneNumber] WHERE idu=? GROUP BY idu"
            ]

            transformed_data = {}
            for sql_query in queries:
                cursor.execute(sql_query, (idu,))
                rows = cursor.fetchall()
                if rows and rows[0][0]:
                    col_name = cursor.description[0][0]
                    transformed_data[col_name] = rows[0][0].split(',')
                else:
                    col_name = cursor.description[0][0]
                    transformed_data[col_name] = []

            # 合并结果并返回
            output_data.update(transformed_data)
            merged_json = json.dumps(output_data, indent=4, default=str)
            return Response(content=merged_json, media_type="application/json")

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

额外优化建议 #

• 调整连接池max_connections值，不要超过数据库允许的最大连接数，避免压垮数据库。
• 添加数据库连接异常捕获，返回更友好的错误提示。
• 对高频查询的用户数据增加缓存（如Redis），减少数据库请求次数。

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