每日一句正能量

不帮别人欺负自己，别人质疑你时，不要急于自我批评，先问问自己内心的判断。
别人的质疑是别人的视角，不等于事实。 在自我批评之前，先回到内心，问问自己：“我认同这个质疑吗？”

1. 背景与问题

在人工智能与数据分析驱动的现代架构中，Python（搭配 FastAPI、Flask 或 Django）成为了构建轻量级数据服务的首选。随着政企客户全面推进信创改造，将 Python 服务的后端数据库从 MySQL 或 PostgreSQL 切换为人大金仓（KingbaseES）成为了刚需。

业务痛点场景：
我们的一个基于 FastAPI 构建的“用户行为画像服务”，在底层数据库切换为金仓 V8R6 后，遇到了严重的性能瓶颈与稳定性危机：

1. 并发阻塞与吞吐断崖：原有的代码使用同步的 psycopg2 驱动直连数据库。当并发请求数超过 50 时，由于 GIL（全局解释器锁）和同步 I/O 的限制，导致大量的 Web 工作线程被阻塞，响应时间从 20ms 飙升至 3000ms。
2. 事务状态破坏 (Transaction Aborted)：与 Java/C# 类似，金仓数据库继承了 PostgreSQL 严苛的事务控制机制。当 Python 代码执行一条存在语法错误或违反唯一约束的 SQL 后，若没有显式捕获异常并执行 conn.rollback()，该连接后续的所有查询都会报出 current transaction is aborted，最终导致连接池被“脏连接”彻底污染，服务雪崩。

2. 环境与数据

为了对比同步与异步访问在金仓上的真实表现，我们搭建了以下标准测试环境：

• 运行环境: Python 3.10, FastAPI 0.103
• 目标数据库: 人大金仓 KingbaseES V8R6
• 底层驱动: psycopg (即原生支持异步的 psycopg3，版本 3.1.10) 和 psycopg_pool
• 测试数据表 (user_action_log):

初始化 SQL (schema.sql):

CREATE TABLE user_action_log (
    log_id BIGSERIAL PRIMARY KEY,
    user_id VARCHAR(50) NOT NULL,
    action_type VARCHAR(20) NOT NULL,
    action_data JSONB,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    CONSTRAINT uk_user_action UNIQUE(user_id, action_type)
);

3. 复现过程

3.1 传统的同步阻塞式代码 (存在连接池污染隐患)

许多开发者习惯于使用传统的同步方式，且对异常边界的处理十分粗糙：

# 存在风险的同步代码片段
import psycopg

# 假设全局只维持了一个简单的连接
conn_str = "host=192.168.1.100 port=5432 dbname=testdb user=system password=secret"

def sync_save_action_legacy(user_id, action_type):
    # 问题1：每次请求建立新连接（或者没有使用池），耗时极大
    # 问题2：同步阻塞，严重降低并发吞吐
    with psycopg.connect(conn_str) as conn:
        with conn.cursor() as cur:
            try:
                cur.execute(
                    "INSERT INTO user_action_log (user_id, action_type) VALUES (%s, %s)", 
                    (user_id, action_type)
                )
                conn.commit()
            except Exception as e:
                print(f"Error: {e}")
                # 致命缺陷：如果报错（如唯一约束冲突），这里直接打印错误。
                # 但并没有执行 conn.rollback()，导致当前连接处于 aborted 状态！

复现结果： 当发生主键冲突（UniqueViolation）时，当前请求报错。如果在带有连接池的场景下，这个连接被归还，下一个请求拿到该连接执行 SELECT，金仓直接拒绝服务并抛出：InFailedSqlTransaction: current transaction is aborted, commands ignored until end of transaction block。

4. 方案实施

为了榨干 Python 与金仓交互的性能，并保证系统的绝对稳定，我们全面采用 psycopg3。该版本采用了全新的架构，原生支持 Python 的 asyncio 异步生态，并提供了完善的高并发连接池组件。

我们需要从异步连接池生命周期管理、异步 SQL 执行以及严格的事务回滚边界三个层面进行重构。

4.1 异步连接池的生命周期注入 (FastAPI 示例)

在应用启动时，初始化异步连接池；在应用关闭时，优雅释放。

main.py 连接池配置:

