Go 数据访问：管好连接、事务和 SQL 边界

一段查询可以很短，却同时藏着四种问题：结果没有关闭、等待连接没有超时、事务内混进池上的操作、用户输入被拼进 SQL。换 ORM 不能自动消除这些问题，底层资源和数据库语义仍然在那里。

本文围绕 database/sql 讲连接池、预处理、事务封装、NULL、注入和 N+1。Go 实验实际使用 1.27.2、Linux/amd64，配一个完整可控测试驱动，不连接真实数据库。它验证标准库的资源生命周期，不能证明 PostgreSQL/MySQL 的隔离、网络取消或 SQL 执行保证。另用实际 SQLite 3.53.1 做参数绑定反例，单独标明边界。全部程序与测试在文末。

DB 是池，Tx 才是一条事务

对象	表示什么	生命周期
`*sql.DB`	并发安全的句柄，管理连接池	通常应用启动时创建并复用，退出时关闭
`*sql.Conn`	从池里专门借出的一条连接	用完 Close，归还池或按条件关闭底层连接
`*sql.Tx`	绑定一条连接的事务	必须 Commit 或 Rollback，结束后不能继续使用
`*sql.Stmt`	准备好的语句句柄	用完关闭；范围与创建它的 DB、Conn 或 Tx 有关
`*sql.Rows`	仍在读取的结果集	读完或显式 Close；还要检查 Err

sql.Open 常常只是验证参数、建立句柄，不保证已经成功连上服务器。启动时可用带超时的 PingContext 检查可达性，但它只说明那一刻的结果。支持哪个数据库、DSN 如何解释、占位符用什么，取决于具体 driver。^[1]^[2]

正在绘制图示…

查看图源代码

flowchart LR
    A["多个 goroutine"] --> D["长期复用的 sql.DB"]
    D --> P["连接池"]
    P --> Q["Query"]
    P --> X["Exec"]
    X --> Y["执行完成"]
    Y --> P
    P --> T["BeginTx 借出连接"]
    T --> O["Tx 上的全部操作"]
    O --> E["Commit / Rollback"]
    E --> P
    Q --> R["Rows 读完或关闭"]
    R --> P

原始尺寸下可横向或纵向滚动；按 Escape 关闭。

正在绘制图示…

查看图源代码

flowchart LR
    A["多个 goroutine"] --> D["长期复用的 sql.DB"]
    D --> P["连接池"]
    P --> Q["Query"]
    P --> X["Exec"]
    X --> Y["执行完成"]
    Y --> P
    P --> T["BeginTx 借出连接"]
    T --> O["Tx 上的全部操作"]
    O --> E["Commit / Rollback"]
    E --> P
    Q --> R["Rows 读完或关闭"]
    R --> P

不要每个 HTTP 请求都 Open 一次，否则可能得到许多独立连接池。需要固定同一会话的状态，也不能在两次 db.Exec 之间假定底层连接相同。事务用 Tx；其他明确需要专用会话的操作用 Conn。^[3]

四个连接池参数，各管一件事

设置	作用	零值或默认行为	容易误会的地方
SetMaxOpenConns(n)	限制最大打开连接数，含空闲与在用	n≤0 不设上限	调大可能把数据库压入争用，不一定更快
SetMaxIdleConns(n)	保留可复用的空闲连接数	1.27.2 文档当前默认 2；n≤0 不保留空闲	不是最大并发数，也不预建这么多连接
SetConnMaxLifetime(d)	限制连接可复用的最大年龄	d≤0 不因年龄关闭；可在复用前懒关闭	不等于查询超时，不会到点强杀工作中的事务
SetConnMaxIdleTime(d)	限制连接空闲时间	d≤0 不因空闲时长关闭	与 lifetime 的计时起点不同

有限的 MaxOpenConns 也限制 MaxIdleConns。生命周期设置有助于避开基础设施的连接寿命限制，但不能替代断线处理或事务重试。^[4]

连接预算要按全部应用实例一起算。假设 20 个实例各允许 20 条连接，池预算就是 400，还没算后台任务和管理连接。这是核算例子，不是推荐配置。先给数据库留出预算，再按实测并发与延迟调整。

