FastAPI多线程场景下SQLite3数据库单次连接写入方案咨询

核心问题拆解 #

SQLite3本身不支持多线程同时写入，默认连接模式下多线程共享单个连接会导致锁冲突甚至数据损坏；反复开关连接确实会带来额外IO开销，影响性能。

可行优化方案 #

1. 线程本地存储复用连接 #

每个线程维护一个独立的数据库连接，避免反复开关，同时规避多线程共享连接的问题。
示例代码：

import sqlite3
from threading import local

# 线程本地存储，每个线程独有一个连接实例
thread_local = local()

def get_db_connection():
    if not hasattr(thread_local, "conn"):
        # 配置线程安全模式，开启WAL提升并发性能
        thread_local.conn = sqlite3.connect("your_db.db", check_same_thread=False)
        thread_local.conn.execute("PRAGMA journal_mode=WAL;")
    return thread_local.conn

# FastAPI接口示例
@app.post("/write-data")
def write_data(data: dict):
    conn = get_db_connection()
    cursor = conn.cursor()
    try:
        cursor.execute("INSERT INTO your_table (col1, col2) VALUES (?, ?)", (data["col1"], data["col2"]))
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise e
    finally:
        cursor.close()  # 仅关闭游标，连接留作当前线程后续请求复用

• 注意：check_same_thread=False配合thread_local使用，能保证每个线程只操作自己的连接；WAL模式让SQLite支持多读单写的并发场景，比默认模式性能提升明显。

2. 使用连接池管理连接 #

借助SQLAlchemy实现连接池，自动管理连接的复用与分配，适合高并发场景。
示例代码：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from fastapi import Depends

# 创建带连接池的引擎，可根据并发量调整池大小
engine = create_engine(
    "sqlite:///your_db.db",
    connect_args={"check_same_thread": False},
    pool_size=10,
    max_overflow=20
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 依赖项：每个请求获取一个会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# FastAPI接口示例
@app.post("/write-data")
def write_data(data: dict, db = Depends(get_db)):
    db.execute("INSERT INTO your_table (col1, col2) VALUES (:col1, :col2)", data)
    db.commit()

• 优势：连接池自动处理连接的创建、复用和回收，无需手动管理线程本地存储；SQLAlchemy的ORM还能简化复杂数据操作。

3. 替代方案：切换高并发数据库 #

如果业务并发写入量较高，SQLite的单写限制会成为瓶颈，可切换到PostgreSQL、MySQL等客户端-服务器型数据库，原生支持高并发多线程写入，适合大规模场景。

关键注意事项 #

• 无论用哪种方案，必须开启SQLite的WAL模式，这是提升并发性能的基础；
• 多线程下绝对不能共享单个数据库连接，否则会引发线程安全问题；
• 使用连接池时，需根据实际并发量调整池大小，避免连接过多导致资源浪费。

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