Go 客户端 pgx
用 pgx v5（当前最新 v5.11.0，2026-09-07 发布）直连 PostgreSQL：连接池、预编译语句、批量导入、事务重试、类型映射——以及它在 PgBouncer、prepared statement 这些经典地雷区里的正确姿势。
为什么是 pgx
Go 里连 PG 大致三条路，先把选型讲清楚再谈 API：
- pgx v5（原生接口 +
pgxpool）：直接实现 PostgreSQL 前端/后端协议，不走database/sql抽象。二进制协议、服务端 prepared statement 缓存、COPY、Batch、Listen/Notify、每查询可覆盖执行模式，都是它的独占优势。jackc/pgx 的 README 自己就写着"新项目建议优先使用 pgx 而不是 database/sql"。 - lib/pq：老牌纯 Go 驱动，只支持
database/sql。维护者已宣布项目进入维护模式（只保证跟得上新 PG 大版本，不再加功能），缺连接池（要配第三方 pool）、缺 context 取消的可靠传递、类型映射靠反射。存量代码能用，新代码没必要选它。 - GORM 的 postgres driver：底层其实包的就是 pgx v5。老项目继续用没问题（见
gorm/手册），但直写 SQL 的场景套一层 ORM 是白交税。
结论：新项目直接用 pgx 原生接口。只有两种情况例外——必须塞进依赖 *sql.DB 的老框架（用兼容层 github.com/jackc/pgx/v5/stdlib，驱动名注册为 pgx / pgx/v5，stdlib.OpenDB 可以从已有的 *pgx.ConnConfig 或整个池子 OpenDBFromPool 建 *sql.DB）；或者团队已有大量 database/sql 肌肉记忆。一个必须提前知道的边界：pgx v5.11.0 的 go.mod 要求 Go 1.25（模块指令 go 1.25.0），老工具链装不上新版，要么锁旧版 pgx，要么升 Go。
本章的 Go 片段统一假设下面两张演示表——为了每段代码自包含，这里不复用 02-数据类型与DDL.md 的底座 users（同名不同形，别混着跑）：
CREATE TABLE users (
id serial PRIMARY KEY,
name text NOT NULL UNIQUE,
email text,
tags text[] NOT NULL DEFAULT '{}',
created_at timestamptz NOT NULL DEFAULT now()
);
CREATE TABLE accounts (
id int PRIMARY KEY,
owner text,
balance numeric(12,2) NOT NULL DEFAULT 0
);连接与池：pgxpool
一个连接串两种写法等价：DSN（host=... user=...）或 URL（postgres://user:pass@host:5432/db?sslmode=disable）。URL 里 sslmode 缺省是 prefer——生产建议显式写 require 或 verify-full（TLS 细节见 12-安全与权限管理.md）。
package main
import (
"context"
"log"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
cfg, err := pgxpool.ParseConfig("postgres://app:secret@localhost:5432/shop?sslmode=disable")
if err != nil {
log.Fatal(err)
}
cfg.MaxConns = 16 // 默认是 max(4, NumCPU)
cfg.MinIdleConns = 2 // 预热，砍掉首请求建连延迟
cfg.MaxConnLifetime = 30 * time.Minute // 默认 1h
cfg.MaxConnIdleTime = 5 * time.Minute // 默认 30min
cfg.HealthCheckPeriod = 30 * time.Second // 默认 1min，顺带回收超时空闲
cfg.AfterConnect = func(ctx context.Context, c *pgx.Conn) error {
// 每个新物理连接执行一次：适合设 application_name、注册自定义类型
_, err := c.Exec(ctx, `SET application_name = 'order-svc'`)
return err
}
pool, err := pgxpool.NewWithConfig(ctx, cfg)
if err != nil {
log.Fatal(err)
}
defer pool.Close()
if err := pool.Ping(ctx); err != nil { // New 是懒建连，Ping 才真正探活
log.Fatal(err)
}
}池参数里最反直觉的是 MaxConns 默认不是"无限"也不是"很大"，而是 max(4, 核数)。这是刻意设计：PG 是"一连接一进程"模型（见 01-架构与核心概念.md、05-事务与锁.md），连接数超过核数后吞吐不升反降，每个连接还要吃 ~10MB 级内存和一份 work_mem 预算。所以调池的正确姿势不是加大 pgxpool，而是想清楚"并发查询数"和"CPU/IO 容量"的关系（展开见 09-查询优化与EXPLAIN.md 的连接池一节）。
ParseConfig 还认这些 URL 参数，可以不写代码直接调池：pool_max_conns、pool_min_conns、pool_min_idle_conns、pool_max_conn_lifetime、pool_max_conn_idle_time、pool_health_check_period。另外 MaxConnLifetimeJitter 值得设一下（比如 lifetime 的 10%），避免整池连接同一秒到期被重建；BeforeConnect/AfterConnect 是钩子家族里现在的正主——v4 时代的 conncmap 包在 v5 源码树里已经不存在，网上教程里见到它直接跳过。
监控用 pool.Stat()：TotalConns/IdleConns/AcquiredConns/ConstructingConns/MaxConns 给水位，AcquireCount/EmptyAcquireCount/EmptyAcquireWaitTime 给"池不够用"的证据，NewConnsCount/MaxLifetimeDestroyCount/MaxIdleDestroyCount 给建连销毁速率。注意 v4 的 GetStatuses() 在 v5 已删，只有 Stat()。
查询姿势
四种姿势按场景选，别一种用到死。
package main
import (
"context"
"errors"
"fmt"
"log"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
type User struct {
ID int64 `db:"id"`
Name string `db:"name"`
Email string `db:"email"`
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
pool, err := pgxpool.New(ctx, "postgres://app:secret@localhost:5432/shop?sslmode=disable")
if err != nil {
log.Fatal(err)
}
defer pool.Close()
// 姿势一：Query + 手动 Scan —— 大结果集流式处理的首选
rows, err := pool.Query(ctx,
`SELECT id, name FROM users WHERE created_at > $1`,
time.Now().AddDate(0, 0, -7))
if err != nil {
log.Fatal(err)
}
defer rows.Close()
for rows.Next() {
var id int64
var name string
if err := rows.Scan(&id, &name); err != nil {
log.Fatal(err)
}
fmt.Println(id, name)
}
if err := rows.Err(); err != nil {
// Next 返回 false 不代表正常读完，必须查 Err()
log.Fatal(err)
}
// 姿势二：QueryRow —— 取标量/单行，无行时 Scan 返回 pgx.ErrNoRows
var email string
err = pool.QueryRow(ctx, `SELECT email FROM users WHERE id = $1`, 42).Scan(&email)
if errors.Is(err, pgx.ErrNoRows) {
fmt.Println("用户不存在")
} else if err != nil {
log.Fatal(err)
}
// 姿势三：CollectRows + 结构体映射 —— 小结果集直接收成切片
rows2, err := pool.Query(ctx, `SELECT id, name, email FROM users ORDER BY id LIMIT 100`)
if err != nil {
log.Fatal(err)
}
users, err := pgx.CollectRows(rows2, pgx.RowToStructByNameLax[User])
if err != nil {
log.Fatal(err)
}
fmt.Println(len(users))
// 姿势四：Exec —— 不取结果，CommandTag 拿影响行数
ct, err := pool.Exec(ctx, `UPDATE users SET name = $1 WHERE id > $2`, "renamed", 100)
if err != nil {
log.Fatal(err)
}
fmt.Println(ct.String(), ct.RowsAffected()) // 例如 "UPDATE 37" 37
}结构体映射三兄弟的差异在匹配方式，全部支持 db 标签（db:"-" 字段被跳过）：
RowToStructByPos：按位置匹配，结构体公有字段数必须等于列数，SELECT列顺序必须和字段声明顺序严格一致。改了 SELECT 顺序忘了改结构体，得到的是"类型对不上"的扫描错误而不是报错在哪个字段——列多的大表慎用。RowToStructByName：按列名匹配（大小写不敏感，可被db:"xx"标签覆盖），要求结构体"命名字段数 == 列数"，一列多余就报错。RowToStructByNameLax：同上，但只要求字段数 >= 列数，多余列忽略。日常默认选它。
还有一组变体：CollectOneRow（空结果返回 pgx.ErrNoRows）、CollectExactlyOneRow（多于一行就报错）、AppendRows（往已有切片追加）、RowTo[T]（单列直接收成 []T，比如只查 id 时 pgx.CollectRows(rows, pgx.RowTo[int64])）、RowToMap（列名不确定时用）；不想全量收集就用回调版 pgx.ForEachRow(rows, []any{&id, &name}, func() error { ... })。没有 ForRows——某些帖子写的"泛型 for-range 迭代行"在 v5.11.0 源码里不存在，别照着抄。
注意
CollectRows会把整个结果集读进内存，百万行请回到姿势一（流式）或加LIMIT；另外pool.Query返回的 rows 占用池连接，没Close/Collect干净之前连接不会归还。
参数、预编译与执行模式
占位符是 $1, $2，不是 MySQL 式的 ?。pgx 默认走扩展协议：每条 SQL 第一次执行时 PREPARE，之后命中 LRU 缓存复用，缓存容量 StatementCacheCapacity 默认 512（v5.11.0 源码实测；网上广泛流传的"默认 100"是旧版本数值）。
两个由此而来的坑：
- 服务端 prepared 数量没有上限。PG 不会替你限，每个 backend 里每条 prepared 都占内存，代码里如果动态拼 SQL（把日期、租户 id 拼进语句文本）会让缓存疯狂 miss、疯狂 PREPARE，backend 内存悄悄涨到 OOM。参数一律走
$n，让语句文本保持稳定。 - PgBouncer 的 transaction 池模式会和 prepared statement 打架：prepared 语句挂在具体 server 连接上，客户端下次被分到另一条连接时
prepared statement "xx" does not exist。三条解法按优先级：升级 PgBouncer 到 1.21+ 并开启协议级 prepared 支持（max_prepared_statements，有数量上限要规划）；或把 pgx 降档——cfg.ConnConfig.DefaultQueryExecMode = pgx.QueryExecModeExec（每条重新描述、无名字语句，缓存友好度差但兼容池化）或QueryExecModeDescribeExec；单条查询也可以临时降档——把pgx.QueryExecModeSimpleProtocol之类的模式值作为 SQL 之后的第一个参数传入。展开见09-查询优化与EXPLAIN.md。
超时控制靠 context，这是 pgx 相对 lib/pq 的硬优势——ctx 取消/到期会真发 CancelRequest 终止服务端查询：
ctx, cancel := context.WithTimeout(ctx, 500*time.Millisecond)
defer cancel()
// 超时时服务端会报 "canceling statement due to user request"
users, err := pgx.CollectRows(pool.Query(ctx, `SELECT ... 慢查询 ...`), pgx.RowToStructByNameLax[User])注意 pool.Query 返回两个值时不能直接塞进 CollectRows，Go 的多值传参要求占满全部形参——像本章例子那样先接住 rows, err 再传。
批量与导入：Batch 与 CopyFrom
Batch 把多条语句打包成一次往返。代价是错误语义很粗：一批在服务端跑在隐式事务里，某条报错，后面的语句全部不执行，BatchResults 的后续读取都返回中止错误——所以别拿 Batch 当"并行尽力而为"用，它本质是一条事务语句流。
var b pgx.Batch
for i := 1; i <= 1000; i++ {
b.Queue(`INSERT INTO users (name, email) VALUES ($1, $2)`,
fmt.Sprintf("user%d", i), fmt.Sprintf("user%d@example.com", i))
}
br := pool.SendBatch(ctx, &b) // pool/ctx 同上
defer br.Close() // 必须 Close：批内出错时连接状态靠它收拾
for i := 0; i < b.Len(); i++ {
if _, err := br.Exec(); err != nil {
log.Fatalf("第 %d 条失败: %v", i, err) // 此后剩余语句不再执行
}
}真正的吞吐王者是 CopyFrom——走 COPY 协议，批量插入比逐行 INSERT 快一个数量级（少了逐条 WAL/fsync 排队与解析开销），万级起步就应换它。它同时是 PG 端大批量写入的第一选择（对应 psql 的 \copy，见 11-备份恢复与运维.md）：
rows := make([][]any, 0, 100_000)
for i := 0; i < 100_000; i++ {
rows = append(rows, []any{fmt.Sprintf("u%d", i), fmt.Sprintf("u%d@example.com", i)})
}
n, err := pool.CopyFrom(ctx,
pgx.Identifier{"users"}, // 带 schema 写成 {"public", "users"}，元素会被正确加引号
[]string{"name", "email"},
pgx.CopyFromRows(rows)) // 返回实际写入行数
if err != nil {
log.Fatal(err)
}
fmt.Println("导入", n, "行")数据源接口 CopyFromSource 还有两个便利构造：CopyFromSlice(len, func(i int) ([]any, error)) 适合从已有切片惰性转换，CopyFromFunc 包一个自定义迭代器（读文件/分页拉取时用）。批大小经验：十万行级以内直接全量 CopyFromRows 没问题，更大时切成 1–5 万一档多次调用，避免内存峰值。
事务与重试
pool.Begin() / pool.BeginTx(ctx, pgx.TxOptions{...}) 拿 Tx，隔离级别是强类型常量：pgx.Serializable、pgx.RepeatableRead、pgx.ReadCommitted、pgx.ReadUncommitted（语义与 PG 对应关系见 05-事务与锁.md）。Tx 上也有一整套 Exec/Query/CopyFrom/SendBatch，tx.Begin() 会落到 SAVEPOINT 模拟嵌套。
tx, err := pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx) // 已 Commit 后 Rollback 返回 pgx.ErrTxClosed，忽略即可
if _, err := tx.Exec(ctx, `UPDATE accounts SET balance = balance - $1 WHERE id = $2`, amount, from); err != nil {
return err
}
if _, err := tx.Exec(ctx, `UPDATE accounts SET balance = balance + $1 WHERE id = $2`, amount, to); err != nil {
return err
}
return tx.Commit(ctx)嫌手写回滚烦可以用官方 helper：pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error {...}) / pgx.BeginTxFunc，内部处理"函数返回错误就 Rollback、否则 Commit"。注意一个细节：文档明说 ctx 只作用于 BEGIN/COMMIT/ROLLBACK 语句，不会自动包进你的函数体——业务逻辑超时还是要自己在 fn 里派生 ctx。
Serializable 下写写冲突会收到 40001（serialization_failure）、死锁是 40P01，这两类是"换个时间再来大概率能成"的可重试错误，标准封装：
// 依赖：context / errors / time / pgx / pgconn / pgxpool
func withRetry(ctx context.Context, pool *pgxpool.Pool, fn func(pgx.Tx) error) error {
var lastErr error
for attempt := 1; attempt <= 5; attempt++ {
err := pgx.BeginTxFunc(ctx, pool, pgx.TxOptions{IsoLevel: pgx.Serializable}, fn)
if err == nil {
return nil
}
var pgErr *pgconn.PgError
if errors.As(err, &pgErr) && (pgErr.Code == "40001" || pgErr.Code == "40P01") {
lastErr = err
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(time.Duration(attempt*attempt) * 50 * time.Millisecond):
}
continue
}
return err // 非并发冲突，重试无意义
}
return lastErr
}
// 用法——upsert 这类"撞了大概率能成"的写事务：
err := withRetry(ctx, pool, func(tx pgx.Tx) error {
_, err := tx.Exec(ctx, `
INSERT INTO users (id, name) VALUES ($1, $2)
ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name`, 1, "alice")
return err
})类型映射
PG ↔ Go 的对应关系，记主干就够：text/varchar→string、int2/int4/int8→int16/int32/int64（注意 int4[]→[]int32，不是 []int）、timestamptz→time.Time、bool→bool、bytea→[]byte、enum→string、interval→pgtype.Interval。剩下的疑难类型各说一句：
- NULL：扫进
*string/*int64（nil 表示 NULL）最省事；要区分"NULL 与零值"且保持值类型，用pgtype.Text{String, Valid}、pgtype.Int8{Int64, Valid}这一族包装体。 - numeric：对应
pgtype.Numeric{Int *big.Int, Exp int32, Valid bool}。钱绝不要扫成float64（精度丢失，见02-数据类型与DDL.md选 numeric 的理由）；展示时n.Float64Value()返回pgtype.Float8，或直接用Int/Exp转十进制字符串。 - jsonb/json：默认解成
[]byte，自己json.Unmarshal到目标结构体；pgx 也支持把实现了json.Unmarshaler的类型直接扫进去。 - uuid：原生是
pgtype.UUID{Bytes [16]byte, Valid bool}；社区更常直接用google/uuid——它实现了sql.Scanner/driver.Valuer，Scan(&uid)和作参数传入都开箱即用。 - 数组：一维
[]T直接映射；多维要[][]T且维度整齐，NULL 元素只能落在指针切片。
var (
id int64
name *string // NULL 安全：扫出来 nil
createdAt time.Time
price pgtype.Numeric
profile []byte // jsonb 先拿原文
uid pgtype.UUID
tags []string // text[]
role string // enum 列就是它的字符串
)
err := row.Scan(&id, &name, &createdAt, &price, &profile, &uid, &tags, &role)
if err != nil {
log.Fatal(err)
}错误处理与 Listen/Notify
结构化错误是 *pgconn.PgError（含 Code/Message/ConstraintName/Detail），用 errors.As 取。最常用的 SQLSTATE：23505 unique_violation、23503 foreign_key_violation、40001 serialization_failure、40P01 deadlock_detected。惯用判断：
func isUniqueViolation(err error) bool {
var pgErr *pgconn.PgError
return errors.As(err, &pgErr) && pgErr.Code == "23505"
}INSERT ... ON CONFLICT 能覆盖的场景优先用它，捕获 23505 适合"并发注册抢注"这类冲突当业务信号的场合；需要 RETURNING 回读落库结果配合（见 03-SQL查询实战.md）。
LISTEN/Notify（做缓存失效广播很顺手，见 11-备份恢复与运维.md）必须挂在一条固定的专用连接上——监听状态是 backend 进程级的，过池拿连接没有意义：
conn, err := pool.Acquire(ctx) // 独占一条物理连接，用完 Release
if err != nil {
log.Fatal(err)
}
defer conn.Release()
c := conn.Conn()
if _, err := c.Exec(ctx, `LISTEN cache_invalidate`); err != nil {
log.Fatal(err)
}
for {
n, err := c.WaitForNotification(ctx) // 阻塞直到有通知或 ctx 结束
if err != nil {
log.Println("停止监听:", err)
return
}
fmt.Printf("[%s] %s (pid %d)\n", n.Channel, n.Payload, n.PID)
}注意 独占连接会长期占住池的一个槽位，
MaxConns记得给它留余量；通知载荷上限约 8KB，只发 key 不发整个对象。
测试策略
pgx 的价值（COPY、Batch、扩展类型）在 mock 层基本测不出来，以集成测试为主：
- 本地起 PG 最简单——复用
00-安装与配置.md里的 docker compose，测试用独立 database，go test -count=1直接连。 - CI 用
testcontainers-go（github.com/testcontainers/testcontainers-go/modules/postgres），每个测试套件起一个postgres:18容器，跑完销毁；配合pool.Ping等待就绪。 - 业务层单测要隔离就隔离仓库接口（自己定义 interface，pgx 代码是实现），别去 mock
*sql.DB的每一行——那测的是你对 mock 的想象；建表语句保持单源（跑02-数据类型与DDL.md的 DDL），别在 Go 字符串里维护第二套 schema。
完整示例：建池、建表、CopyFrom、事务 upsert、CollectRows
go.mod：
module pgx-demo
go 1.25
require github.com/jackc/pgx/v5 v5.11.0main.go（连本地 docker 起的 PG，环境变量 PGURL 传入连接串）：
package main
import (
"context"
"fmt"
"log"
"os"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
type User struct {
ID int32 `db:"id"`
Name string `db:"name"`
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
pool, err := pgxpool.New(ctx, os.Getenv("PGURL"))
if err != nil {
log.Fatal(err)
}
defer pool.Close()
if _, err := pool.Exec(ctx, `
CREATE TABLE IF NOT EXISTS users (
id serial PRIMARY KEY,
name text NOT NULL UNIQUE,
email text,
tags text[] NOT NULL DEFAULT '{}',
created_at timestamptz NOT NULL DEFAULT now()
)`); err != nil {
log.Fatal(err)
}
// 1) CopyFrom 批量导入 50k 行，和逐行 Batch 对比耗时
names := []string{"alice", "bob", "carol", "dave", "erin"}
rows := make([][]any, 0, 50_000)
for i := 0; i < 50_000; i++ {
rows = append(rows, []any{fmt.Sprintf("%s-%d", names[i%len(names)], i), []string{"batch"}})
}
start := time.Now()
n, err := pool.CopyFrom(ctx, pgx.Identifier{"users"},
[]string{"name", "tags"}, pgx.CopyFromRows(rows))
if err != nil {
log.Fatal(err)
}
fmt.Printf("CopyFrom %d 行, 耗时 %s\n", n, time.Since(start).Round(time.Millisecond))
// 2) 事务 upsert：撞 UNIQUE 就转更新
err = pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error {
_, err := tx.Exec(ctx, `
INSERT INTO users (name, tags) VALUES ($1, $2)
ON CONFLICT (name) DO UPDATE SET tags = EXCLUDED.tags`,
"alice-0", []string{"vip"})
return err
})
if err != nil {
log.Fatal(err)
}
// 3) CollectRows 读回打印
qctx, qcancel := context.WithTimeout(ctx, 3*time.Second)
defer qcancel()
r, err := pool.Query(qctx, `SELECT id, name FROM users ORDER BY id LIMIT 5`)
if err != nil {
log.Fatal(err)
}
users, err := pgx.CollectRows(r, pgx.RowToStructByNameLax[User])
if err != nil {
log.Fatal(err)
}
for _, u := range users {
fmt.Printf("user(%d, %s)\n", u.ID, u.Name)
}
}跑起来对 CopyFrom 的体感应该是毫秒到秒级，换成逐行 Exec 会直接差出十倍量级——这就是本章反复强调"导入必须 COPY"的原因。
实战建议
- 池起步值
MaxConns = 核数，压测再调；MinIdleConns预热两三条砍掉尾延迟，MaxConnLifetime配 jitter 让连接滚动重建。 - 小结果集用
CollectRows + RowToStructByNameLax，大结果集流式Query/Next/Scan，别忘了rows.Err()。 - 一切参数走
$n占位符，杜绝拼 SQL——否则 prepared 缓存爆炸、backend 内存暗涨；语句文本保持稳定。 - 过 PgBouncer 时：要么 PgBouncer ≥1.21 开启 prepared 支持，要么 pgx 侧
QueryExecModeExec/SimpleProtocol降档，别让默认 cache_statement 裸奔（连接与限流全景见09-查询优化与EXPLAIN.md）。 - 万行以上导入一律
CopyFrom；“多条语句一次往返"才用Batch，并把它当隐式事务对待（一败全败）。 - Serializable 事务包一层 40001/40P01 重试 helper；非这两类错误直接抛，别拿重试掩盖逻辑 bug。
- 每个数据库调用都传带 deadline 的
ctx——这是 pgx 相对database/sql驱动最可靠的差异点，用足它。 - 钱用
pgtype.Numeric，NULL 用指针或pgtype.*包装，enum 用 string——类型映射表在项目 wiki 里贴一份，别靠猜。