观察 DB.Stats() 的 OpenConnections、InUse、Idle、WaitCount、WaitDuration，以及因空闲或年龄关闭的计数。累计值要看窗口差值；等待上涨可能是池小，也可能是 SQL 变慢、事务变长、Rows 没关闭。直接扩大池可能让后两种问题更严重。^[5]

实际把池压到一条连接

测试先 SetMaxOpenConns(1)，再通过 db.Conn 借走唯一连接，然后发一个 20 毫秒期限的 ExecContext。等待超时后归还专用连接，再执行一次操作。完整测试检查每步错误，实际输出：

text

pool: timeout=true wait=1 inuse=0 idle=1

这次超时发生在拿连接时，测试驱动根本没有执行等待中的操作。held.Close() 归还连接后，下一次调用才继续。这解释了一个常见卡死：事务占着唯一连接，回调里又调用 db.Query，第二次借连接会等待自己释放第一条。即便数据库没有死锁，应用也能卡住自己。^[3:1]

另一个实验将 lifetime 设为 1 毫秒，持有连接后等待至少 3 毫秒，归还并再次操作，确认过期连接被淘汰：

text

lifetime: expired connection retired=true

这只验证当前标准库的归还/复用路径，没有验证代理断开、真实服务端连接寿命或准时定时关闭。

Context 管期限，Rows 管结果资源

把请求 ctx 传到 QueryContext、ExecContext、BeginTx。数据库需要更短预算时，从请求 ctx 派生超时，并调用返回的 cancel。上层取消可以传播下来，但真实驱动是否及时终止服务端查询、连接能否继续复用，要查驱动并做集成测试。^[6]^[2:1]

操作	入口	错误在哪里取
不返回结果行的写入	ExecContext	返回的 error；需要时检查 RowsAffected 的 error
一行结果	QueryRowContext 后 Scan	Scan；查不到是 sql.ErrNoRows
多行结果	QueryContext → Next → Scan	初始错误、每次 Scan 错误、循环后的 Rows.Err
确认只有一行	让查询键受唯一约束保护	QueryRow 多行时只取第一行，不替你验证唯一性

Next 返回 false 既可能是结束，也可能是读取失败。下面这段收尾不能省：^[7]

for rows.Next() {
    var name string
    if err := rows.Scan(&name); err != nil {
        return nil, err
    }
    names = append(names, name)
}
if err := rows.Err(); err != nil {
    return nil, err
}

测试驱动先返回 Ada，下一次读取故意报 stream interrupted。完整函数返回错误，没有把半份名单当作成功结果。实际输出的相关部分为：

text

rows: late-error=true inuse=0; null-string=false zero-valid=true

InUse 为零确认此错误路径没有继续占用连接。获得 Rows 后立即 defer Close，覆盖提前 return 的路径；完整消费单个结果集通常也会自动关闭。循环里打开很多结果集时，要每批及时关闭，别把所有 defer 堆到一个很晚才退出的外层函数。^[8]

sql.RawBytes 引用的内存只在下一次 Next、Scan 或 Close 等相关操作前有效，想留下就复制。扫描成功不表示永久拥有驱动的缓冲区。

事务封装要把 Commit 错误带回去

事务内所有 SQL 都通过 tx 执行。db 仍是连接池，在事务回调里使用它可能借到另一连接，操作不会自动加入当前事务。更不能手写池上的 BEGIN，再期待后面的池操作使用同一连接。^[9]

封装主线很短：

tx, err := db.BeginTx(ctx, nil)
if err != nil { return err }
defer tx.Rollback()
if err := fn(tx); err != nil { return err }
return tx.Commit()

延迟回滚是兜底，提交之后再回滚会得到 ErrTxDone。文末完整 InTx 还保留意外 Rollback 错误，不覆盖原业务错误。回调 panic 时 defer 仍执行，但封装不吞掉 panic。^[10]

实测依次执行成功事务、回调失败事务、另一个成功事务，并尝试使用已结束的 Tx：

text

tx: commits=2 rollbacks=1 reused-after-end=ErrTxDone

