Psycopg3 Async连接池泄漏/超时问题排查求助

在Flask应用中扩展功能时，使用Psycopg3 Async Connection Pool遇到连接泄漏问题：服务器启动时创建异步连接池，每次请求时从池获取连接检索数据、执行查询后返回结果。初期运行正常，但数次使用后抛出psycopg_pool.PoolTimeout异常，查看池统计信息显示pool_available = 0，requests_queued和requests_errors随重试持续增长，可用连接已耗尽，请求帮忙排查根源。

可复用连接池实现 #

# db.py
import os
from psycopg_pool import AsyncConnectionPool

class DB:
    def __init__(self):
        self.async_pool = None

    def connect_async(self):
        conninfo = (f"dbname={os.environ['DBNAME']} "
                    f"user={os.environ['POSTGRES_USER']} "
                    f"password={os.environ['POSTGRES_PASSWORD']} "
                    f"host={os.environ['POSTGRES_HOST']} "
                    f"port={os.environ['POSTGRES_PORT']}")
        self.async_pool = AsyncConnectionPool(
            conninfo, min_size=4, max_size=10, timeout=15)

    def get_async_pool(self):
        if self.async_pool is None:
            self.connect_async()
        return self.async_pool

# extensions.py
from db import DB

my_db = DB()

查询执行逻辑 #

# search.py
from extensions import my_db

async def keyword_search(query):
    async with my_db.get_async_pool().connection() as aconn:
        async with aconn.cursor() as cur:
            # query = ...  keyword search logic
            await cur.execute(query)
            documents = await cur.fetchall()
            column_names = [desc[0] for desc in cur.description]
    return [dict(zip(column_names, doc)) for doc in documents]

async def semantic_search(query):
    async with my_db.get_async_pool().connection() as aconn:
        async with aconn.cursor() as cur:
            # query = ...  semantic search logic
            await cur.execute(query)
            documents = await cur.fetchall()
            column_names = [desc[0] for desc in cur.description]
    return [dict(zip(column_names, doc)) for doc in documents]

async def full_search(query):
    ks, ss = await asyncio.gather(keyword_search(query), semantic_search(query))
    return ks + ss

Flask异步路由实现 #

from extensions import my_db
from search import full_search

@bp.route('/search', methods=['GET'])
async def search_fn():
    query = request.args.get('query')
    for _ in range(2):
        try:
            results = await full_search(query)
            return jsonify(results)
        except psycopg_pool.PoolTimeout as e:
            stats = my_db.get_async_pool().get_stats()
            app.logger.error({'error': str(e), 'pool_stats': stats})
            
    return jsonify({'error': 'Database connection timeout'}), 500

补充信息 #

• 应用中仅search_fn为异步路由，其余均为同步路由。

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