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 里贴一份,别靠猜。