驱动只计数 Commit/Rollback 调用，不模拟数据、隔离级别或网络故障。这能发现封装漏收尾，不能证明 InnoDB 锁行为。

BeginTx 可以传 TxOptions 选择隔离级别和只读，驱动不支持时可能报错。ctx 覆盖事务直到结束，取消会触发标准库回滚处理，应用仍须检查 Commit。发送 COMMIT 后连接断开，结果可能不明；盲目重做扣款有重复风险，需要业务幂等和结果核查。死锁、序列化失败的重试则要重跑完整事务中的读取、判断和写入，并限制总时间。^[10:1]^[9:1]

预处理语句和参数绑定，别混成一件事

PrepareContext 返回可复用的 Stmt，随后传参数执行。准备于 DB 的语句可能在多个底层连接上分别准备；准备于 Tx 或 Conn 的语句受那条连接的生命周期限制。不能假定一个 Stmt 在服务器上永远只占一份资源。^[11]^[12]

反复执行相同 SQL 时，预处理可能减少部分解析或协议成本，但收益要测。一次性查询不需要为安全额外手动 Prepare：db.QueryContext(ctx, query, args...) 正确绑定参数，也将语句和值分开交给驱动。^[13]

测试把带引号、OR 和注释符的输入交给预处理语句，驱动实际收到：

text

binding: query="SELECT id FROM users WHERE name = ?" value="' OR 1=1 --"

SQL 文本没有被改写，输入是一项独立参数。测试驱动不解析 SQL，这条输出本身不能证明真实数据库的防注入实现。另在 SQLite 3.53.1 内存数据库执行完整负例：三名用户，拼接查询返回 3 行，绑定同一输入返回 0 行：

text

SQLite 3.53.1: concatenated=3 bound=0

补充实验使用 Python/SQLite，不能冒充 Go MySQL 驱动测试；它观察的是“输入改变了语句结构”与“输入只是一个待匹配值”的区别。^[14]

占位符不能代表任意 SQL 结构

值占位符不是文本替换器。MySQL 常用问号，PostgreSQL 驱动通常用 $1 等；表名、列名、ASC/DESC 不能靠一个值参数随意替换。动态排序应把用户选项映射到有限的、预先写好的 SQL 片段，其他选项拒绝。^[13:1]^[15]

输入	合适处理	不合适处理
用户名、ID、金额、日期	绑定参数并做业务校验	fmt.Sprintf 或加号拼接 SQL
排序列、方向	白名单映射到固定标识符/关键字	直接追加用户原文
IN 中的多个值	对应数量占位符，逐项传参；或使用受支持的数组方案	把逗号文本作为一个标量，期待自动展开
LIKE 搜索词	绑定参数；要字面匹配时按方言处理通配符	以为防注入等于百分号、下划线不再有通配意义

安全绑定与访问控制、限额是不同防线。合法的巨大查询仍能占用很多资源，还需要结果上限、分页和请求预算。

sql.Null* 让零值与缺失值分开

NullString、NullInt64、NullBool、NullTime 等以值字段加 Valid 表示 SQL NULL。Valid 为 false 时，不能把内部零值当作业务值。Go 1.22 起还有 sql.Null[T]，值字段叫 V，T 应是 driver.Value 接受的类型。^[16]

var s sql.NullString
var n sql.Null[int64]
s.Scan(nil)
n.Scan(int64(0))
// 文末测试实际检查这两次 Scan 的错误。

本次实测 s.Valid 为 false，n.Valid 为 true：数字零确实存在，字符串值缺失。用 COALESCE 给默认值会有意合并这些含义，应该先确认业务需要。

数据库可空类型也不自动等于 API 的 JSON 协议。对外输出 null、省略字段还是默认值，需要通过 DTO 转换或自定义编码明确处理，不能直接拿数据库层结构代替协议设计。

N+1：每次很快，合起来仍有 N 次往返

先查 N 名客户，再逐个查订单，就是 1+N 次调用。网络延迟、解析、连接竞争被重复支付，ORM 的延迟加载或循环中的便捷方法可能掩盖这种结构。减少它的关键是按需要成批取数。

