并发写入PostgreSQL时表数据丢失及序列化失败问题求助

1. 为什么SERIALIZABLE隔离级别未阻止数据丢失？ #

你的代码存在三个核心问题导致隔离级别未生效，进而引发数据丢失：

• 隔离级别设置错误：session.connection(execution_options={"isolation_level": "SERIALIZABLE"}) 仅在获取连接时临时设置，并未确保整个事务周期都使用该级别。SQLAlchemy中需在会话创建时指定隔离级别，才能覆盖全事务流程。
• 事务处理逻辑缺陷：并发冲突触发SerializationFailure或DeadlockDetected时，代码未执行事务回滚就直接抛出异常，部分未提交操作可能处于不确定状态；同时无重试机制，导致整个chunk的数据直接丢失而非重新写入。
• UPSERT逻辑隐患：手动构造update_val可能存在字段遗漏或错误，若并发时多个实例同时更新同一行，会导致数据被错误覆盖。

2. 如何检测不同脚本实例的并发事务？ #

可通过PostgreSQL系统视图监控并发事务：

• 执行以下SQL查询目标表的活跃事务：

SELECT pid, query_start, state, application_name, query
FROM pg_stat_activity
WHERE datname = '你的数据库名'
  AND relation = '你的目标表名'::regclass
  AND state IN ('active', 'idle in transaction');

• 给每个脚本实例设置唯一application_name（创建引擎时指定：create_engine(..., connect_args={"application_name": "instance_1"})），方便在查询结果中区分不同实例的事务。
• 该方式仅用于监控，不能作为避免并发的核心手段，仍需依赖事务隔离和重试逻辑。

3. 如何实现事务的等待/重试？ #

针对PostgreSQL的并发异常，需实现事务级别的重试机制，同时优化UPSERT逻辑减少冲突概率，修改后的代码示例如下：

from sqlalchemy import Session
from sqlalchemy.dialects.postgresql import insert as insert_postgresql
import psycopg2
from psycopg2.errors import SerializationFailure, DeadlockDetected

_CACHE_SIZE = 5000
MAX_RETRIES = 3  # 可根据实际情况调整

def upsert_postgresql(self):
    retry_count = 0
    while retry_count < MAX_RETRIES:
        # 创建会话时直接指定隔离级别，确保整个事务生效
        with Session(self.eng, execution_options={"isolation_level": "SERIALIZABLE"}) as session:
            try:
                # 批量构造UPSERT语句，减少数据库交互次数，降低锁竞争
                stmt = insert_postgresql(self.sqla_table).values(self.chunk)
                # 使用excluded对象获取冲突行的字段值，避免手动构造的错误
                update_val = {col: stmt.excluded[col] for col in self.col_all_minus}
                stmt = stmt.on_conflict_do_update(
                    index_elements=self.idx_label,
                    set_=update_val
                )
                session.execute(stmt)
                session.commit()
                return  # 成功写入，退出重试循环
            except (SerializationFailure, DeadlockDetected):
                # 并发冲突，回滚事务后重试
                session.rollback()
                retry_count += 1
            except Exception as e:
                session.rollback()
                raise ValueError(f"写入失败: {str(e)}")
    # 重试次数耗尽仍失败
    raise ValueError(f"已重试{MAX_RETRIES}次，写入仍失败")

核心优化点： #

• 批量UPSERT：替代原循环逐个执行语句的方式，减少锁持有时间和冲突概率，提升写入性能。
• 正确设置隔离级别：在创建Session时通过execution_options指定，确保整个事务周期都使用SERIALIZABLE级别。
• 针对性重试：仅捕获PostgreSQL的并发异常，重试前强制回滚事务，避免会话处于错误状态。
• 使用stmt.excluded：确保冲突时更新为当前批次的字段值，避免手动构造update_val的逻辑错误。

额外优化建议 #

• 确认self.idx_label对应的是表的唯一索引，这是ON CONFLICT DO UPDATE生效的必要条件。
• 若业务允许，可将隔离级别降低至REPEATABLE READ，该级别冲突概率更低，性能开销更小，结合重试机制可满足多数一致性需求。
• 监控数据库锁状态，通过pg_locks视图排查长期持有的锁：

SELECT * FROM pg_locks WHERE relation = '你的目标表名'::regclass;

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