异步SQLite3查询需等待耗时API调用完成才执行的问题排查

核心问题原因 #

SQLite是单文件数据库，默认写操作会独占整个数据库文件锁，即便用异步驱动aiosqlite也无法绕过这个限制——当/api/call这类长时间运行的请求占用连接时，后续数据库查询会被阻塞，直到锁释放。此外代码还有几处异步配置错误：

• Flask原生不支持异步路由，用async def定义路由却没搭配ASGI服务器，异步代码实际在同步线程池运行，没发挥异步优势。
• 同时混用flask_sqlalchemy的同步实例和手动创建的异步引擎，存在上下文冲突。
• aiosqlite基于线程实现，受SQLite文件锁限制，无法真正实现多并发数据库操作。

修复步骤 #

1. 替换为ASGI服务器 #

Flask默认开发服务器不支持异步路由，需使用uvicorn这类ASGI服务器：

pip install uvicorn

启动命令改为：

uvicorn app:app --host 0.0.0.0 --port 5000

2. 统一异步数据库配置 #

修改database/db.py，移除冗余同步配置，仅保留异步SQLAlchemy设置：

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

# SQLite异步引擎，开启check_same_thread适配SQLite特性
engine = create_async_engine(
    'sqlite+aiosqlite:///main.sqlite3',
    connect_args={"check_same_thread": False},
    pool_recycle=300
)

# 创建异步会话工厂
AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    expire_on_commit=False,
    class_=AsyncSession
)

# 会话获取依赖函数
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

3. 更新数据库操作函数 #

修改database/actions.py，适配新的异步会话获取方式：

from sqlalchemy import select
from database.models import ApiKey
from database.db import get_db

async def get_customer(api_key):
    async for session in get_db():
        result = await session.execute(
            select(ApiKey).where(ApiKey.api_key == api_key)
        )
        key = result.scalars().first()
        return {'id': key.id, 'api_key': key.api_key, 'owner_name': key.owner_name} if key else None

async def get_all_customers():
    async for session in get_db():
        result = await session.execute(select(ApiKey))
        return [{'id': c.id, 'api_key': c.api_key, 'owner_name': c.owner_name} for c in result.scalars().all()]

4. 优化Flask应用配置 #

修改app.py，移除同步SQLAlchemy实例，确保路由真正异步：

from flask import Flask, request, jsonify
from database.actions import get_all_customers, get_customer
import asyncio

app = Flask(__name__)

@app.route('/api/customers', methods=['GET'])
async def customer_get():
    api_key = request.args.get('api-key')
    try:
        if api_key:
            result = await get_customer(api_key)
            return jsonify(result or {})
        else:
            result = await get_all_customers()
            return jsonify(result)
    except Exception as e:
        print(e)
        return jsonify({'error': str(e)}), 500

@app.route('/api/call', methods=['GET'])
async def call():
    api_key = request.args.get('api-key')
    try:
        if api_key:
            # 确保heavy_call是异步函数，同步阻塞操作需用asyncio.to_thread包装
            result = await heavy_call(api_key)
            return jsonify(result or {})
        return jsonify({}), 400
    except Exception as e:
        print(e)
        return jsonify({'error': str(e)}), 500

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=5000)

5. 解决SQLite锁阻塞的终极方案 #

若上述优化仍有阻塞，需针对SQLite锁机制做调整：

• 开启WAL模式，提升读并发性能（读操作不会被写操作阻塞），在数据库初始化时执行：

  async def init_db():
      async with engine.begin() as conn:
          await conn.execute("PRAGMA journal_mode=WAL;")
          await conn.run_sync(ApiKey.metadata.create_all)
  

  应用启动时调用await init_db()。
• 若并发需求较高，直接切换到PostgreSQL/MySQL这类支持多并发的数据库，搭配对应异步驱动（asyncpg/aiomysql）。

关键注意事项 #

• 若heavy_call是同步阻塞操作（如用requests库），必须用asyncio.to_thread包装，避免阻塞事件循环：

  async def heavy_call(api_key):
      return await asyncio.to_thread(sync_heavy_call, api_key)
  

• 异步代码中禁止直接调用同步阻塞IO操作，否则会导致所有请求排队等待。

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