正在绘制图示…

查看图源代码

flowchart LR
    A["一次查询父记录"] --> B["N 个父 ID"]
    B --> C["逐个查子记录<br/>额外 N 次往返"]
    B --> D["批量查子记录<br/>按父 ID 组装"]
    B --> E["JOIN 配对结果<br/>还原父子对象"]

原始尺寸下可横向或纵向滚动；按 Escape 关闭。

正在绘制图示…

查看图源代码

flowchart LR
    A["一次查询父记录"] --> B["N 个父 ID"]
    B --> C["逐个查子记录<br/>额外 N 次往返"]
    B --> D["批量查子记录<br/>按父 ID 组装"]
    B --> E["JOIN 配对结果<br/>还原父子对象"]

本次驱动返回三个父 ID，计数实际输出：

text

query-count: n+1=4 batched=2

它只验证代码调用次数，不测真实 SQL、网络耗时或批量方案吞吐。

改法	好处	代价和坑
JOIN	一次往返，数据库配对	父字段重复；父分页与连接后 LIMIT 不同
先查父，再批量查子	通常两次往返，组装清楚	ID 太多要分批；两次查询间可能有变化
预加载	框架帮助批量读取	必须检查实际 SQL，名称不能保证策略
只查聚合	避免传不需要的明细	仅适用于业务只要数量、总额等结果

大 IN、宽 JOIN 和一次读全量结果也有成本。分页要稳定排序，跨两次读取的一致性按业务选择事务或接受变化。请求级别的查询次数、总行数、数据库时间，再配合池等待增量，比只看单条 SQL 快不快更有用。

实验范围与检查结果

go test -v -count=1 ./...、go test -race -count=1 ./...、go vet ./... 均实际通过。测试覆盖池等待/归还/过期连接、事务收尾、迟到的 Rows 错误、NULL、参数分离和 N+1 次数。race 通过只说明执行路径没有报告竞争，不证明任意业务并发正确。

没有测试真实网络驱动、数据库锁冲突、取消协议、服务端预处理缓存或 N+1 性能。测试驱动是检查标准库边界的工具，之后仍要用目标引擎做集成测试。

参考资料

完整复现材料

把 Go 文件放在同一目录，使用 Go 1.27.2。仅依赖标准库，不连接数据库；驱动中的 names、parents 等是受控测试命令，不能在真实数据库执行。事务驱动只计数，不实现持久存储。

go test -v -count=1 ./...
go test -race -count=1 ./...
go vet ./...
python3 injection.py

最后一条用 Python 自带 sqlite3，只建立内存测试表；故意不安全的拼接仅作为负例，不可复制到应用代码。

go.mod

module example.com/storageaccess

go 1.27

access.go

代码块 · 43 行 · 点击收起 / 展开

package storageaccess

import (
	"context"
	"database/sql"
	"errors"
)

func InTx(ctx context.Context, db *sql.DB, fn func(*sql.Tx) error) (err error) {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() {
		if e := tx.Rollback(); e != nil && !errors.Is(e, sql.ErrTxDone) {
			err = errors.Join(err, e)
		}
	}()
	if err = fn(tx); err != nil {
		return err
	}
	return tx.Commit()
}

func Names(ctx context.Context, db *sql.DB) ([]string, error) {
	rows, err := db.QueryContext(ctx, "names")
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var names []string
	for rows.Next() {
		var name string
		if err := rows.Scan(&name); err != nil {
			return nil, err
		}
		names = append(names, name)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return names, nil
}

access_test.go

代码块 · 308 行 · 点击收起 / 展开

package storageaccess

import (
	"context"
	"database/sql"
	"database/sql/driver"
	"errors"
	"fmt"
	"io"
	"reflect"
	"sync"
	"testing"
	"time"
)

var errStream = errors.New("stream interrupted")

type event struct {
	query string
	args  []driver.NamedValue
}
type state struct {
	mu                                sync.Mutex
	opens, closes, commits, rollbacks int
	events                            []event
}

func (s *state) record(q string, args []driver.NamedValue) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.events = append(s.events, event{q, append([]driver.NamedValue(nil), args...)})
}