from fastapi import FastAPI, HTTPException
from psycopg_pool import AsyncConnectionPool
from contextlib import asynccontextmanager

# 金仓数据库连接字符串
DSN = "host=192.168.1.100 port=5432 dbname=testdb user=system password=secret"

# 异步连接池实例
pool: AsyncConnectionPool

@asynccontextmanager
async def lifespan(app: FastAPI):
    global pool
    # 初始化异步连接池 (最小10，最大50连接)
    pool = AsyncConnectionPool(
        conninfo=DSN,
        min_size=10,
        max_size=50,
        timeout=30.0,
        kwargs={"autocommit": False} # 保持默认的事务控制
    )
    await pool.open()
    yield
    await pool.close()

app = FastAPI(lifespan=lifespan)

4.2 强事务边界的异步业务处理代码

在处理业务逻辑时，使用 async with 借出连接，并使用严格的 try...except 块。对于金仓抛出的异常，必须精准捕获，并在 except 块中立即触发 await conn.rollback()。

main.py 路由与业务代码:

import psycopg
from psycopg.errors import UniqueViolation, InFailedSqlTransaction
from pydantic import BaseModel

class ActionRequest(BaseModel):
    user_id: str
    action_type: str

@app.post("/api/v1/action/async")
async def save_action_async(req: ActionRequest):
    # 从异步连接池中获取连接
    async with pool.connection() as conn:
        # 开启游标
        async with conn.cursor() as cur:
            try:
                # 异步执行 SQL，避免阻塞当前线程 (Event Loop 继续处理其他请求)
                await cur.execute(
                    "INSERT INTO user_action_log (user_id, action_type) VALUES (%s, %s)",
                    (req.user_id, req.action_type)
                )
                await conn.commit()
                return {"status": "success"}
                
            except UniqueViolation as e:
                # 核心防线：必须显式回滚，清除 Aborted 状态
                await conn.rollback()
                raise HTTPException(status_code=409, detail="Action already exists")
                
            except psycopg.Error as e:
                # 兜底防线：任何数据库层面的报错，先回滚，再抛出
                await conn.rollback()
                raise HTTPException(status_code=500, detail="Database internal error")

5. 结果对比

我们将同步版本（基于 psycopg2 直连）与优化后的异步版本（FastAPI + psycopg3 AsyncConnectionPool）进行了极限施压，测试客户端发起 1000 并发请求。

5.1 架构并发模型流转对比

下图展示了同步模型导致线程阻塞与异步非阻塞模型（Event Loop）的工作流差异：

执行模型对比图

5.2 压测吞吐量与耗时指标

在 1000 并发的场景下，异步机制将吞吐量提升了超过一个数量级，同时完美规避了连接池污染问题。

性能指标对比图

5.3 核心优化数据表

6. 风险与复盘

在 Python 环境下适配金仓数据库，有三个极其容易被忽略的“隐形地雷”，架构师必须在上线前进行排查：

1. psycopg2 vs psycopg3 的深渊：
   市面上大量老旧教程还在推荐 psycopg2。但在配合最新的金仓版本进行高并发异步编程时，psycopg2 依赖于第三方的 aiopg，不仅配置繁琐，而且内存泄漏风险高。强烈建议在信创改造中直接使用 psycopg (v3)，它是针对 Python 3 的原生异步重新设计的，性能和稳定性有着质的飞跃。
2. PgBouncer 与连接池的冲突：
   如果在金仓服务端部署了 PgBouncer 等中间件来进行连接复用（Transaction mode），那么在 Python 的 AsyncConnectionPool 配置中，应当尽量缩短连接的生命周期，或者依赖服务端的池化能力，避免发生连接保持探测（Keepalive）冲突导致的伪断连。
3. JSONB 数据的序列化反序列化：
   向金仓插入 JSONB 字段时，Python 中的 dict 不能直接拼接。在 psycopg3 中，不需要手动 json.dumps()，你可以利用 from psycopg.types.json import Jsonb，在执行参数中传入 Jsonb({"key": "value"})，驱动会在底层自动将其转化为金仓原生的高性能 JSONB 二进制协议。

转载自：https://blog.csdn.net/u014727709/article/details/163528994
欢迎 👍点赞✍评论⭐收藏，欢迎指正