Psycopg2 RealDictCursor返回非确定性结果的原因排查

问题详情 #

使用psycopg2结合RealDictCursor查询PostgreSQL时，重复执行同一查询偶尔返回异常结果，该现象出现在每秒多请求的测试环境中。

连接与游标初始化代码 #

import psycopg2
from psycopg2.extras import RealDictCursor 

self.conn = psycopg2.connect(
            host="db",
            database=os.environ["DB_NAME"],
            user=os.environ["DB_USER"],
            password=os.environ["DOCKERDBPASS"],
        )
self.cur = self.conn.cursor(cursor_factory=RealDictCursor)

表结构 #

CREATE TABLE IF NOT EXISTS ct_session(
    id_nb SERIAL PRIMARY KEY,
    bk INT NOT NULL,
    status VARCHAR(10) NOT NULL,
    created_on TIMESTAMP DEFAULT NOW(),
    closed_on TIMESTAMP NULL,
    CONSTRAINT fk_bk FOREIGN KEY(bk) REFERENCES app_user(django_id)
);

查询代码 #

query = """
SELECT
    id_nb
FROM ct_session
WHERE status = 'open'
;
"""

self.cur.execute(query)
result = self.cur.fetchone()
print("====DEBUG====", result)
if result:
    return result["id_nb"]
else:
    return None

异常现象 #

多数情况下结果符合预期：

====DEBUG==== RealDictRow([('id_nb', 1)])

偶尔出现异常结果，例如：

====DEBUG==== RealDictRow([('status', 1)])

或：

====DEBUG==== RealDictRow([(<class 'psycopg2.extras.RealDictRow'>, ['id', 'user', 'status']), ('id', 1)])

原因分析 #

这是并发场景下共享非线程安全的数据库连接/游标导致的：

• psycopg2的connection和cursor对象都不是线程安全的，多个请求/线程同时使用同一个游标执行查询时，会互相干扰对方的查询状态。
• 当一个请求的execute还未完成，另一个请求复用同一个游标执行其他查询（甚至是其他表的查询），就会导致结果集的字段、数据被混乱覆盖，出现你看到的异常字段或奇怪的元组结构。

解决方案 #

1. 每个请求/线程独立创建连接和游标
   不要在全局类实例中共享连接或游标，每次处理请求时创建新的连接和游标，用完后关闭：

   def get_open_session_id():
       conn = psycopg2.connect(host="db", database=os.environ["DB_NAME"], user=os.environ["DB_USER"], password=os.environ["DOCKERDBPASS"])
       try:
           cur = conn.cursor(cursor_factory=RealDictCursor)
           cur.execute(query)
           result = cur.fetchone()
           return result["id_nb"] if result else None
       finally:
           cur.close()
           conn.close()
   

2. 使用连接池管理连接
   生产环境推荐用连接池（如psycopg2.pool.SimpleConnectionPool），确保每个请求获取独立的连接实例，避免共享：

   # 初始化连接池（全局只执行一次）
   pool = psycopg2.pool.SimpleConnectionPool(
       minconn=1,
       maxconn=10,
       host="db",
       database=os.environ["DB_NAME"],
       user=os.environ["DB_USER"],
       password=os.environ["DOCKERDBPASS"],
   )
   
   def get_open_session_id():
       conn = pool.getconn()
       try:
           cur = conn.cursor(cursor_factory=RealDictCursor)
           cur.execute(query)
           result = cur.fetchone()
           return result["id_nb"] if result else None
       finally:
           cur.close()
           pool.putconn(conn)
   

3. 确保游标串行使用
   如果必须共享连接，要加锁确保同一时间只有一个请求使用游标，但这种方式会降低并发性能，不推荐。

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