type connector struct{ s *state }

func (c connector) Connect(ctx context.Context) (driver.Conn, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	c.s.mu.Lock()
	c.s.opens++
	c.s.mu.Unlock()
	return &connection{s: c.s}, nil
}
func (c connector) Driver() driver.Driver { return drv{c.s} }

type drv struct{ s *state }

func (d drv) Open(string) (driver.Conn, error) { return connector{d.s}.Connect(context.Background()) }

type connection struct{ s *state }

func (c *connection) Prepare(q string) (driver.Stmt, error) { return &statement{c, q}, nil }
func (c *connection) Close() error                          { c.s.mu.Lock(); defer c.s.mu.Unlock(); c.s.closes++; return nil }
func (c *connection) Begin() (driver.Tx, error) {
	return c.BeginTx(context.Background(), driver.TxOptions{})
}
func (c *connection) BeginTx(ctx context.Context, _ driver.TxOptions) (driver.Tx, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	return &transaction{c.s}, nil
}
func (c *connection) ExecContext(ctx context.Context, q string, a []driver.NamedValue) (driver.Result, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	c.s.record(q, a)
	return driver.RowsAffected(1), nil
}
func (c *connection) QueryContext(ctx context.Context, q string, a []driver.NamedValue) (driver.Rows, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	c.s.record(q, a)
	if q == "names" {
		return &rows{data: [][]driver.Value{{"Ada"}}, late: errStream}, nil
	}
	if q == "parents" {
		return &rows{data: [][]driver.Value{{int64(1)}, {int64(2)}, {int64(3)}}}, nil
	}
	return &rows{data: [][]driver.Value{{int64(1)}}}, nil
}

type transaction struct{ s *state }

func (t *transaction) Commit() error { t.s.mu.Lock(); defer t.s.mu.Unlock(); t.s.commits++; return nil }
func (t *transaction) Rollback() error {
	t.s.mu.Lock()
	defer t.s.mu.Unlock()
	t.s.rollbacks++
	return nil
}

type rows struct {
	data [][]driver.Value
	i    int
	late error
}

func (*rows) Columns() []string { return []string{"value"} }
func (*rows) Close() error      { return nil }
func (r *rows) Next(dest []driver.Value) error {
	if r.i < len(r.data) {
		copy(dest, r.data[r.i])
		r.i++
		return nil
	}
	if r.late != nil {
		return r.late
	}
	return io.EOF
}

type statement struct {
	c *connection
	q string
}

