不少朋友看到Python 链接PG数据库这个标题第一反应是这不就是两行代码的事。我在前面几年也是这么想的直到线上服务在凌晨因为连接数打满被拖垮直到批量导入十万条数据跑了四十分钟直到某次排查发现报错日志里那个模糊的connection already closed其实是连接池里躺着一批被数据库单方面踢掉的僵尸连接。Python 连接 PostgreSQL 这件事写通和写好之间隔着的是一整条生产环境的鸿沟——驱动怎么选、连接池开多大、事务边界划在哪里、超时怎么设、异常怎么兜每一个点都能单独写一篇。这篇内容我把这几年从原型到生产踩过的路重新走了一遍把能跑到稳跑之间那些文档里不写、但你必须知道的细节都摊开来说。适合刚接触 Python 操作关系型数据库的新手按步骤复现也适合已经在用但偶尔被连接池和事务坑到的同学对照检查。标题里写的是链接实际大家在说的是连接下文统一按连接来说。1. 先把这件事拆开Python 连 PG 到底在连什么很多人对连接数据库的理解停留在提供一个账号密码就能查数据这种认知在写脚本时没有问题一旦进入服务化阶段就会处处碰壁。要把它做好得先知道这条链路上到底有几个环节、每个环节各自负责什么、出问题时会以什么形式暴露出来。1.1 一条 SQL 从 Python 到 PG 走过的四段路从你写下一行cursor.execute(select 1)开始到结果集回到你的 Python 进程中间其实穿了四道门。第一道是驱动层。你的 Python 代码并不直接和数据库进程说话中间必须有一个实现了数据库通信协议的库负责把 Python 对象序列化成 PG 认识的线协议格式再把返回的字节流反序列化成 Python 对象。这个角色在 Python 生态里主要就是 psycopg2 和 psycopg3 两位它们都是 C 扩展实现的性能接近也有纯 Python 的替代实现但吞吐差距明显生产环境基本不考虑。第二道是连接层也就是 TCP 握手 认证 会话初始化。这一步的成本经常被低估一次完整的连接建立包含 DNS 解析、TCP 三次握手、可能的 TLS 协商、PG 的认证往返口令认证至少一个 RTT、以及连接建立后驱动层发出的一系列SET语句客户端编码、时区、只读标记等。实测在同一个机房内网一次连接的建立耗时在 3 到 15 毫秒之间波动跨可用区能到 30 毫秒以上。这意味着如果你的业务是每次请求新建一个连接光是连接开销就能吃掉相当一部分响应时间。第三道是会话层连接建立之后就进入会话状态。这里有个新手容易忽略的事实PG 的连接是有状态的。你在会话里SET过的参数、开启过的事务、创建的临时表、拿到的 advisory lock都绑定在这条连接上。一旦这条连接被连接池复用给别的请求这些状态就会跟着流过去。我见过最典型的事故是某个接口为了调优在会话里设了statement_timeout 200结果这条连接被复用后另一个跑批任务的所有语句全部超时被杀。第四道是服务端PG 为每条连接 fork 一个独立的进程不是线程这个进程会占用一定的内存最小大概几 MB随 work_mem 等参数增长。这就决定了 PG 的连接数不是可以随便开的几百条活跃连接就可能把内存吃紧而且进程切换带来的上下文开销也会拖慢整体吞吐。理解这四段路的价值在于你的每一个优化决策都对应着其中某一段。用连接池是在优化第二段调work_mem是在优化第四段把关掉自动提交是在优化第三段——后面几章的内容基本都是围绕这四个层次展开的。1.2 三种主流接入姿势的分层与适用边界实际项目里Python 连 PG 不一定都走同一条路常见的有三种姿势它们不是互斥的很多项目是混着用的。裸驱动直连直接import psycopg2自己管连接、管游标、管事务。优点是零抽象、行为完全可控、排查问题链路最短缺点是所有东西都要自己写尤其事务和异常处理容易漏。适合一次性脚本、数据迁移工具、定时任务、以及需要精细控制 COPY 批量导入的场景。驱动 连接池在裸驱动外面套一层池请求来时借、用完还。这是绝大多数 Web 服务和后端 API 的标准做法。池的实现可以是驱动自带的简易池也可以是更成熟的三方池。优点是把连接建立成本摊薄了把并发控制住了缺点是池参数配错比不用池还危险这个后面单独讲。ORM / 查询构建器走 SQLAlchemy、Django ORM 这类框架SQL 由框架生成。优点是模型定义清晰、迁移方便、团队协作成本低缺点是生成的 SQL 可能不是你想要的复杂查询要么写不出要么性能差出问题时还得回头读框架生成的语句。我的经验是业务 CRUD 用 ORM 省事报表和批处理老老实实写原生 SQL两者在同一个项目里共存完全正常。接入方式典型场景连接建立成本可控性上手难度我的建议裸驱动直连脚本、迁移、批处理每次都有最高低一次性任务首选驱动 连接池Web 服务、常驻进程摊薄到接近零高中后端服务标配ORM 框架业务 CRUD、多表关联由框架池管理中中高业务层用分析层绕开直连 COPY 批量大批量导入导出一次最高中数据类任务必用1.3 一个判断标准你的场景需不需要池不是所有场景都需要连接池这一点很多教程没说清楚导致有人给一个跑五分钟就退出的脚本也套上了池纯属增加复杂度。我给一个简单的判断口径进程生命周期内数据库操作的次数少于 10 次且对延迟不敏感——直连就行用完就关。进程常驻且会持续接收外部请求——必须用池没有任何商量余地。进程常驻但是单线程跑批数据库操作很密集——可以用池但池大小设 1 到 2 就够重点是复用而不是并发。有多个线程或协程并发访问——必须用池且池大小要与并发度匹配这是最容易出事的一类。还有一个反向的判断如果你的服务是 Serverless 形态每个请求一个冷启动实例那么本地连接池的意义会被大幅削弱因为实例本身就不常驻。这时候更合适的是在数据库前面加一层连接代理让代理去承接真实连接应用侧只维持轻量连接。这个思路后面在池参数计算那一节会再提。2. 环境准备驱动选型与依赖安装的那些细节环境这一步看着最没技术含量但我统计过自己遇到过的连不上问题超过一半出在这一步——装错了包、装错了版本、装到了全局环境、缺了系统库。所以这一章值得单独展开。2.1 psycopg2 与 psycopg2-binary以及要不要上 psycopg3先说最容易踩的坑PyPI 上有两个名字极度相似的包psycopg2和psycopg2-binary。psycopg2是源码包安装时会现场编译需要本机有 C 编译器、Python 头文件以及 PostgreSQL 的客户端开发库提供 libpq。好处是它链接的是你系统上的 libpq能拿到最新的协议特性和安全补丁坏处是编译过程容易失败尤其在精简镜像、Alpine 基础镜像、没有 root 权限的环境里报错信息往往只有一句command gcc failed。psycopg2-binary是预编译好的 wheel安装即用不需要编译器和系统库。适合本地开发、CI 环境、快速验证。但它打包的 libpq 是构建时的那一版不会随系统更新官方文档也明确建议不要在生产环境长期依赖它。至于 psycopg3包名psycopg是同一批维护者的新一代实现。它有几点实质性改进原生支持 async/await、支持服务端参数绑定不用在客户端拼 SQL 字符串、支持流水线模式pipeline mode一次网络往返提交多条语句、连接和游标都支持上下文管理器。新项目我倾向于直接上 psycopg3老项目迁移则要看有没有依赖 psycopg2 特有的行为比如某些扩展类型的适配器注册方式不同。包名安装方式是否需编译生产建议典型适用psycopg2源码编译是推荐生产环境、需要跟随系统 libpqpsycopg2-binarywheel否不建议长期用本地开发、快速验证、CIpsycopgwheel否推荐新项目、需要 async、需要流水线安装命令大致是这样注意先确认自己在虚拟环境里python -m venv .venv source .venv/bin/activate pip install psycopg[binary] # psycopg3带二进制 libpq # 或者 pip install psycopg2-binary # 开发阶段图省事注意在容器镜像里装psycopg2之前基础镜像需要先装libpq-dev和build-essentialDebian 系装完再清理掉这些构建依赖否则镜像会白白膨胀几百 MB。多阶段构建是标准做法。2.2 虚拟环境与版本锁定的必要性我见过太多我本地能跑服务器上连不上的案例追下去八成是版本不一致。数据库驱动这种东西小版本之间的行为差异是实打实存在的比如某个版本改变了时间类型返回值的时区处理方式某个版本调整了连接失败时的异常类型。这些变化不会让你的代码报语法错而是让它在某个边界条件下悄悄出错。所以我的做法固定为三步虚拟环境隔离、依赖清单锁定、镜像与本地使用同一份清单。依赖清单里把主版本号卡住用~或者精确版本都行关键是不要留一个裸的包名在里面。# requirements.txt 片段 psycopg[binary]~3.1另外Windows 上开发、Linux 上部署的组合也值得提前在容器里跑一遍。我遇到过本地开发一切正常、部署到 Linux 后因为系统时区数据缺失导致时间字段全部偏移八小时的案例排查了两个小时才定位到是基础镜像没装 tzdata。2.3 连接参数逐个拆解哪些必须给哪些别乱给先把一个完整的连接串摆出来然后逐个说。import psycopg conn psycopg.connect( host10.0.0.12, port5432, dbnameappdb, userapp_user, password******, connect_timeout5, application_nameorder-service, options-c statement_timeout30000 -c idle_in_transaction_session_timeout60000, keepalives1, keepalives_idle30, keepalives_interval10, keepalives_count3, )connect_timeout是最容易被忽略的一个。它的默认值是无限等待这意味着如果数据库主机网络不可达比如防火墙静默丢包你的进程会卡在那里很久表现出来就是服务整体无响应。给它一个 3 到 5 秒的值让失败尽快暴露。application_name强烈建议一定要设。它会在pg_stat_activity里显示出来排查哪条连接是谁开的、占着锁不放手的时候这个字段能帮你省下大量时间。设成服务名加实例标识比如order-service-pod3。options可以传会话级参数。statement_timeout防止单条语句跑飞idle_in_transaction_session_timeout防止事务开着不结束白占连接——这两个参数我基本上每个服务都会设置属于最低成本的保险。keepalives 系列用于识别被中间设备静默断开的连接。云环境里的负载均衡器、NAT 网关普遍有空闲连接回收策略短则 5 分钟长则 1 小时。如果不同时设置 TCP keepalive 和应用层探活你会拿到一批看起来还活着、实际早就断了的连接等到真正要写入的时候才报错。提示密码不要硬编码在代码里。用环境变量、配置文件挂载或者专门的密钥管理服务具体选哪种看你的部署环境原则只有一条——密码不进入版本库、不进入日志、不进入错误堆栈。3. 从一次连接到稳定读写核心代码怎么写环境齐了接下来是代码。这一章我按先能跑通再跑稳的顺序来每一段代码都说明它为什么这么写。3.1 最小可用的连接与查询先看一个能跑的最小版本import psycopg with psycopg.connect( host10.0.0.12, port5432, dbnameappdb, userapp_user, password******, connect_timeout5, ) as conn: with conn.cursor() as cur: cur.execute(select id, name from users where status %s limit 10, (active,)) rows cur.fetchall() for r in rows: print(r)这段代码里有两个with含义完全不同很多人搞混。外层的with psycopg.connect(...)是连接级上下文正常退出时它负责提交事务并关闭连接异常退出时回滚并关闭。内层的with conn.cursor()只是游标级上下文退出时关闭游标不涉及事务提交。关键点在于 psycopg 默认不自动提交。也就是说你执行完一条insert如果不 commit在连接关闭时会被回滚。新手最常见的数据插进去了怎么又没了根因就在这里。想让它自动提交有两种写法连接时传autocommitTrue或者在执行后显式调用conn.commit()。我的习惯是脚本类任务用autocommitTrue服务类用显式事务控制。原因是脚本往往是一批独立操作逐条提交反而更符合语义服务里通常一组操作要么全成要么全败需要清晰的事务边界。3.2 参数化查询不只是防注入也是性能问题看到有人在代码里写fselect * from users where name {name}的时候我都会立刻拦下来。这不仅是因为 SQL 注入——虽然那已经足够致命——还因为拼字符串会让数据库完全无法复用执行计划。PG 接收到一条 SQL 之后会做解析、重写、生成执行计划这个过程的成本不低。对于结构相同、只有参数不同的语句PG 可以缓存执行计划直接复用。但你每次拼一个不同的字符串进去在它眼里就是一条全新的 SQL缓存全部失效解析开销翻倍。正确写法是参数化# 正确参数由驱动负责转义与绑定 cur.execute(select * from users where name %s and status %s, (name, status)) # 批量用 executemany或者更好的是 execute_values / COPY cur.executemany( insert into events (id, kind, created_at) values (%s, %s, %s), [(1, click, now), (2, view, now)], )注意参数占位符是%s不管字段是字符串、整数还是日期统一用%s由驱动根据 Python 对象的类型决定怎么编码。不要写成%d或?那是别的数据库驱动的写法混用会直接抛异常。还有一个细节不要用%s去拼表名或字段名。参数化只对值生效对标识符无效。如果你的表名是动态的必须做白名单校验只允许预定义的一组名字通过绝不能把用户输入直接当作标识符。3.3 事务边界三个必须说清楚的坑坑一未提交事务导致锁堆积。事务一旦开始它持有的行锁、表锁就会一直存在直到提交或回滚。如果代码在事务中间做了一件慢事比如调外部接口、写文件、等消息队列这段时间数据库侧的锁就一直在其他请求全被堵住。这就是为什么一定要设idle_in_transaction_session_timeout——它会把长时间挂在事务里不干活的连接直接杀掉虽然看起来粗暴但比拖垮整个库强。坑二自动提交与事务混用。有人把连接设成autocommitTrue然后在代码里又调conn.commit()这时候某些驱动版本会抛异常或者静默忽略行为不一致。要么全程自动提交要么全程显式控制不要混。坑三异常路径没回滚。一段事务代码里如果抛了异常却被捕获吞掉了事务会继续持有后续操作全在这个已损坏的事务里。稳妥的写法是把事务控制交给上下文管理器或者用明确的 try/except 里回滚再抛出。try: with conn.transaction(): with conn.cursor() as cur: cur.execute(update accounts set balance balance - %s where id %s, (amt, from_id)) cur.execute(update accounts set balance balance %s where id %s, (amt, to_id)) except Exception: # 上下文管理器已回滚这里只做日志与上抛 logger.exception(transfer failed) raiseconn.transaction()会创建一个保存点级别的子事务正常退出提交、异常退出回滚这样能把事务边界限制在一小段代码内比手动 commit/rollback 可靠得多。3.4 批量写入executemany 不是终点批量写是性能差异最悬殊的场景。我做过一组对比同一台机器、同一张表、十万条记录写入方式耗时相对值适用量级备注逐条 execute commit100x千级以下千万别这么干executemany12x万级内部仍是逐条发送execute_valuespsycopg22x十万级一条多值 INSERTpsycopg3 executemany 批模式2x十万级需设置批大小COPY1x百万级以上最快但错误处理粗糙结论很直接数据量上了万级就不要再逐条插。psycopg3 的executemany支持returning和批处理可以在连接上设置cursor.executemany(..., returningFalse)配合conn.pipeline()使用能显著减少网络往返。psycopg2 则用psycopg2.extras.execute_values。再往上到百万级直接上 COPYimport psycopg with psycopg.connect(dsn) as conn, conn.cursor() as cur: with cur.copy(copy events (id, kind, created_at) from stdin with (format csv)) as copy: for row in iter_rows(): copy.write_row(row) # psycopg3 的写法自动处理转义 conn.commit()COPY 的代价是它的错误处理不够友好一旦某一行数据格式不对整批都会失败且错误信息里给的行号有时候和你的源数据对不上。我的做法是先把数据打到临时表create temp table ... (like target including defaults)COPY 进临时表然后用一条insert ... select ... where not exists做校验和合并。这样既拿到了 COPY 的速度又能用 SQL 做数据清洗和冲突处理。注意COPY 有事务语义如果你在 COPY 进行中异常退出整批数据都不会落库。另外 COPY 期间表上会持有锁大表导入尽量选低峰期或者用分区表按批灌入。4. 连接池配对了是加速器配错了是定时炸弹到这一章基本进入生产环境的核心议题了。我在生产上遇到过的问题一大半和池有关。4.1 为什么每次请求新建连接是必然失败的先算一笔账。假设你的服务每秒处理 100 个请求每个请求一次数据库操作每次连接建立耗时 8 毫秒。如果每条请求都新建连接光连接建立就要消耗 0.8 个 CPU 秒每秒这还没算服务端 fork 进程的开销。更严重的是并发峰值100 QPS 如果不做任何限制全部同时打进来数据库侧会瞬间出现上百个连接进程内存和上下文切换一起上去响应变慢然后请求堆积然后更多连接被创建——正反馈崩溃。连接池的本质是用有限的连接服务无限的请求它把连接数这个变量从跟随请求量变成了一个可控的固定值。代价是请求可能要排队等连接但排队是可控的崩溃不是。4.2 池参数的估算方法与一个可复现的例子池里最关键的三个参数是最小连接数、最大连接数、空闲连接回收时间。市面上流传的估算公式很多我用的思路是按并发度倒推而不是按 QPS。先确定两个量单实例的并发工作线程数比如 Web 服务开了 8 个 worker每个 worker 同步处理一个请求那就最多 8 个并发数据库操作和单个操作的平均持有时长占比比如一次请求里数据库操作占 20 毫秒整个请求 100 毫秒那占用率是 20%。那么理论池大小 ≈ 并发线程数 × 占用率 × 安全系数。8 × 0.2 × 2 3.2取 4 到 8 都合理。这个结果往往比大家直觉上设的 50、100 小得多。再用全局视角校验一遍数据库侧max_connections假设是 200留 20 给运维和备份剩 180 给应用。应用部署了 6 个实例每实例池上限 20合计 120安全余量充足。如果每实例设 40合计 240直接超了——这就是常见的踩坑方式每个实例单独看都还好合起来把数据库撑爆。参数建议起点调整依据设太大的后果设太小的后果池最小连接数2 至 4稳态并发空闲连接白占内存突发时集中建连池最大连接数并发数×2压测观察等待时间打爆数据库连接上限请求排队、延迟升高空闲回收时间300 秒业务空闲周期频繁建连大量僵尸连接获取连接超时3 至 5 秒可接受排队时长故障时请求堆积轻微抖动就报错提示获取连接超时必须设而且要比你的接口超时小。不设的话池满了以后请求会无限等待表现出来是线程全部挂起比直接报错难排查一百倍。4.3 池化之后仍然断连探活与重连怎么做上了池不代表连接不会断。数据库主备切换、网络设备重启、服务端主动终止空闲连接idle_session_timeout、运维手动 kill——任何一种都会让池里出现失效连接。如果池不做探活你借到的第一条连接可能就是死的。psycopg3 的连接池提供了check参数借出连接时会做一次轻量检查失效则重建。psycopg2 自带的 pool 则没有这个能力需要自己包一层借出后先执行select 1捕获连接类异常就丢弃重取最多重试两次。def get_conn(pool, retries2): for i in range(retries 1): conn pool.getconn() try: with conn.cursor() as cur: cur.execute(select 1) return conn except psycopg.OperationalError: pool.putconn(conn, closeTrue) # 明确丢弃不要放回池里 if i retries: raise这里有个细节必须强调失效连接一定要用 close 标记归还。如果直接 putconn 放回池里下次借到它的线程还得再失败一次形成一个恶性的循环。另外重试要区分错误类型。连接类错误网络中断、连接已关闭可以重试业务类错误唯一键冲突、字段超长重试没有意义只会放大问题。区分方式是按驱动抛出的异常类型判断OperationalError通常可重试IntegrityError、ProgrammingError不要重试。4.4 一个容易被忽略的组合问题连接代理与预编译语句如果数据库前面加了一层连接代理用来承接大量短连接、做连接复用那么代理的工作模式会直接影响驱动的行为。代理如果工作在会话模式一条客户端连接对应一条服务端连接直到断开那基本没有兼容问题。如果工作在事务模式每条事务结束后客户端连接会被重新绑定到别的服务端连接上此时任何会话级状态都会失效——包括预编译语句。psycopg3 默认会在同一会话中缓存并使用服务端预编译语句遇到事务模式的代理就会报出prepared statement does not exist之类的错误。解决办法是在连接参数里关掉自动准备conn psycopg.connect(dsn, prepare_thresholdNone)或者显式设置为 0。这个坑我踩过一次报错信息完全不指向连接代理当时排查了很久。判断方法很简单把代理摘掉直连数据库如果错误消失基本可以确认是这个原因。5. 常见异常与排查实录写代码的时间远少于查问题的时间这一章是我这几年攒下来的排查清单按报错现象 → 可能原因 → 处理方式整理。5.1 连接阶段报错速查表报错关键词最可能原因快速验证处理方式connection refused端口不通、服务未启动telnet 或 nc 探端口检查监听地址与防火墙connection timed out网络不可达、丢包加 connect_timeout 复现检查路由与安全组password authentication failed密码错、认证方式不匹配用同一账号命令行登录核对凭据与加密方式no pg_hba.conf entry来源 IP 未授权查服务端访问控制配置补授权条目并重载database does not exist库名拼错、连到了别的实例列出库列表确认修正 dbnametoo many clients already连接数打满查当前连接数与会话状态降池上限或加代理SSL required服务端强制加密连接串加 sslmode 试配置证书并开启加密关于认证失败有一个隐性的坑口令加密方式不同客户端 libpq 版本太老时会无法完成握手。表现出来不一定是明确的报错有时候是认证失败但密码明明是对的。这种情况升级驱动或者换用带新版 libpq 的 wheel 即可。5.2 时区与编码两个静默出错的重灾区时区问题。PG 的timestamptz类型内部统一按 UTC 存储返回给客户端时按会话时区转换。如果你的连接没有指定时区客户端拿到的是服务器默认时区的时间而 psycopg 会把它转成带时区信息的 datetime 对象。麻烦出在两种情况下一是代码里把带时区的对象和 naive 对象做比较Python 直接抛TypeError二是把 naive 对象存进timestamptz驱动会按本地时区解释产生偏移。我的统一做法是连接建立后立刻SET TIME ZONE UTC应用内所有时间处理都用带时区的对象展示层再做本地化。这样无论是数据库、应用还是日志时间语义都是一致的跨时区部署也不会有歧义。with conn.cursor() as cur: cur.execute(set time zone utc)编码问题。PG 服务端建库时指定了编码客户端连接时也有一个client_encoding两者必须匹配。不匹配时中文会出现乱码或者插入报错。稳妥的做法是建库就用 UTF8连接时显式指定conn psycopg.connect(dsn, client_encodingutf8)还有个更隐蔽的情况某些字符在 UTF8 里是合法的但包含零字节\x00PG 的文本类型不接受零字节插入时会报invalid byte sequence。清洗数据时记得把这类字符过滤掉。5.3 慢查询与锁等待怎么快速定位到具体那条语句服务变慢的时候第一件事不是猜代码而是去数据库看当前在跑什么。有几句查询我基本上是肌肉记忆-- 当前活跃会话按执行时长排序 select pid, usename, application_name, state, now() - query_start as duration, wait_event_type, wait_event, left(query, 200) as query_head from pg_stat_activity where state idle order by duration desc limit 20;看到wait_event_type Lock就说明在等锁用这个查谁堵了谁select blocked.pid as blocked_pid, blocked.query as blocked_query, blocking.pid as blocking_pid, blocking.state as blocking_state, blocking.query as blocking_query from pg_stat_activity blocked join pg_stat_activity blocking on blocking.pid any(pg_blocking_pids(blocked.pid)) where cardinality(pg_blocking_pids(blocked.pid)) 0;如果blocking_state是idle in transaction那就对上了——某个应用拿到连接开了事务却不提交锁一直不放。这时候看application_name就知道是哪个服务干的直接去找对应代码里漏掉的 commit。5.4 三条踩出来的经验第一条不要在事务里做网络调用。我见过一个接口在事务中调用第三方支付接口对方偶发超时 10 秒直接把数据库连接占满整个服务雪崩。正确的顺序是先完成本地事务再发外部请求必要时用补偿机制处理一致性。这条规则的价值高于绝大多数性能优化。第二条批量任务的连接数要单独限制。跑批任务往往一次性处理大量数据如果它和在线服务共用同一个池跑批一开始就会把池占满在线请求全部排队。我给跑批任务单独配一个池上限设得很小比如 2宁可它跑慢一点也不能影响在线。这个隔离思路在资源紧张的时候特别管用。第三条所有超时都要设置包括你认为不会超时的那些。连接超时、语句超时、事务空闲超时、获取连接超时四个都要设。没有超时保护的代码就像没有刹车的车平时看不出问题出事就是大事。设超时的另一个好处是它把偶发卡死变成了明确的错误而明确的错误是可以被日志和告警捕获的卡死不能。6. 把它封装成一个能长期用的模块前面讲的都是零散的知识点真正落到项目里你需要一个可复用的封装。这一章说说我在实际项目里怎么组织这部分代码以及几个封装时容易过度设计的地方。6.1 配置外置与最小连接工厂配置一定要从代码里拿出来。我的做法是用环境变量做最终来源本地开发时用.env文件加载进去部署时由平台注入。这样同一份代码在开发、测试、生产之间切换只需要换环境变量。import os from contextlib import contextmanager import psycopg from psycopg_pool import ConnectionPool def build_dsn() - str: return ( fhost{os.environ[PG_HOST]} fport{os.environ.get(PG_PORT, 5432)} fdbname{os.environ[PG_DB]} fuser{os.environ[PG_USER]} fpassword{os.environ[PG_PASSWORD]} fconnect_timeout5 fapplication_name{os.environ.get(APP_NAME, unknown)} foptions-c statement_timeout30000 ) _pool ConnectionPool( build_dsn(), min_size2, max_sizeint(os.environ.get(PG_POOL_MAX, 10)), timeout5, # 获取连接的最长等待 max_idle300, # 空闲回收 checkConnectionPool.check_connection, # 借出时探活 openTrue, )这里有两个设计取舍值得说明。一是池在模块加载时就创建而不是每次调用时创建——池是长生命周期的对象反复创建销毁等于没用池。二是池大小从环境变量读因为不同部署形态单机、多实例、跑批合适的值不一样硬编码会让同一份镜像没法复用。6.2 上下文管理器封装让调用方不可能忘记归还池用起来最容易出的错是借了忘还一旦漏掉一条池里就少一个可用连接跑久了池就空了。用上下文管理器可以把这个风险降到接近零contextmanager def get_cursor(commit: bool True): with _pool.connection() as conn: # 退出时自动归还 with conn.cursor() as cur: yield cur if commit: conn.commit() else: conn.rollback()调用方这样写with get_cursor() as cur: cur.execute(insert into orders (id, amount) values (%s, %s), (oid, amt))这里commit参数的设计是个小技巧。默认提交符合大多数写操作的直觉但对于只读查询显式传False让它走回滚路径可以避免在只读事务里留下不必要的快照。也有人在只读查询上直接开autocommit效果类似。注意不要在这个封装里再套一层 try/except 把异常吞掉。异常必须往上抛否则调用方无法知道操作失败了这是最糟糕的封装方式。6.3 什么时候该停手封装的反模式封装到一定程度就该停过度的抽象反而会带来麻烦。我列几个自己踩过的反模式。反模式一把所有查询都包装成query_one、query_many、execute三个函数。看起来整齐实际上把 SQL 和业务逻辑割裂了。复杂查询需要拼条件、需要分页、需要不同的事务隔离级别硬塞进统一接口会写出一堆if分支。我的做法是只封装连接获取和事务边界SQL 让业务代码自己写写清楚比封装漂亮重要。反模式二在封装层做自动重试。有些人喜欢在 get_cursor 里做失败自动重试三次看起来很健壮实际上非常危险。写操作重试可能导致重复插入事务重试可能掩盖真实的数据一致性问题。重试应该放在明确的、幂等的操作上由业务代码显式决定。反模式三用单例连接代替连接池。我见过为了省资源全局共享一条连接的实现然后在多线程环境下出现各种诡异的数据错乱——游标状态互相干扰、事务边界混乱。连接不是线程安全的对象多线程下必须一条线程一条连接或者用池。这条没有妥协余地。反模式四把连接对象存到请求上下文里跨请求复用。有些框架里能看到把连接塞进全局字典的写法本质上是手写了一个没有回收机制的池跑起来必然泄漏。要么用真正的池要么用完就关不要自己造中间态。6.4 和其他组件衔接时的两个注意点与 pandas 结合。数据分析场景常把查询结果直接转 DataFrame。方便是方便但要注意连接的生命周期read_sql内部会自己借连接如果传的是 SQLAlchemy 的 engine它会自动归还如果传的是裸连接要确保读完之后连接被正确释放。大结果集直接从数据库读进内存也容易 OOM超过几十万行建议分批读或者先用 COPY 导到本地文件再用 pandas 读。与异步框架结合。在 async 框架里使用同步驱动会阻塞事件循环把整个服务的并发能力废掉。要么用 psycopg3 的异步接口要么把数据库操作丢到线程池里执行。前者性能更好后者改动更小选哪个看现有代码的迁移成本。千万别在 async 函数里直接调同步驱动的 execute那是最隐蔽的性能杀手——单看功能都对压测才发现并发上不去。写到这里差不多了。这几年用 Python 操作 PG最大的感受是真正难的从来不是怎么连上而是连上之后怎么一直稳。驱动和连接池的 API 都很简单几行就能跑起来但线上真正出问题的永远是细节——一个没设的超时、一个忘了归还的连接、一个在事务里发出的外部请求。所以我现在写这类代码有个固定习惯先把四个超时全部设上再把连接池参数从环境变量读出来最后写一个最小的压测脚本跑十分钟看看连接数曲线。这三件事加起来花不了半小时但能挡掉后面大部分的深夜告警。如果只让我留一条经验那就是把application_name好好设上它在你最需要排查问题的时候是唯一能告诉你这条连接是谁的线索。 
长河君 · 2026/11/3 21:24:03 · 阅读 1.2k Python连接PostgreSQL生产实践：驱动选型、连接池与事务避坑
不少朋友看到Python 链接PG数据库这个标题第一反应是这不就是两行代码的事。我在前面几年也是这么想的直到线上服务在凌晨因为连接数打满被拖垮直到批量导入十万条数据跑了四十分钟直到某次排查发现报错日志里那个模糊的connection already closed其实是连接池里躺着一批被数据库单方面踢掉的僵尸连接。Python 连接 PostgreSQL 这件事写通和写好之间隔着的是一整条生产环境的鸿沟——驱动怎么选、连接池开多大、事务边界划在哪里、超时怎么设、异常怎么兜每一个点都能单独写一篇。这篇内容我把这几年从原型到生产踩过的路重新走了一遍把能跑到稳跑之间那些文档里不写、但你必须知道的细节都摊开来说。适合刚接触 Python 操作关系型数据库的新手按步骤复现也适合已经在用但偶尔被连接池和事务坑到的同学对照检查。标题里写的是链接实际大家在说的是连接下文统一按连接来说。1. 先把这件事拆开Python 连 PG 到底在连什么很多人对连接数据库的理解停留在提供一个账号密码就能查数据这种认知在写脚本时没有问题一旦进入服务化阶段就会处处碰壁。要把它做好得先知道这条链路上到底有几个环节、每个环节各自负责什么、出问题时会以什么形式暴露出来。1.1 一条 SQL 从 Python 到 PG 走过的四段路从你写下一行cursor.execute(select 1)开始到结果集回到你的 Python 进程中间其实穿了四道门。第一道是驱动层。你的 Python 代码并不直接和数据库进程说话中间必须有一个实现了数据库通信协议的库负责把 Python 对象序列化成 PG 认识的线协议格式再把返回的字节流反序列化成 Python 对象。这个角色在 Python 生态里主要就是 psycopg2 和 psycopg3 两位它们都是 C 扩展实现的性能接近也有纯 Python 的替代实现但吞吐差距明显生产环境基本不考虑。第二道是连接层也就是 TCP 握手 认证 会话初始化。这一步的成本经常被低估一次完整的连接建立包含 DNS 解析、TCP 三次握手、可能的 TLS 协商、PG 的认证往返口令认证至少一个 RTT、以及连接建立后驱动层发出的一系列SET语句客户端编码、时区、只读标记等。实测在同一个机房内网一次连接的建立耗时在 3 到 15 毫秒之间波动跨可用区能到 30 毫秒以上。这意味着如果你的业务是每次请求新建一个连接光是连接开销就能吃掉相当一部分响应时间。第三道是会话层连接建立之后就进入会话状态。这里有个新手容易忽略的事实PG 的连接是有状态的。你在会话里SET过的参数、开启过的事务、创建的临时表、拿到的 advisory lock都绑定在这条连接上。一旦这条连接被连接池复用给别的请求这些状态就会跟着流过去。我见过最典型的事故是某个接口为了调优在会话里设了statement_timeout 200结果这条连接被复用后另一个跑批任务的所有语句全部超时被杀。第四道是服务端PG 为每条连接 fork 一个独立的进程不是线程这个进程会占用一定的内存最小大概几 MB随 work_mem 等参数增长。这就决定了 PG 的连接数不是可以随便开的几百条活跃连接就可能把内存吃紧而且进程切换带来的上下文开销也会拖慢整体吞吐。理解这四段路的价值在于你的每一个优化决策都对应着其中某一段。用连接池是在优化第二段调work_mem是在优化第四段把关掉自动提交是在优化第三段——后面几章的内容基本都是围绕这四个层次展开的。1.2 三种主流接入姿势的分层与适用边界实际项目里Python 连 PG 不一定都走同一条路常见的有三种姿势它们不是互斥的很多项目是混着用的。裸驱动直连直接import psycopg2自己管连接、管游标、管事务。优点是零抽象、行为完全可控、排查问题链路最短缺点是所有东西都要自己写尤其事务和异常处理容易漏。适合一次性脚本、数据迁移工具、定时任务、以及需要精细控制 COPY 批量导入的场景。驱动 连接池在裸驱动外面套一层池请求来时借、用完还。这是绝大多数 Web 服务和后端 API 的标准做法。池的实现可以是驱动自带的简易池也可以是更成熟的三方池。优点是把连接建立成本摊薄了把并发控制住了缺点是池参数配错比不用池还危险这个后面单独讲。ORM / 查询构建器走 SQLAlchemy、Django ORM 这类框架SQL 由框架生成。优点是模型定义清晰、迁移方便、团队协作成本低缺点是生成的 SQL 可能不是你想要的复杂查询要么写不出要么性能差出问题时还得回头读框架生成的语句。我的经验是业务 CRUD 用 ORM 省事报表和批处理老老实实写原生 SQL两者在同一个项目里共存完全正常。接入方式典型场景连接建立成本可控性上手难度我的建议裸驱动直连脚本、迁移、批处理每次都有最高低一次性任务首选驱动 连接池Web 服务、常驻进程摊薄到接近零高中后端服务标配ORM 框架业务 CRUD、多表关联由框架池管理中中高业务层用分析层绕开直连 COPY 批量大批量导入导出一次最高中数据类任务必用1.3 一个判断标准你的场景需不需要池不是所有场景都需要连接池这一点很多教程没说清楚导致有人给一个跑五分钟就退出的脚本也套上了池纯属增加复杂度。我给一个简单的判断口径进程生命周期内数据库操作的次数少于 10 次且对延迟不敏感——直连就行用完就关。进程常驻且会持续接收外部请求——必须用池没有任何商量余地。进程常驻但是单线程跑批数据库操作很密集——可以用池但池大小设 1 到 2 就够重点是复用而不是并发。有多个线程或协程并发访问——必须用池且池大小要与并发度匹配这是最容易出事的一类。还有一个反向的判断如果你的服务是 Serverless 形态每个请求一个冷启动实例那么本地连接池的意义会被大幅削弱因为实例本身就不常驻。这时候更合适的是在数据库前面加一层连接代理让代理去承接真实连接应用侧只维持轻量连接。这个思路后面在池参数计算那一节会再提。2. 环境准备驱动选型与依赖安装的那些细节环境这一步看着最没技术含量但我统计过自己遇到过的连不上问题超过一半出在这一步——装错了包、装错了版本、装到了全局环境、缺了系统库。所以这一章值得单独展开。2.1 psycopg2 与 psycopg2-binary以及要不要上 psycopg3先说最容易踩的坑PyPI 上有两个名字极度相似的包psycopg2和psycopg2-binary。psycopg2是源码包安装时会现场编译需要本机有 C 编译器、Python 头文件以及 PostgreSQL 的客户端开发库提供 libpq。好处是它链接的是你系统上的 libpq能拿到最新的协议特性和安全补丁坏处是编译过程容易失败尤其在精简镜像、Alpine 基础镜像、没有 root 权限的环境里报错信息往往只有一句command gcc failed。psycopg2-binary是预编译好的 wheel安装即用不需要编译器和系统库。适合本地开发、CI 环境、快速验证。但它打包的 libpq 是构建时的那一版不会随系统更新官方文档也明确建议不要在生产环境长期依赖它。至于 psycopg3包名psycopg是同一批维护者的新一代实现。它有几点实质性改进原生支持 async/await、支持服务端参数绑定不用在客户端拼 SQL 字符串、支持流水线模式pipeline mode一次网络往返提交多条语句、连接和游标都支持上下文管理器。新项目我倾向于直接上 psycopg3老项目迁移则要看有没有依赖 psycopg2 特有的行为比如某些扩展类型的适配器注册方式不同。包名安装方式是否需编译生产建议典型适用psycopg2源码编译是推荐生产环境、需要跟随系统 libpqpsycopg2-binarywheel否不建议长期用本地开发、快速验证、CIpsycopgwheel否推荐新项目、需要 async、需要流水线安装命令大致是这样注意先确认自己在虚拟环境里python -m venv .venv source .venv/bin/activate pip install psycopg[binary] # psycopg3带二进制 libpq # 或者 pip install psycopg2-binary # 开发阶段图省事注意在容器镜像里装psycopg2之前基础镜像需要先装libpq-dev和build-essentialDebian 系装完再清理掉这些构建依赖否则镜像会白白膨胀几百 MB。多阶段构建是标准做法。2.2 虚拟环境与版本锁定的必要性我见过太多我本地能跑服务器上连不上的案例追下去八成是版本不一致。数据库驱动这种东西小版本之间的行为差异是实打实存在的比如某个版本改变了时间类型返回值的时区处理方式某个版本调整了连接失败时的异常类型。这些变化不会让你的代码报语法错而是让它在某个边界条件下悄悄出错。所以我的做法固定为三步虚拟环境隔离、依赖清单锁定、镜像与本地使用同一份清单。依赖清单里把主版本号卡住用~或者精确版本都行关键是不要留一个裸的包名在里面。# requirements.txt 片段 psycopg[binary]~3.1另外Windows 上开发、Linux 上部署的组合也值得提前在容器里跑一遍。我遇到过本地开发一切正常、部署到 Linux 后因为系统时区数据缺失导致时间字段全部偏移八小时的案例排查了两个小时才定位到是基础镜像没装 tzdata。2.3 连接参数逐个拆解哪些必须给哪些别乱给先把一个完整的连接串摆出来然后逐个说。import psycopg conn psycopg.connect( host10.0.0.12, port5432, dbnameappdb, userapp_user, password******, connect_timeout5, application_nameorder-service, options-c statement_timeout30000 -c idle_in_transaction_session_timeout60000, keepalives1, keepalives_idle30, keepalives_interval10, keepalives_count3, )connect_timeout是最容易被忽略的一个。它的默认值是无限等待这意味着如果数据库主机网络不可达比如防火墙静默丢包你的进程会卡在那里很久表现出来就是服务整体无响应。给它一个 3 到 5 秒的值让失败尽快暴露。application_name强烈建议一定要设。它会在pg_stat_activity里显示出来排查哪条连接是谁开的、占着锁不放手的时候这个字段能帮你省下大量时间。设成服务名加实例标识比如order-service-pod3。options可以传会话级参数。statement_timeout防止单条语句跑飞idle_in_transaction_session_timeout防止事务开着不结束白占连接——这两个参数我基本上每个服务都会设置属于最低成本的保险。keepalives 系列用于识别被中间设备静默断开的连接。云环境里的负载均衡器、NAT 网关普遍有空闲连接回收策略短则 5 分钟长则 1 小时。如果不同时设置 TCP keepalive 和应用层探活你会拿到一批看起来还活着、实际早就断了的连接等到真正要写入的时候才报错。提示密码不要硬编码在代码里。用环境变量、配置文件挂载或者专门的密钥管理服务具体选哪种看你的部署环境原则只有一条——密码不进入版本库、不进入日志、不进入错误堆栈。3. 从一次连接到稳定读写核心代码怎么写环境齐了接下来是代码。这一章我按先能跑通再跑稳的顺序来每一段代码都说明它为什么这么写。3.1 最小可用的连接与查询先看一个能跑的最小版本import psycopg with psycopg.connect( host10.0.0.12, port5432, dbnameappdb, userapp_user, password******, connect_timeout5, ) as conn: with conn.cursor() as cur: cur.execute(select id, name from users where status %s limit 10, (active,)) rows cur.fetchall() for r in rows: print(r)这段代码里有两个with含义完全不同很多人搞混。外层的with psycopg.connect(...)是连接级上下文正常退出时它负责提交事务并关闭连接异常退出时回滚并关闭。内层的with conn.cursor()只是游标级上下文退出时关闭游标不涉及事务提交。关键点在于 psycopg 默认不自动提交。也就是说你执行完一条insert如果不 commit在连接关闭时会被回滚。新手最常见的数据插进去了怎么又没了根因就在这里。想让它自动提交有两种写法连接时传autocommitTrue或者在执行后显式调用conn.commit()。我的习惯是脚本类任务用autocommitTrue服务类用显式事务控制。原因是脚本往往是一批独立操作逐条提交反而更符合语义服务里通常一组操作要么全成要么全败需要清晰的事务边界。3.2 参数化查询不只是防注入也是性能问题看到有人在代码里写fselect * from users where name {name}的时候我都会立刻拦下来。这不仅是因为 SQL 注入——虽然那已经足够致命——还因为拼字符串会让数据库完全无法复用执行计划。PG 接收到一条 SQL 之后会做解析、重写、生成执行计划这个过程的成本不低。对于结构相同、只有参数不同的语句PG 可以缓存执行计划直接复用。但你每次拼一个不同的字符串进去在它眼里就是一条全新的 SQL缓存全部失效解析开销翻倍。正确写法是参数化# 正确参数由驱动负责转义与绑定 cur.execute(select * from users where name %s and status %s, (name, status)) # 批量用 executemany或者更好的是 execute_values / COPY cur.executemany( insert into events (id, kind, created_at) values (%s, %s, %s), [(1, click, now), (2, view, now)], )注意参数占位符是%s不管字段是字符串、整数还是日期统一用%s由驱动根据 Python 对象的类型决定怎么编码。不要写成%d或?那是别的数据库驱动的写法混用会直接抛异常。还有一个细节不要用%s去拼表名或字段名。参数化只对值生效对标识符无效。如果你的表名是动态的必须做白名单校验只允许预定义的一组名字通过绝不能把用户输入直接当作标识符。3.3 事务边界三个必须说清楚的坑坑一未提交事务导致锁堆积。事务一旦开始它持有的行锁、表锁就会一直存在直到提交或回滚。如果代码在事务中间做了一件慢事比如调外部接口、写文件、等消息队列这段时间数据库侧的锁就一直在其他请求全被堵住。这就是为什么一定要设idle_in_transaction_session_timeout——它会把长时间挂在事务里不干活的连接直接杀掉虽然看起来粗暴但比拖垮整个库强。坑二自动提交与事务混用。有人把连接设成autocommitTrue然后在代码里又调conn.commit()这时候某些驱动版本会抛异常或者静默忽略行为不一致。要么全程自动提交要么全程显式控制不要混。坑三异常路径没回滚。一段事务代码里如果抛了异常却被捕获吞掉了事务会继续持有后续操作全在这个已损坏的事务里。稳妥的写法是把事务控制交给上下文管理器或者用明确的 try/except 里回滚再抛出。try: with conn.transaction(): with conn.cursor() as cur: cur.execute(update accounts set balance balance - %s where id %s, (amt, from_id)) cur.execute(update accounts set balance balance %s where id %s, (amt, to_id)) except Exception: # 上下文管理器已回滚这里只做日志与上抛 logger.exception(transfer failed) raiseconn.transaction()会创建一个保存点级别的子事务正常退出提交、异常退出回滚这样能把事务边界限制在一小段代码内比手动 commit/rollback 可靠得多。3.4 批量写入executemany 不是终点批量写是性能差异最悬殊的场景。我做过一组对比同一台机器、同一张表、十万条记录写入方式耗时相对值适用量级备注逐条 execute commit100x千级以下千万别这么干executemany12x万级内部仍是逐条发送execute_valuespsycopg22x十万级一条多值 INSERTpsycopg3 executemany 批模式2x十万级需设置批大小COPY1x百万级以上最快但错误处理粗糙结论很直接数据量上了万级就不要再逐条插。psycopg3 的executemany支持returning和批处理可以在连接上设置cursor.executemany(..., returningFalse)配合conn.pipeline()使用能显著减少网络往返。psycopg2 则用psycopg2.extras.execute_values。再往上到百万级直接上 COPYimport psycopg with psycopg.connect(dsn) as conn, conn.cursor() as cur: with cur.copy(copy events (id, kind, created_at) from stdin with (format csv)) as copy: for row in iter_rows(): copy.write_row(row) # psycopg3 的写法自动处理转义 conn.commit()COPY 的代价是它的错误处理不够友好一旦某一行数据格式不对整批都会失败且错误信息里给的行号有时候和你的源数据对不上。我的做法是先把数据打到临时表create temp table ... (like target including defaults)COPY 进临时表然后用一条insert ... select ... where not exists做校验和合并。这样既拿到了 COPY 的速度又能用 SQL 做数据清洗和冲突处理。注意COPY 有事务语义如果你在 COPY 进行中异常退出整批数据都不会落库。另外 COPY 期间表上会持有锁大表导入尽量选低峰期或者用分区表按批灌入。4. 连接池配对了是加速器配错了是定时炸弹到这一章基本进入生产环境的核心议题了。我在生产上遇到过的问题一大半和池有关。4.1 为什么每次请求新建连接是必然失败的先算一笔账。假设你的服务每秒处理 100 个请求每个请求一次数据库操作每次连接建立耗时 8 毫秒。如果每条请求都新建连接光连接建立就要消耗 0.8 个 CPU 秒每秒这还没算服务端 fork 进程的开销。更严重的是并发峰值100 QPS 如果不做任何限制全部同时打进来数据库侧会瞬间出现上百个连接进程内存和上下文切换一起上去响应变慢然后请求堆积然后更多连接被创建——正反馈崩溃。连接池的本质是用有限的连接服务无限的请求它把连接数这个变量从跟随请求量变成了一个可控的固定值。代价是请求可能要排队等连接但排队是可控的崩溃不是。4.2 池参数的估算方法与一个可复现的例子池里最关键的三个参数是最小连接数、最大连接数、空闲连接回收时间。市面上流传的估算公式很多我用的思路是按并发度倒推而不是按 QPS。先确定两个量单实例的并发工作线程数比如 Web 服务开了 8 个 worker每个 worker 同步处理一个请求那就最多 8 个并发数据库操作和单个操作的平均持有时长占比比如一次请求里数据库操作占 20 毫秒整个请求 100 毫秒那占用率是 20%。那么理论池大小 ≈ 并发线程数 × 占用率 × 安全系数。8 × 0.2 × 2 3.2取 4 到 8 都合理。这个结果往往比大家直觉上设的 50、100 小得多。再用全局视角校验一遍数据库侧max_connections假设是 200留 20 给运维和备份剩 180 给应用。应用部署了 6 个实例每实例池上限 20合计 120安全余量充足。如果每实例设 40合计 240直接超了——这就是常见的踩坑方式每个实例单独看都还好合起来把数据库撑爆。参数建议起点调整依据设太大的后果设太小的后果池最小连接数2 至 4稳态并发空闲连接白占内存突发时集中建连池最大连接数并发数×2压测观察等待时间打爆数据库连接上限请求排队、延迟升高空闲回收时间300 秒业务空闲周期频繁建连大量僵尸连接获取连接超时3 至 5 秒可接受排队时长故障时请求堆积轻微抖动就报错提示获取连接超时必须设而且要比你的接口超时小。不设的话池满了以后请求会无限等待表现出来是线程全部挂起比直接报错难排查一百倍。4.3 池化之后仍然断连探活与重连怎么做上了池不代表连接不会断。数据库主备切换、网络设备重启、服务端主动终止空闲连接idle_session_timeout、运维手动 kill——任何一种都会让池里出现失效连接。如果池不做探活你借到的第一条连接可能就是死的。psycopg3 的连接池提供了check参数借出连接时会做一次轻量检查失效则重建。psycopg2 自带的 pool 则没有这个能力需要自己包一层借出后先执行select 1捕获连接类异常就丢弃重取最多重试两次。def get_conn(pool, retries2): for i in range(retries 1): conn pool.getconn() try: with conn.cursor() as cur: cur.execute(select 1) return conn except psycopg.OperationalError: pool.putconn(conn, closeTrue) # 明确丢弃不要放回池里 if i retries: raise这里有个细节必须强调失效连接一定要用 close 标记归还。如果直接 putconn 放回池里下次借到它的线程还得再失败一次形成一个恶性的循环。另外重试要区分错误类型。连接类错误网络中断、连接已关闭可以重试业务类错误唯一键冲突、字段超长重试没有意义只会放大问题。区分方式是按驱动抛出的异常类型判断OperationalError通常可重试IntegrityError、ProgrammingError不要重试。4.4 一个容易被忽略的组合问题连接代理与预编译语句如果数据库前面加了一层连接代理用来承接大量短连接、做连接复用那么代理的工作模式会直接影响驱动的行为。代理如果工作在会话模式一条客户端连接对应一条服务端连接直到断开那基本没有兼容问题。如果工作在事务模式每条事务结束后客户端连接会被重新绑定到别的服务端连接上此时任何会话级状态都会失效——包括预编译语句。psycopg3 默认会在同一会话中缓存并使用服务端预编译语句遇到事务模式的代理就会报出prepared statement does not exist之类的错误。解决办法是在连接参数里关掉自动准备conn psycopg.connect(dsn, prepare_thresholdNone)或者显式设置为 0。这个坑我踩过一次报错信息完全不指向连接代理当时排查了很久。判断方法很简单把代理摘掉直连数据库如果错误消失基本可以确认是这个原因。5. 常见异常与排查实录写代码的时间远少于查问题的时间这一章是我这几年攒下来的排查清单按报错现象 → 可能原因 → 处理方式整理。5.1 连接阶段报错速查表报错关键词最可能原因快速验证处理方式connection refused端口不通、服务未启动telnet 或 nc 探端口检查监听地址与防火墙connection timed out网络不可达、丢包加 connect_timeout 复现检查路由与安全组password authentication failed密码错、认证方式不匹配用同一账号命令行登录核对凭据与加密方式no pg_hba.conf entry来源 IP 未授权查服务端访问控制配置补授权条目并重载database does not exist库名拼错、连到了别的实例列出库列表确认修正 dbnametoo many clients already连接数打满查当前连接数与会话状态降池上限或加代理SSL required服务端强制加密连接串加 sslmode 试配置证书并开启加密关于认证失败有一个隐性的坑口令加密方式不同客户端 libpq 版本太老时会无法完成握手。表现出来不一定是明确的报错有时候是认证失败但密码明明是对的。这种情况升级驱动或者换用带新版 libpq 的 wheel 即可。5.2 时区与编码两个静默出错的重灾区时区问题。PG 的timestamptz类型内部统一按 UTC 存储返回给客户端时按会话时区转换。如果你的连接没有指定时区客户端拿到的是服务器默认时区的时间而 psycopg 会把它转成带时区信息的 datetime 对象。麻烦出在两种情况下一是代码里把带时区的对象和 naive 对象做比较Python 直接抛TypeError二是把 naive 对象存进timestamptz驱动会按本地时区解释产生偏移。我的统一做法是连接建立后立刻SET TIME ZONE UTC应用内所有时间处理都用带时区的对象展示层再做本地化。这样无论是数据库、应用还是日志时间语义都是一致的跨时区部署也不会有歧义。with conn.cursor() as cur: cur.execute(set time zone utc)编码问题。PG 服务端建库时指定了编码客户端连接时也有一个client_encoding两者必须匹配。不匹配时中文会出现乱码或者插入报错。稳妥的做法是建库就用 UTF8连接时显式指定conn psycopg.connect(dsn, client_encodingutf8)还有个更隐蔽的情况某些字符在 UTF8 里是合法的但包含零字节\x00PG 的文本类型不接受零字节插入时会报invalid byte sequence。清洗数据时记得把这类字符过滤掉。5.3 慢查询与锁等待怎么快速定位到具体那条语句服务变慢的时候第一件事不是猜代码而是去数据库看当前在跑什么。有几句查询我基本上是肌肉记忆-- 当前活跃会话按执行时长排序 select pid, usename, application_name, state, now() - query_start as duration, wait_event_type, wait_event, left(query, 200) as query_head from pg_stat_activity where state idle order by duration desc limit 20;看到wait_event_type Lock就说明在等锁用这个查谁堵了谁select blocked.pid as blocked_pid, blocked.query as blocked_query, blocking.pid as blocking_pid, blocking.state as blocking_state, blocking.query as blocking_query from pg_stat_activity blocked join pg_stat_activity blocking on blocking.pid any(pg_blocking_pids(blocked.pid)) where cardinality(pg_blocking_pids(blocked.pid)) 0;如果blocking_state是idle in transaction那就对上了——某个应用拿到连接开了事务却不提交锁一直不放。这时候看application_name就知道是哪个服务干的直接去找对应代码里漏掉的 commit。5.4 三条踩出来的经验第一条不要在事务里做网络调用。我见过一个接口在事务中调用第三方支付接口对方偶发超时 10 秒直接把数据库连接占满整个服务雪崩。正确的顺序是先完成本地事务再发外部请求必要时用补偿机制处理一致性。这条规则的价值高于绝大多数性能优化。第二条批量任务的连接数要单独限制。跑批任务往往一次性处理大量数据如果它和在线服务共用同一个池跑批一开始就会把池占满在线请求全部排队。我给跑批任务单独配一个池上限设得很小比如 2宁可它跑慢一点也不能影响在线。这个隔离思路在资源紧张的时候特别管用。第三条所有超时都要设置包括你认为不会超时的那些。连接超时、语句超时、事务空闲超时、获取连接超时四个都要设。没有超时保护的代码就像没有刹车的车平时看不出问题出事就是大事。设超时的另一个好处是它把偶发卡死变成了明确的错误而明确的错误是可以被日志和告警捕获的卡死不能。6. 把它封装成一个能长期用的模块前面讲的都是零散的知识点真正落到项目里你需要一个可复用的封装。这一章说说我在实际项目里怎么组织这部分代码以及几个封装时容易过度设计的地方。6.1 配置外置与最小连接工厂配置一定要从代码里拿出来。我的做法是用环境变量做最终来源本地开发时用.env文件加载进去部署时由平台注入。这样同一份代码在开发、测试、生产之间切换只需要换环境变量。import os from contextlib import contextmanager import psycopg from psycopg_pool import ConnectionPool def build_dsn() - str: return ( fhost{os.environ[PG_HOST]} fport{os.environ.get(PG_PORT, 5432)} fdbname{os.environ[PG_DB]} fuser{os.environ[PG_USER]} fpassword{os.environ[PG_PASSWORD]} fconnect_timeout5 fapplication_name{os.environ.get(APP_NAME, unknown)} foptions-c statement_timeout30000 ) _pool ConnectionPool( build_dsn(), min_size2, max_sizeint(os.environ.get(PG_POOL_MAX, 10)), timeout5, # 获取连接的最长等待 max_idle300, # 空闲回收 checkConnectionPool.check_connection, # 借出时探活 openTrue, )这里有两个设计取舍值得说明。一是池在模块加载时就创建而不是每次调用时创建——池是长生命周期的对象反复创建销毁等于没用池。二是池大小从环境变量读因为不同部署形态单机、多实例、跑批合适的值不一样硬编码会让同一份镜像没法复用。6.2 上下文管理器封装让调用方不可能忘记归还池用起来最容易出的错是借了忘还一旦漏掉一条池里就少一个可用连接跑久了池就空了。用上下文管理器可以把这个风险降到接近零contextmanager def get_cursor(commit: bool True): with _pool.connection() as conn: # 退出时自动归还 with conn.cursor() as cur: yield cur if commit: conn.commit() else: conn.rollback()调用方这样写with get_cursor() as cur: cur.execute(insert into orders (id, amount) values (%s, %s), (oid, amt))这里commit参数的设计是个小技巧。默认提交符合大多数写操作的直觉但对于只读查询显式传False让它走回滚路径可以避免在只读事务里留下不必要的快照。也有人在只读查询上直接开autocommit效果类似。注意不要在这个封装里再套一层 try/except 把异常吞掉。异常必须往上抛否则调用方无法知道操作失败了这是最糟糕的封装方式。6.3 什么时候该停手封装的反模式封装到一定程度就该停过度的抽象反而会带来麻烦。我列几个自己踩过的反模式。反模式一把所有查询都包装成query_one、query_many、execute三个函数。看起来整齐实际上把 SQL 和业务逻辑割裂了。复杂查询需要拼条件、需要分页、需要不同的事务隔离级别硬塞进统一接口会写出一堆if分支。我的做法是只封装连接获取和事务边界SQL 让业务代码自己写写清楚比封装漂亮重要。反模式二在封装层做自动重试。有些人喜欢在 get_cursor 里做失败自动重试三次看起来很健壮实际上非常危险。写操作重试可能导致重复插入事务重试可能掩盖真实的数据一致性问题。重试应该放在明确的、幂等的操作上由业务代码显式决定。反模式三用单例连接代替连接池。我见过为了省资源全局共享一条连接的实现然后在多线程环境下出现各种诡异的数据错乱——游标状态互相干扰、事务边界混乱。连接不是线程安全的对象多线程下必须一条线程一条连接或者用池。这条没有妥协余地。反模式四把连接对象存到请求上下文里跨请求复用。有些框架里能看到把连接塞进全局字典的写法本质上是手写了一个没有回收机制的池跑起来必然泄漏。要么用真正的池要么用完就关不要自己造中间态。6.4 和其他组件衔接时的两个注意点与 pandas 结合。数据分析场景常把查询结果直接转 DataFrame。方便是方便但要注意连接的生命周期read_sql内部会自己借连接如果传的是 SQLAlchemy 的 engine它会自动归还如果传的是裸连接要确保读完之后连接被正确释放。大结果集直接从数据库读进内存也容易 OOM超过几十万行建议分批读或者先用 COPY 导到本地文件再用 pandas 读。与异步框架结合。在 async 框架里使用同步驱动会阻塞事件循环把整个服务的并发能力废掉。要么用 psycopg3 的异步接口要么把数据库操作丢到线程池里执行。前者性能更好后者改动更小选哪个看现有代码的迁移成本。千万别在 async 函数里直接调同步驱动的 execute那是最隐蔽的性能杀手——单看功能都对压测才发现并发上不去。写到这里差不多了。这几年用 Python 操作 PG最大的感受是真正难的从来不是怎么连上而是连上之后怎么一直稳。驱动和连接池的 API 都很简单几行就能跑起来但线上真正出问题的永远是细节——一个没设的超时、一个忘了归还的连接、一个在事务里发出的外部请求。所以我现在写这类代码有个固定习惯先把四个超时全部设上再把连接池参数从环境变量读出来最后写一个最小的压测脚本跑十分钟看看连接数曲线。这三件事加起来花不了半小时但能挡掉后面大部分的深夜告警。如果只让我留一条经验那就是把application_name好好设上它在你最需要排查问题的时候是唯一能告诉你这条连接是谁的线索。