func (*statement) Close() error  { return nil }
func (*statement) NumInput() int { return -1 }
func (s *statement) Exec(a []driver.Value) (driver.Result, error) {
	return s.ExecContext(context.Background(), named(a))
}
func (s *statement) Query(a []driver.Value) (driver.Rows, error) {
	return s.QueryContext(context.Background(), named(a))
}
func (s *statement) ExecContext(ctx context.Context, a []driver.NamedValue) (driver.Result, error) {
	return s.c.ExecContext(ctx, s.q, a)
}
func (s *statement) QueryContext(ctx context.Context, a []driver.NamedValue) (driver.Rows, error) {
	return s.c.QueryContext(ctx, s.q, a)
}
func named(a []driver.Value) []driver.NamedValue {
	out := make([]driver.NamedValue, len(a))
	for i, v := range a {
		out[i] = driver.NamedValue{Ordinal: i + 1, Value: v}
	}
	return out
}
func openDB(t *testing.T) (*sql.DB, *state) {
	t.Helper()
	s := &state{}
	db := sql.OpenDB(connector{s})
	t.Cleanup(func() { db.Close() })
	return db, s
}
func TestPoolWaitAndRelease(t *testing.T) {
	db, _ := openDB(t)
	db.SetMaxOpenConns(1)
	db.SetMaxIdleConns(1)
	held, err := db.Conn(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
	defer cancel()
	_, err = db.ExecContext(ctx, "waiting")
	if !errors.Is(err, context.DeadlineExceeded) {
		t.Fatal(err)
	}
	if err = held.Close(); err != nil {
		t.Fatal(err)
	}
	if _, err = db.ExecContext(context.Background(), "released"); err != nil {
		t.Fatal(err)
	}
	st := db.Stats()
	if st.WaitCount != 1 || st.InUse != 0 || st.Idle != 1 {
		t.Fatalf("%+v", st)
	}
	fmt.Printf("pool: timeout=%v wait=%d inuse=%d idle=%d\n", errors.Is(ctx.Err(), context.DeadlineExceeded), st.WaitCount, st.InUse, st.Idle)
}
func TestLifetime(t *testing.T) {
	db, _ := openDB(t)
	db.SetMaxOpenConns(1)
	db.SetMaxIdleConns(1)
	db.SetConnMaxLifetime(time.Millisecond)
	held, err := db.Conn(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	time.Sleep(3 * time.Millisecond)
	if err = held.Close(); err != nil {
		t.Fatal(err)
	}
	if _, err = db.ExecContext(context.Background(), "new"); err != nil {
		t.Fatal(err)
	}
	if db.Stats().MaxLifetimeClosed < 1 {
		t.Fatal("expired connection not retired")
	}
	fmt.Println("lifetime: expired connection retired=true")
}
func TestTxLifecycle(t *testing.T) {
	db, s := openDB(t)
	ctx := context.Background()
	want := errors.New("business rejected")
	if err := InTx(ctx, db, func(tx *sql.Tx) error { _, e := tx.ExecContext(ctx, "change"); return e }); err != nil {
		t.Fatal(err)
	}
	if err := InTx(ctx, db, func(*sql.Tx) error { return want }); !errors.Is(err, want) {
		t.Fatal(err)
	}
	var done *sql.Tx
	if err := InTx(ctx, db, func(tx *sql.Tx) error { done = tx; return nil }); err != nil {
		t.Fatal(err)
	}
	if _, err := done.ExecContext(ctx, "too late"); !errors.Is(err, sql.ErrTxDone) {
		t.Fatal(err)
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.commits != 2 || s.rollbacks != 1 {
		t.Fatal(s.commits, s.rollbacks)
	}
	fmt.Printf("tx: commits=%d rollbacks=%d reused-after-end=ErrTxDone\n", s.commits, s.rollbacks)
}
func TestRowsErrorAndNull(t *testing.T) {
	db, _ := openDB(t)
	names, err := Names(context.Background(), db)
	if names != nil || !errors.Is(err, errStream) || db.Stats().InUse != 0 {
		t.Fatal(names, err, db.Stats())
	}
	lateError := errors.Is(err, errStream)
	var s sql.NullString
	var n sql.Null[int64]
	if err = s.Scan(nil); err != nil {
		t.Fatal(err)
	}
	if err = n.Scan(int64(0)); err != nil {
		t.Fatal(err)
	}
	if s.Valid || !n.Valid || n.V != 0 {
		t.Fatal(s, n)
	}
	fmt.Printf("rows: late-error=%v inuse=%d; null-string=%v zero-valid=%v\n", lateError, db.Stats().InUse, s.Valid, n.Valid)
}
func TestPreparedKeepsValuesSeparate(t *testing.T) {
	db, s := openDB(t)
	ctx := context.Background()
	query := "SELECT id FROM users WHERE name = ?"
	value := "' OR 1=1 --"
	stmt, err := db.PrepareContext(ctx, query)
	if err != nil {
		t.Fatal(err)
	}
	defer stmt.Close()
	var id int
	if err = stmt.QueryRowContext(ctx, value).Scan(&id); err != nil {
		t.Fatal(err)
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if len(s.events) != 1 || s.events[0].query != query || len(s.events[0].args) != 1 || s.events[0].args[0].Value != value {
		t.Fatal(s.events)
	}
	fmt.Printf("binding: query=%q value=%q\n", s.events[0].query, s.events[0].args[0].Value)
}
func TestNPlusOneCount(t *testing.T) {
	db, s := openDB(t)
	ctx := context.Background()
	getParents := func() []int {
		rr, err := db.QueryContext(ctx, "parents")
		if err != nil {
			t.Fatal(err)
		}
		defer rr.Close()
		var ids []int
		for rr.Next() {
			var id int
			if err := rr.Scan(&id); err != nil {
				t.Fatal(err)
			}
			ids = append(ids, id)
		}
		if err := rr.Err(); err != nil {
			t.Fatal(err)
		}
		return ids
	}
	ids := getParents()
	if !reflect.DeepEqual(ids, []int{1, 2, 3}) {
		t.Fatal(ids)
	}
	for _, id := range ids {
		var n int
		if err := db.QueryRowContext(ctx, "child", id).Scan(&n); err != nil {
			t.Fatal(err)
		}
	}
	s.mu.Lock()
	before := len(s.events)
	s.events = nil
	s.mu.Unlock()
	ids = getParents()
	var n int
	if err := db.QueryRowContext(ctx, "children for all parents", ids[0], ids[1], ids[2]).Scan(&n); err != nil {
		t.Fatal(err)
	}
	s.mu.Lock()
	after := len(s.events)
	s.mu.Unlock()
	if before != 4 || after != 2 {
		t.Fatal(before, after)
	}
	fmt.Printf("query-count: n+1=%d batched=%d\n", before, after)
}

injection.py

python

import sqlite3
with sqlite3.connect(':memory:') as db:
    db.execute('CREATE TABLE users(id INTEGER PRIMARY KEY,name TEXT NOT NULL)')
    db.executemany('INSERT INTO users VALUES(?,?)', [(1,'Ada'),(2,'Bo'),(3,'Cy')])
    value = "' OR 1=1 --"
    # Deliberately unsafe negative fixture, only in this disposable database.
    unsafe = db.execute("SELECT id FROM users WHERE name = '"+value+"'").fetchall()
    safe = db.execute('SELECT id FROM users WHERE name = ?', (value,)).fetchall()
    assert len(unsafe) == 3 and safe == []
    print(f'SQLite {sqlite3.sqlite_version}: concatenated={len(unsafe)} bound={len(safe)}')

Go 1.27.2 DB。 ↩︎
Go 1.27.2 driver；同时以本地同版本 go doc 与源码核对。 ↩︎ ↩︎
Go 官方：Managing connections。 ↩︎ ↩︎
MaxOpenConns、MaxIdleConns、ConnMaxLifetime、ConnMaxIdleTime。 ↩︎
DBStats。 ↩︎
Go 官方：Canceling operations。 ↩︎
Go 官方：Querying for data。 ↩︎
Rows、RawBytes。 ↩︎
Go 官方：Executing transactions。 ↩︎ ↩︎
Tx、BeginTx。 ↩︎ ↩︎
Go 官方：Prepared statements。 ↩︎
Stmt 的连接范围。 ↩︎
Go 官方：Avoiding SQL injection。 ↩︎ ↩︎
Python 官方 sqlite3 参数绑定。 ↩︎
PostgreSQL 18 参数表达式。 ↩︎
Null、NullString。 ↩︎

Go 数据访问：管好连接、事务和 SQL 边界 ​

DB 是池，Tx 才是一条事务 ​

四个连接池参数，各管一件事 ​

实际把池压到一条连接 ​

Context 管期限，Rows 管结果资源 ​

事务封装要把 Commit 错误带回去 ​

预处理语句和参数绑定，别混成一件事 ​

占位符不能代表任意 SQL 结构 ​

sql.Null* 让零值与缺失值分开 ​

N+1：每次很快，合起来仍有 N 次往返 ​

实验范围与检查结果 ​

参考资料 ​

完整复现材料 ​

go.mod ​

access.go ​

access_test.go ​

injection.py ​

Go 数据访问：管好连接、事务和 SQL 边界

DB 是池，Tx 才是一条事务

四个连接池参数，各管一件事

实际把池压到一条连接

Context 管期限，Rows 管结果资源

事务封装要把 Commit 错误带回去

预处理语句和参数绑定，别混成一件事

占位符不能代表任意 SQL 结构

sql.Null* 让零值与缺失值分开

N+1：每次很快，合起来仍有 N 次往返

实验范围与检查结果

参考资料

完整复现材料

go.mod

access.go

access_test.go

injection.py