golang-database-patterns

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Go Database Patterns

Go数据库模式

Overview

概述

Go's database ecosystem provides multiple layers of abstraction for SQL database integration. From the standard library's

database/sql

to enhanced libraries like

sqlx

and PostgreSQL-optimized

pgx

, developers can choose the right tool for their performance and ergonomics needs.

Key Features:

🔌 database/sql: Standard interface for any SQL database
🚀 sqlx: Convenience methods with struct scanning and named queries
🐘 pgx: PostgreSQL-native driver with maximum performance
📦 Repository Pattern: Interface-based data access for testability
🔄 Migrations: Schema versioning with golang-migrate
⚡ Connection Pooling: Production-ready connection management
🔒 Transaction Safety: Context-aware transaction handling

Go的数据库生态系统为SQL数据库集成提供了多层抽象。从标准库的

database/sql

到增强库如

sqlx

，再到针对PostgreSQL优化的

pgx

，开发者可以根据性能和易用性需求选择合适的工具。

核心特性：

🔌 database/sql: 适用于所有SQL数据库的标准接口
🚀 sqlx: 提供结构体扫描和命名查询等便捷方法
🐘 pgx: 原生PostgreSQL驱动，性能拉满
📦 仓库模式: 基于接口的数据访问方式，提升可测试性
🔄 迁移: 使用golang-migrate进行Schema版本管理
⚡ 连接池: 生产环境就绪的连接管理方案
🔒 事务安全: 支持上下文感知的事务处理

When to Use This Skill

适用场景

Activate this skill when:

Building CRUD operations with type safety
Implementing data access layers for web services
Managing database schema evolution across environments
Optimizing database connection pooling for production
Testing database code with mock repositories
Handling concurrent database access patterns
Migrating from ORMs to SQL-first approaches
Integrating PostgreSQL-specific features (COPY, LISTEN/NOTIFY)

在以下场景中可以使用本技能：

构建类型安全的CRUD操作
为Web服务实现数据访问层
跨环境管理数据库Schema演进
针对生产环境优化数据库连接池
使用Mock仓库测试数据库代码
处理并发数据库访问模式
从ORM迁移到SQL优先的开发方式
集成PostgreSQL专属特性（COPY、LISTEN/NOTIFY）

Core Database Libraries

核心数据库库

Decision Tree: Choosing Your Database Library

决策树：选择合适的数据库库

┌─────────────────────────────────────┐
│ What database are you using?       │
└──────────────┬──────────────────────┘
               │
    ┌──────────┴──────────┐
    │                     │
PostgreSQL            Other SQL DB
    │                     │
    ▼                     ▼
┌─────────────────┐   Use database/sql
│ Need max perf?  │   + sqlx for convenience
└─────┬───────────┘
      │
   ┌──┴──┐
  Yes    No
   │      │
  pgx   sqlx + pq driver

Use database/sql when:

Working with any SQL database (MySQL, SQLite, PostgreSQL, etc.)
Need database portability
Want standard library stability with no dependencies

Use sqlx when:

Want convenience methods (Get, Select, StructScan)
Need named parameter queries
Using IN clause expansion
Prefer less boilerplate than database/sql

Use pgx when:

PostgreSQL-only application
Need maximum performance (30-50% faster than lib/pq)
Want advanced PostgreSQL features (COPY, LISTEN/NOTIFY, prepared statement caching)
Building high-throughput systems

┌─────────────────────────────────────┐
│ 你使用的是哪种数据库？              │
└──────────────┬──────────────────────┘
               │
    ┌──────────┴──────────┐
    │                     │
PostgreSQL            其他SQL数据库
    │                     │
    ▼                     ▼
┌─────────────────┐   使用database/sql
│ 需要极致性能？  │   + sqlx提升便捷性
└─────┬───────────┘
      │
   ┌──┴──┐
  是    否
   │      │
  pgx   sqlx + pq驱动

使用database/sql的场景：

操作任意SQL数据库（MySQL、SQLite、PostgreSQL等）
需要数据库可移植性
希望依赖标准库，保持稳定性且无额外依赖

使用sqlx的场景：

需要便捷方法（Get、Select、StructScan）
需使用命名参数查询
要处理IN子句扩展
相比database/sql，希望减少样板代码

使用pgx的场景：

仅使用PostgreSQL的应用
需要极致性能（比lib/pq快30-50%）
要使用PostgreSQL高级特性（COPY、LISTEN/NOTIFY、预编译语句缓存）
构建高吞吐量系统

database/sql: The Standard Foundation

database/sql：标准基础库

Core Concepts:

package main

import (
    "context"
    "database/sql"
    "time"

    _ "github.com/lib/pq" // PostgreSQL driver
)

func setupDB(dsn string) (*sql.DB, error) {
    db, err := sql.Open("postgres", dsn)
    if err != nil {
        return nil, err
    }

    // Connection pooling configuration
    db.SetMaxOpenConns(25)                 // Max open connections
    db.SetMaxIdleConns(5)                  // Max idle connections
    db.SetConnMaxLifetime(5 * time.Minute) // Max connection lifetime
    db.SetConnMaxIdleTime(1 * time.Minute) // Max idle time

    // Verify connection
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    if err := db.PingContext(ctx); err != nil {
        return nil, err
    }

    return db, nil
}

Key Patterns:

// Query single row
func GetUserByID(ctx context.Context, db *sql.DB, id int) (*User, error) {
    var user User
    query := `SELECT id, name, email, created_at FROM users WHERE id = $1`

    err := db.QueryRowContext(ctx, query, id).Scan(
        &user.ID, &user.Name, &user.Email, &user.CreatedAt,
    )

    if err == sql.ErrNoRows {
        return nil, ErrUserNotFound // Custom error
    }
    if err != nil {
        return nil, fmt.Errorf("query user: %w", err)
    }

    return &user, nil
}

// Query multiple rows
func ListActiveUsers(ctx context.Context, db *sql.DB) ([]User, error) {
    query := `SELECT id, name, email, created_at FROM users WHERE active = true`

    rows, err := db.QueryContext(ctx, query)
    if err != nil {
        return nil, fmt.Errorf("query users: %w", err)
    }
    defer rows.Close() // CRITICAL: Always close rows

    var users []User
    for rows.Next() {
        var user User
        if err := rows.Scan(&user.ID, &user.Name, &user.Email, &user.CreatedAt); err != nil {
            return nil, fmt.Errorf("scan user: %w", err)
        }
        users = append(users, user)
    }

    // Check for errors during iteration
    if err := rows.Err(); err != nil {
        return nil, fmt.Errorf("iterate users: %w", err)
    }

    return users, nil
}

核心概念：

package main

import (
    "context"
    "database/sql"
    "time"

    _ "github.com/lib/pq" // PostgreSQL驱动
)

func setupDB(dsn string) (*sql.DB, error) {
    db, err := sql.Open("postgres", dsn)
    if err != nil {
        return nil, err
    }

    // 连接池配置
    db.SetMaxOpenConns(25)                 // 最大打开连接数
    db.SetMaxIdleConns(5)                  // 最大空闲连接数
    db.SetConnMaxLifetime(5 * time.Minute) // 连接最大生命周期
    db.SetConnMaxIdleTime(1 * time.Minute) // 连接最大空闲时间

    // 验证连接
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    if err := db.PingContext(ctx); err != nil {
        return nil, err
    }

    return db, nil
}

核心模式：

// 查询单行数据
func GetUserByID(ctx context.Context, db *sql.DB, id int) (*User, error) {
    var user User
    query := `SELECT id, name, email, created_at FROM users WHERE id = $1`

    err := db.QueryRowContext(ctx, query, id).Scan(
        &user.ID, &user.Name, &user.Email, &user.CreatedAt,
    )

    if err == sql.ErrNoRows {
        return nil, ErrUserNotFound // 自定义错误
    }
    if err != nil {
        return nil, fmt.Errorf("query user: %w", err)
    }

    return &user, nil
}

// 查询多行数据
func ListActiveUsers(ctx context.Context, db *sql.DB) ([]User, error) {
    query := `SELECT id, name, email, created_at FROM users WHERE active = true`

    rows, err := db.QueryContext(ctx, query)
    if err != nil {
        return nil, fmt.Errorf("query users: %w", err)
    }
    defer rows.Close() // 关键：必须关闭rows

    var users []User
    for rows.Next() {
        var user User
        if err := rows.Scan(&user.ID, &user.Name, &user.Email, &user.CreatedAt); err != nil {
            return nil, fmt.Errorf("scan user: %w", err)
        }
        users = append(users, user)
    }

    // 检查迭代过程中的错误
    if err := rows.Err(); err != nil {
        return nil, fmt.Errorf("iterate users: %w", err)
    }

    return users, nil
}

sqlx: Ergonomic Extensions

sqlx：易用性扩展库

Installation:

bash

go get github.com/jmoiron/sqlx

Core Features:

package main

import (
    "context"

    "github.com/jmoiron/sqlx"
    _ "github.com/lib/pq"
)

type User struct {
    ID        int       `db:"id"`
    Name      string    `db:"name"`
    Email     string    `db:"email"`
    CreatedAt time.Time `db:"created_at"`
}

// Get single struct
func GetUserByID(ctx context.Context, db *sqlx.DB, id int) (*User, error) {
    var user User
    query := `SELECT id, name, email, created_at FROM users WHERE id = $1`

    err := db.GetContext(ctx, &user, query, id)
    if err == sql.ErrNoRows {
        return nil, ErrUserNotFound
    }
    return &user, err
}

// Select multiple structs
func ListUsers(ctx context.Context, db *sqlx.DB, limit int) ([]User, error) {
    var users []User
    query := `SELECT id, name, email, created_at FROM users LIMIT $1`

    err := db.SelectContext(ctx, &users, query, limit)
    return users, err
}

// Named queries
func FindUsersByName(ctx context.Context, db *sqlx.DB, name string) ([]User, error) {
    var users []User
    query := `SELECT * FROM users WHERE name LIKE :name || '%'`

    nstmt, err := db.PrepareNamedContext(ctx, query)
    if err != nil {
        return nil, err
    }
    defer nstmt.Close()

    err = nstmt.SelectContext(ctx, &users, map[string]interface{}{"name": name})
    return users, err
}

// IN clause expansion
func GetUsersByIDs(ctx context.Context, db *sqlx.DB, ids []int) ([]User, error) {
    var users []User
    query, args, err := sqlx.In(`SELECT * FROM users WHERE id IN (?)`, ids)
    if err != nil {
        return nil, err
    }

    // Rebind for PostgreSQL ($1, $2, ...) vs MySQL (?, ?, ...)
    query = db.Rebind(query)

    err = db.SelectContext(ctx, &users, query, args...)
    return users, err
}

安装：

bash

go get github.com/jmoiron/sqlx

核心特性：

package main

import (
    "context"

    "github.com/jmoiron/sqlx"
    _ "github.com/lib/pq"
)

type User struct {
    ID        int       `db:"id"`
    Name      string    `db:"name"`
    Email     string    `db:"email"`
    CreatedAt time.Time `db:"created_at"`
}

// 获取单个结构体
func GetUserByID(ctx context.Context, db *sqlx.DB, id int) (*User, error) {
    var user User
    query := `SELECT id, name, email, created_at FROM users WHERE id = $1`

    err := db.GetContext(ctx, &user, query, id)
    if err == sql.ErrNoRows {
        return nil, ErrUserNotFound
    }
    return &user, err
}

// 选择多个结构体
func ListUsers(ctx context.Context, db *sqlx.DB, limit int) ([]User, error) {
    var users []User
    query := `SELECT id, name, email, created_at FROM users LIMIT $1`

    err := db.SelectContext(ctx, &users, query, limit)
    return users, err
}

// 命名查询
func FindUsersByName(ctx context.Context, db *sqlx.DB, name string) ([]User, error) {
    var users []User
    query := `SELECT * FROM users WHERE name LIKE :name || '%'`

    nstmt, err := db.PrepareNamedContext(ctx, query)
    if err != nil {
        return nil, err
    }
    defer nstmt.Close()

    err = nstmt.SelectContext(ctx, &users, map[string]interface{}{"name": name})
    return users, err
}

// IN子句扩展
func GetUsersByIDs(ctx context.Context, db *sqlx.DB, ids []int) ([]User, error) {
    var users []User
    query, args, err := sqlx.In(`SELECT * FROM users WHERE id IN (?)`, ids)
    if err != nil {
        return nil, err
    }

    // 为PostgreSQL重新绑定参数（$1, $2,...），MySQL则是（?, ?, ...）
    query = db.Rebind(query)

    err = db.SelectContext(ctx, &users, query, args...)
    return users, err
}

pgx: PostgreSQL-Native Performance

pgx：PostgreSQL原生高性能库

Installation:

bash

go get github.com/jackc/pgx/v5
go get github.com/jackc/pgx/v5/pgxpool

Connection Pool Setup:

package main

import (
    "context"
    "fmt"

    "github.com/jackc/pgx/v5/pgxpool"
)

func setupPgxPool(ctx context.Context, dsn string) (*pgxpool.Pool, error) {
    config, err := pgxpool.ParseConfig(dsn)
    if err != nil {
        return nil, fmt.Errorf("parse config: %w", err)
    }

    // Connection pool tuning
    config.MaxConns = 25
    config.MinConns = 5
    config.MaxConnLifetime = 1 * time.Hour
    config.MaxConnIdleTime = 30 * time.Minute
    config.HealthCheckPeriod = 1 * time.Minute

    pool, err := pgxpool.NewWithConfig(ctx, config)
    if err != nil {
        return nil, fmt.Errorf("create pool: %w", err)
    }

    // Verify connectivity
    if err := pool.Ping(ctx); err != nil {
        return nil, fmt.Errorf("ping: %w", err)
    }

    return pool, nil
}

Query Patterns:

// Query single row
func GetUser(ctx context.Context, pool *pgxpool.Pool, id int) (*User, error) {
    var user User
    query := `SELECT id, name, email, created_at FROM users WHERE id = $1`

    err := pool.QueryRow(ctx, query, id).Scan(
        &user.ID, &user.Name, &user.Email, &user.CreatedAt,
    )

    if err == pgx.ErrNoRows {
        return nil, ErrUserNotFound
    }
    return &user, err
}

// Batch operations (pgx-specific optimization)
func BatchInsertUsers(ctx context.Context, pool *pgxpool.Pool, users []User) error {
    batch := &pgx.Batch{}
    query := `INSERT INTO users (name, email) VALUES ($1, $2)`

    for _, user := range users {
        batch.Queue(query, user.Name, user.Email)
    }

    results := pool.SendBatch(ctx, batch)
    defer results.Close()

    for range users {
        _, err := results.Exec()
        if err != nil {
            return fmt.Errorf("batch insert: %w", err)
        }
    }

    return nil
}

// COPY for bulk inserts (10x faster than INSERT)
func BulkCopyUsers(ctx context.Context, pool *pgxpool.Pool, users []User) error {
    _, err := pool.CopyFrom(
        ctx,
        pgx.Identifier{"users"},
        []string{"name", "email"},
        pgx.CopyFromSlice(len(users), func(i int) ([]interface{}, error) {
            return []interface{}{users[i].Name, users[i].Email}, nil
        }),
    )
    return err
}

安装：

bash

go get github.com/jackc/pgx/v5
go get github.com/jackc/pgx/v5/pgxpool

连接池设置：

package main

import (
    "context"
    "fmt"

    "github.com/jackc/pgx/v5/pgxpool"
)

func setupPgxPool(ctx context.Context, dsn string) (*pgxpool.Pool, error) {
    config, err := pgxpool.ParseConfig(dsn)
    if err != nil {
        return nil, fmt.Errorf("parse config: %w", err)
    }

    // 连接池调优
    config.MaxConns = 25
    config.MinConns = 5
    config.MaxConnLifetime = 1 * time.Hour
    config.MaxConnIdleTime = 30 * time.Minute
    config.HealthCheckPeriod = 1 * time.Minute

    pool, err := pgxpool.NewWithConfig(ctx, config)
    if err != nil {
        return nil, fmt.Errorf("create pool: %w", err)
    }

    // 验证连接
    if err := pool.Ping(ctx); err != nil {
        return nil, fmt.Errorf("ping: %w", err)
    }

    return pool, nil
}

查询模式：

// 查询单行数据
func GetUser(ctx context.Context, pool *pgxpool.Pool, id int) (*User, error) {
    var user User
    query := `SELECT id, name, email, created_at FROM users WHERE id = $1`

    err := pool.QueryRow(ctx, query, id).Scan(
        &user.ID, &user.Name, &user.Email, &user.CreatedAt,
    )

    if err == pgx.ErrNoRows {
        return nil, ErrUserNotFound
    }
    return &user, err
}

// 批量操作（pgx专属优化）
func BatchInsertUsers(ctx context.Context, pool *pgxpool.Pool, users []User) error {
    batch := &pgx.Batch{}
    query := `INSERT INTO users (name, email) VALUES ($1, $2)`

    for _, user := range users {
        batch.Queue(query, user.Name, user.Email)
    }

    results := pool.SendBatch(ctx, batch)
    defer results.Close()

    for range users {
        _, err := results.Exec()
        if err != nil {
            return fmt.Errorf("batch insert: %w", err)
        }
    }

    return nil
}

// COPY批量插入（比INSERT快10倍）
func BulkCopyUsers(ctx context.Context, pool *pgxpool.Pool, users []User) error {
    _, err := pool.CopyFrom(
        ctx,
        pgx.Identifier{"users"},
        []string{"name", "email"},
        pgx.CopyFromSlice(len(users), func(i int) ([]interface{}, error) {
            return []interface{}{users[i].Name, users[i].Email}, nil
        }),
    )
    return err
}

Repository Pattern Implementation

仓库模式实现

Interface-Based Design

基于接口的设计

package repository

import (
    "context"
    "database/sql"
)

// UserRepository defines data access interface
type UserRepository interface {
    Create(ctx context.Context, user *User) error
    GetByID(ctx context.Context, id int) (*User, error)
    GetByEmail(ctx context.Context, email string) (*User, error)
    Update(ctx context.Context, user *User) error
    Delete(ctx context.Context, id int) error
    List(ctx context.Context, filters ListFilters) ([]User, error)
}

// PostgresUserRepository implements UserRepository
type PostgresUserRepository struct {
    db *sqlx.DB
}

func NewPostgresUserRepository(db *sqlx.DB) *PostgresUserRepository {
    return &PostgresUserRepository{db: db}
}

func (r *PostgresUserRepository) Create(ctx context.Context, user *User) error {
    query := `
        INSERT INTO users (name, email, password_hash)
        VALUES ($1, $2, $3)
        RETURNING id, created_at
    `

    err := r.db.QueryRowContext(
        ctx, query,
        user.Name, user.Email, user.PasswordHash,
    ).Scan(&user.ID, &user.CreatedAt)

    if err != nil {
        return fmt.Errorf("insert user: %w", err)
    }
    return nil
}

func (r *PostgresUserRepository) GetByID(ctx context.Context, id int) (*User, error) {
    var user User
    query := `SELECT id, name, email, created_at, updated_at FROM users WHERE id = $1`

    err := r.db.GetContext(ctx, &user, query, id)
    if err == sql.ErrNoRows {
        return nil, ErrUserNotFound
    }
    if err != nil {
        return nil, fmt.Errorf("get user: %w", err)
    }
    return &user, nil
}

func (r *PostgresUserRepository) Update(ctx context.Context, user *User) error {
    query := `
        UPDATE users
        SET name = $1, email = $2, updated_at = NOW()
        WHERE id = $3
        RETURNING updated_at
    `

    err := r.db.QueryRowContext(
        ctx, query,
        user.Name, user.Email, user.ID,
    ).Scan(&user.UpdatedAt)

    if err == sql.ErrNoRows {
        return ErrUserNotFound
    }
    return err
}

func (r *PostgresUserRepository) Delete(ctx context.Context, id int) error {
    query := `DELETE FROM users WHERE id = $1`
    result, err := r.db.ExecContext(ctx, query, id)
    if err != nil {
        return fmt.Errorf("delete user: %w", err)
    }

    rows, err := result.RowsAffected()
    if err != nil {
        return err
    }

    if rows == 0 {
        return ErrUserNotFound
    }
    return nil
}

package repository

import (
    "context"
    "database/sql"
)

// UserRepository定义数据访问接口
type UserRepository interface {
    Create(ctx context.Context, user *User) error
    GetByID(ctx context.Context, id int) (*User, error)
    GetByEmail(ctx context.Context, email string) (*User, error)
    Update(ctx context.Context, user *User) error
    Delete(ctx context.Context, id int) error
    List(ctx context.Context, filters ListFilters) ([]User, error)
}

// PostgresUserRepository实现UserRepository
type PostgresUserRepository struct {
    db *sqlx.DB
}

func NewPostgresUserRepository(db *sqlx.DB) *PostgresUserRepository {
    return &PostgresUserRepository{db: db}
}

func (r *PostgresUserRepository) Create(ctx context.Context, user *User) error {
    query := `
        INSERT INTO users (name, email, password_hash)
        VALUES ($1, $2, $3)
        RETURNING id, created_at
    `

    err := r.db.QueryRowContext(
        ctx, query,
        user.Name, user.Email, user.PasswordHash,
    ).Scan(&user.ID, &user.CreatedAt)

    if err != nil {
        return fmt.Errorf("insert user: %w", err)
    }
    return nil
}

func (r *PostgresUserRepository) GetByID(ctx context.Context, id int) (*User, error) {
    var user User
    query := `SELECT id, name, email, created_at, updated_at FROM users WHERE id = $1`

    err := r.db.GetContext(ctx, &user, query, id)
    if err == sql.ErrNoRows {
        return nil, ErrUserNotFound
    }
    if err != nil {
        return nil, fmt.Errorf("get user: %w", err)
    }
    return &user, nil
}

func (r *PostgresUserRepository) Update(ctx context.Context, user *User) error {
    query := `
        UPDATE users
        SET name = $1, email = $2, updated_at = NOW()
        WHERE id = $3
        RETURNING updated_at
    `

    err := r.db.QueryRowContext(
        ctx, query,
        user.Name, user.Email, user.ID,
    ).Scan(&user.UpdatedAt)

    if err == sql.ErrNoRows {
        return ErrUserNotFound
    }
    return err
}

func (r *PostgresUserRepository) Delete(ctx context.Context, id int) error {
    query := `DELETE FROM users WHERE id = $1`
    result, err := r.db.ExecContext(ctx, query, id)
    if err != nil {
        return fmt.Errorf("delete user: %w", err)
    }

    rows, err := result.RowsAffected()
    if err != nil {
        return err
    }

    if rows == 0 {
        return ErrUserNotFound
    }
    return nil
}

Testing with Mock Repository

使用Mock仓库测试

package repository_test

import (
    "context"
    "testing"

    "github.com/stretchr/testify/assert"
    "github.com/stretchr/testify/mock"
)

// MockUserRepository for testing
type MockUserRepository struct {
    mock.Mock
}

func (m *MockUserRepository) GetByID(ctx context.Context, id int) (*User, error) {
    args := m.Called(ctx, id)
    if args.Get(0) == nil {
        return nil, args.Error(1)
    }
    return args.Get(0).(*User), args.Error(1)
}

func TestUserService_GetUser(t *testing.T) {
    mockRepo := new(MockUserRepository)
    service := NewUserService(mockRepo)

    expectedUser := &User{ID: 1, Name: "Alice", Email: "alice@example.com"}
    mockRepo.On("GetByID", mock.Anything, 1).Return(expectedUser, nil)

    user, err := service.GetUser(context.Background(), 1)

    assert.NoError(t, err)
    assert.Equal(t, expectedUser, user)
    mockRepo.AssertExpectations(t)
}

package repository_test

import (
    "context"
    "testing"

    "github.com/stretchr/testify/assert"
    "github.com/stretchr/testify/mock"
)

// MockUserRepository用于测试
type MockUserRepository struct {
    mock.Mock
}

func (m *MockUserRepository) GetByID(ctx context.Context, id int) (*User, error) {
    args := m.Called(ctx, id)
    if args.Get(0) == nil {
        return nil, args.Error(1)
    }
    return args.Get(0).(*User), args.Error(1)
}

func TestUserService_GetUser(t *testing.T) {
    mockRepo := new(MockUserRepository)
    service := NewUserService(mockRepo)

    expectedUser := &User{ID: 1, Name: "Alice", Email: "alice@example.com"}
    mockRepo.On("GetByID", mock.Anything, 1).Return(expectedUser, nil)

    user, err := service.GetUser(context.Background(), 1)

    assert.NoError(t, err)
    assert.Equal(t, expectedUser, user)
    mockRepo.AssertExpectations(t)
}

Transaction Handling

事务处理

Basic Transaction Pattern

基础事务模式

func (r *PostgresUserRepository) UpdateWithHistory(ctx context.Context, user *User) error {
    tx, err := r.db.BeginTxx(ctx, nil)
    if err != nil {
        return fmt.Errorf("begin tx: %w", err)
    }
    defer tx.Rollback() // Safe to call even after commit

    // Update user
    query := `UPDATE users SET name = $1, email = $2 WHERE id = $3`
    _, err = tx.ExecContext(ctx, query, user.Name, user.Email, user.ID)
    if err != nil {
        return fmt.Errorf("update user: %w", err)
    }

    // Insert history record
    historyQuery := `INSERT INTO user_history (user_id, name, email, changed_at) VALUES ($1, $2, $3, NOW())`
    _, err = tx.ExecContext(ctx, historyQuery, user.ID, user.Name, user.Email)
    if err != nil {
        return fmt.Errorf("insert history: %w", err)
    }

    if err := tx.Commit(); err != nil {
        return fmt.Errorf("commit tx: %w", err)
    }

    return nil
}

func (r *PostgresUserRepository) UpdateWithHistory(ctx context.Context, user *User) error {
    tx, err := r.db.BeginTxx(ctx, nil)
    if err != nil {
        return fmt.Errorf("begin tx: %w", err)
    }
    defer tx.Rollback() // 即使提交后调用也安全

    // 更新用户
    query := `UPDATE users SET name = $1, email = $2 WHERE id = $3`
    _, err = tx.ExecContext(ctx, query, user.Name, user.Email, user.ID)
    if err != nil {
        return fmt.Errorf("update user: %w", err)
    }

    // 插入历史记录
    historyQuery := `INSERT INTO user_history (user_id, name, email, changed_at) VALUES ($1, $2, $3, NOW())`
    _, err = tx.ExecContext(ctx, historyQuery, user.ID, user.Name, user.Email)
    if err != nil {
        return fmt.Errorf("insert history: %w", err)
    }

    if err := tx.Commit(); err != nil {
        return fmt.Errorf("commit tx: %w", err)
    }

    return nil
}

Transaction Isolation Levels

事务隔离级别

func (r *PostgresUserRepository) TransferBalance(ctx context.Context, fromID, toID int, amount float64) error {
    // Use serializable isolation for financial transactions
    txOpts := &sql.TxOptions{
        Isolation: sql.LevelSerializable,
        ReadOnly:  false,
    }

    tx, err := r.db.BeginTxx(ctx, txOpts)
    if err != nil {
        return err
    }
    defer tx.Rollback()

    // Deduct from sender
    _, err = tx.ExecContext(ctx,
        `UPDATE accounts SET balance = balance - $1 WHERE user_id = $2 AND balance >= $1`,
        amount, fromID,
    )
    if err != nil {
        return fmt.Errorf("deduct balance: %w", err)
    }

    // Add to receiver
    _, err = tx.ExecContext(ctx,
        `UPDATE accounts SET balance = balance + $1 WHERE user_id = $2`,
        amount, toID,
    )
    if err != nil {
        return fmt.Errorf("add balance: %w", err)
    }

    return tx.Commit()
}

func (r *PostgresUserRepository) TransferBalance(ctx context.Context, fromID, toID int, amount float64) error {
    // 金融事务使用可序列化隔离级别
    txOpts := &sql.TxOptions{
        Isolation: sql.LevelSerializable,
        ReadOnly:  false,
    }

    tx, err := r.db.BeginTxx(ctx, txOpts)
    if err != nil {
        return err
    }
    defer tx.Rollback()

    // 从转出方扣款
    _, err = tx.ExecContext(ctx,
        `UPDATE accounts SET balance = balance - $1 WHERE user_id = $2 AND balance >= $1`,
        amount, fromID,
    )
    if err != nil {
        return fmt.Errorf("deduct balance: %w", err)
    }

    // 给转入方加款
    _, err = tx.ExecContext(ctx,
        `UPDATE accounts SET balance = balance + $1 WHERE user_id = $2`,
        amount, toID,
    )
    if err != nil {
        return fmt.Errorf("add balance: %w", err)
    }

    return tx.Commit()
}

Retry Logic for Serialization Failures

序列化失败的重试逻辑

func WithRetry(ctx context.Context, maxRetries int, fn func() error) error {
    for i := 0; i < maxRetries; i++ {
        err := fn()
        if err == nil {
            return nil
        }

        // Check for serialization error (PostgreSQL error code 40001)
        var pgErr *pgconn.PgError
        if errors.As(err, &pgErr) && pgErr.Code == "40001" {
            // Exponential backoff
            time.Sleep(time.Duration(i+1) * 100 * time.Millisecond)
            continue
        }

        return err // Non-retryable error
    }
    return fmt.Errorf("max retries exceeded")
}

// Usage
err := WithRetry(ctx, 3, func() error {
    return r.TransferBalance(ctx, fromID, toID, amount)
})

func WithRetry(ctx context.Context, maxRetries int, fn func() error) error {
    for i := 0; i < maxRetries; i++ {
        err := fn()
        if err == nil {
            return nil
        }

        // 检查序列化错误（PostgreSQL错误码40001）
        var pgErr *pgconn.PgError
        if errors.As(err, &pgErr) && pgErr.Code == "40001" {
            // 指数退避
            time.Sleep(time.Duration(i+1) * 100 * time.Millisecond)
            continue
        }

        return err // 不可重试错误
    }
    return fmt.Errorf("max retries exceeded")
}

// 使用示例
err := WithRetry(ctx, 3, func() error {
    return r.TransferBalance(ctx, fromID, toID, amount)
})

Database Migrations

数据库迁移

Decision Tree: Migration Tools

决策树：选择迁移工具

┌─────────────────────────────────────┐
│ Migration tool selection            │
└──────────────┬──────────────────────┘
               │
    ┌──────────┴──────────┐
    │                     │
Simple SQL           Complex logic
migrations          (Go code needed)
    │                     │
    ▼                     ▼
golang-migrate         goose
(SQL only)        (Go + SQL migrations)

Use golang-migrate when:

Pure SQL migrations (no custom Go logic)
Need CLI tool for manual migrations
Want clean separation of schema and application
Industry standard (most popular)

Use goose when:

Need Go code in migrations (data transformations)
Want flexibility of both SQL and Go
Need custom migration logic

Use sql-migrate when:

Using sqlx already
Want embedded migrations in binary
Need programmatic migration control

┌─────────────────────────────────────┐
│ 迁移工具选择                        │
└──────────────┬──────────────────────┘
               │
    ┌──────────┴──────────┐
    │                     │
简单SQL迁移          复杂逻辑
                    (需要Go代码)
    │                     │
    ▼                     ▼
golang-migrate         goose
(仅支持SQL)        (支持Go+SQL迁移)

使用golang-migrate的场景：

纯SQL迁移（无自定义Go逻辑）
需要CLI工具手动执行迁移
希望Schema与应用代码清晰分离
行业标准（最受欢迎）

使用goose的场景：

迁移中需要Go代码（数据转换）
希望同时支持SQL和Go的灵活性
需要自定义迁移逻辑

使用sql-migrate的场景：

已在使用sqlx
希望将迁移嵌入二进制文件
需要程序化控制迁移

golang-migrate Setup

golang-migrate设置

Installation:

bash

undefined

安装：

bash

undefined

CLI tool

CLI工具

go install -tags 'postgres' github.com/golang-migrate/migrate/v4/cmd/migrate@latest

Library

库依赖

go get -u github.com/golang-migrate/migrate/v4 go get -u github.com/golang-migrate/migrate/v4/database/postgres go get -u github.com/golang-migrate/migrate/v4/source/file


**Migration Files:**

```bash

go get -u github.com/golang-migrate/migrate/v4 go get -u github.com/golang-migrate/migrate/v4/database/postgres go get -u github.com/golang-migrate/migrate/v4/source/file


**迁移文件：**

```bash

Create migration

创建迁移

migrate create -ext sql -dir migrations -seq create_users_table

Generates:

生成文件：

migrations/000001_create_users_table.up.sql

migrations/000001_create_users_table.down.sql


**000001_create_users_table.up.sql:**
```sql
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    email VARCHAR(255) UNIQUE NOT NULL,
    password_hash VARCHAR(255) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX idx_users_email ON users(email);

000001_create_users_table.down.sql:

sql

DROP INDEX IF EXISTS idx_users_email;
DROP TABLE IF EXISTS users;

Programmatic Migration:

package main

import (
    "fmt"

    "github.com/golang-migrate/migrate/v4"
    _ "github.com/golang-migrate/migrate/v4/database/postgres"
    _ "github.com/golang-migrate/migrate/v4/source/file"
)

func runMigrations(databaseURL, migrationsPath string) error {
    m, err := migrate.New(
        fmt.Sprintf("file://%s", migrationsPath),
        databaseURL,
    )
    if err != nil {
        return fmt.Errorf("create migrate instance: %w", err)
    }
    defer m.Close()

    if err := m.Up(); err != nil && err != migrate.ErrNoChange {
        return fmt.Errorf("run migrations: %w", err)
    }

    version, dirty, err := m.Version()
    if err != nil {
        return err
    }

    fmt.Printf("Migration complete. Version: %d, Dirty: %v\n", version, dirty)
    return nil
}

CLI Usage:

bash

undefined


**000001_create_users_table.up.sql：**
```sql
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    email VARCHAR(255) UNIQUE NOT NULL,
    password_hash VARCHAR(255) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX idx_users_email ON users(email);

000001_create_users_table.down.sql：

sql

DROP INDEX IF EXISTS idx_users_email;
DROP TABLE IF EXISTS users;

程序化执行迁移：

package main

import (
    "fmt"

    "github.com/golang-migrate/migrate/v4"
    _ "github.com/golang-migrate/migrate/v4/database/postgres"
    _ "github.com/golang-migrate/migrate/v4/source/file"
)

func runMigrations(databaseURL, migrationsPath string) error {
    m, err := migrate.New(
        fmt.Sprintf("file://%s", migrationsPath),
        databaseURL,
    )
    if err != nil {
        return fmt.Errorf("create migrate instance: %w", err)
    }
    defer m.Close()

    if err := m.Up(); err != nil && err != migrate.ErrNoChange {
        return fmt.Errorf("run migrations: %w", err)
    }

    version, dirty, err := m.Version()
    if err != nil {
        return err
    }

    fmt.Printf("Migration complete. Version: %d, Dirty: %v\n", version, dirty)
    return nil
}

CLI使用：

bash

undefined

Apply all up migrations

应用所有up迁移

migrate -path migrations -database "postgres://user:pass@localhost:5432/db?sslmode=disable" up

Rollback one migration

回滚一个迁移

migrate -path migrations -database $DATABASE_URL down 1

Go to specific version

迁移到指定版本

migrate -path migrations -database $DATABASE_URL goto 5

Check current version

检查当前版本

migrate -path migrations -database $DATABASE_URL version

undefined

migrate -path migrations -database $DATABASE_URL version

undefined

NULL Handling

NULL值处理

Using sql.Null* Types

使用sql.Null*类型

type User struct {
    ID        int            `db:"id"`
    Name      string         `db:"name"`
    Email     string         `db:"email"`
    Phone     sql.NullString `db:"phone"`     // Nullable string
    Age       sql.NullInt64  `db:"age"`       // Nullable int
    UpdatedAt sql.NullTime   `db:"updated_at"` // Nullable timestamp
}

func (r *PostgresUserRepository) GetUser(ctx context.Context, id int) (*User, error) {
    var user User
    err := r.db.GetContext(ctx, &user, `SELECT * FROM users WHERE id = $1`, id)
    if err != nil {
        return nil, err
    }

    // Access nullable fields
    if user.Phone.Valid {
        fmt.Println("Phone:", user.Phone.String)
    }

    return &user, nil
}

// Setting NULL values
func (r *PostgresUserRepository) UpdatePhone(ctx context.Context, userID int, phone *string) error {
    var nullPhone sql.NullString
    if phone != nil {
        nullPhone = sql.NullString{String: *phone, Valid: true}
    }
    // If phone is nil, nullPhone.Valid is false, SQL writes NULL

    query := `UPDATE users SET phone = $1 WHERE id = $2`
    _, err := r.db.ExecContext(ctx, query, nullPhone, userID)
    return err
}

type User struct {
    ID        int            `db:"id"`
    Name      string         `db:"name"`
    Email     string         `db:"email"`
    Phone     sql.NullString `db:"phone"`     // 可空字符串
    Age       sql.NullInt64  `db:"age"`       // 可空整数
    UpdatedAt sql.NullTime   `db:"updated_at"` // 可空时间戳
}

func (r *PostgresUserRepository) GetUser(ctx context.Context, id int) (*User, error) {
    var user User
    err := r.db.GetContext(ctx, &user, `SELECT * FROM users WHERE id = $1`, id)
    if err != nil {
        return nil, err
    }

    // 访问可空字段
    if user.Phone.Valid {
        fmt.Println("Phone:", user.Phone.String)
    }

    return &user, nil
}

// 设置NULL值
func (r *PostgresUserRepository) UpdatePhone(ctx context.Context, userID int, phone *string) error {
    var nullPhone sql.NullString
    if phone != nil {
        nullPhone = sql.NullString{String: *phone, Valid: true}
    }
    // 如果phone是nil，nullPhone.Valid为false，SQL会写入NULL

    query := `UPDATE users SET phone = $1 WHERE id = $2`
    _, err := r.db.ExecContext(ctx, query, nullPhone, userID)
    return err
}

Custom Nullable Types (Preferred Pattern)

自定义可空类型（推荐模式）

// Custom nullable type with JSON marshaling
type NullString struct {
    sql.NullString
}

func (ns NullString) MarshalJSON() ([]byte, error) {
    if !ns.Valid {
        return []byte("null"), nil
    }
    return json.Marshal(ns.String)
}

func (ns *NullString) UnmarshalJSON(data []byte) error {
    if string(data) == "null" {
        ns.Valid = false
        return nil
    }

    var s string
    if err := json.Unmarshal(data, &s); err != nil {
        return err
    }

    ns.String = s
    ns.Valid = true
    return nil
}

// 支持JSON序列化的自定义可空类型
type NullString struct {
    sql.NullString
}

func (ns NullString) MarshalJSON() ([]byte, error) {
    if !ns.Valid {
        return []byte("null"), nil
    }
    return json.Marshal(ns.String)
}

func (ns *NullString) UnmarshalJSON(data []byte) error {
    if string(data) == "null" {
        ns.Valid = false
        return nil
    }

    var s string
    if err := json.Unmarshal(data, &s); err != nil {
        return err
    }

    ns.String = s
    ns.Valid = true
    return nil
}

Anti-Patterns to Avoid

需避免的反模式

❌ N+1 Query Problem

❌ N+1查询问题

Wrong:

func GetUsersWithPosts(ctx context.Context, db *sqlx.DB) ([]UserWithPosts, error) {
    var users []User
    db.SelectContext(ctx, &users, `SELECT * FROM users`)

    for i, user := range users {
        var posts []Post
        // N+1: One query per user!
        db.SelectContext(ctx, &posts, `SELECT * FROM posts WHERE user_id = $1`, user.ID)
        users[i].Posts = posts
    }
    return users, nil
}

Correct:

func GetUsersWithPosts(ctx context.Context, db *sqlx.DB) ([]UserWithPosts, error) {
    // Single query with JOIN
    query := `
        SELECT u.id, u.name, p.id as post_id, p.title, p.content
        FROM users u
        LEFT JOIN posts p ON p.user_id = u.id
        ORDER BY u.id
    `

    rows, err := db.QueryContext(ctx, query)
    if err != nil {
        return nil, err
    }
    defer rows.Close()

    usersMap := make(map[int]*UserWithPosts)
    for rows.Next() {
        var userID int
        var userName string
        var postID sql.NullInt64
        var title, content sql.NullString

        rows.Scan(&userID, &userName, &postID, &title, &content)

        if _, exists := usersMap[userID]; !exists {
            usersMap[userID] = &UserWithPosts{ID: userID, Name: userName}
        }

        if postID.Valid {
            usersMap[userID].Posts = append(usersMap[userID].Posts, Post{
                ID:      int(postID.Int64),
                Title:   title.String,
                Content: content.String,
            })
        }
    }

    result := make([]UserWithPosts, 0, len(usersMap))
    for _, user := range usersMap {
        result = append(result, *user)
    }
    return result, nil
}

错误示例：

func GetUsersWithPosts(ctx context.Context, db *sqlx.DB) ([]UserWithPosts, error) {
    var users []User
    db.SelectContext(ctx, &users, `SELECT * FROM users`)

    for i, user := range users {
        var posts []Post
        // N+1：每个用户都执行一次查询！
        db.SelectContext(ctx, &posts, `SELECT * FROM posts WHERE user_id = $1`, user.ID)
        users[i].Posts = posts
    }
    return users, nil
}

正确示例：

func GetUsersWithPosts(ctx context.Context, db *sqlx.DB) ([]UserWithPosts, error) {
    // 单查询+JOIN
    query := `
        SELECT u.id, u.name, p.id as post_id, p.title, p.content
        FROM users u
        LEFT JOIN posts p ON p.user_id = u.id
        ORDER BY u.id
    `

    rows, err := db.QueryContext(ctx, query)
    if err != nil {
        return nil, err
    }
    defer rows.Close()

    usersMap := make(map[int]*UserWithPosts)
    for rows.Next() {
        var userID int
        var userName string
        var postID sql.NullInt64
        var title, content sql.NullString

        rows.Scan(&userID, &userName, &postID, &title, &content)

        if _, exists := usersMap[userID]; !exists {
            usersMap[userID] = &UserWithPosts{ID: userID, Name: userName}
        }

        if postID.Valid {
            usersMap[userID].Posts = append(usersMap[userID].Posts, Post{
                ID:      int(postID.Int64),
                Title:   title.String,
                Content: content.String,
            })
        }
    }

    result := make([]UserWithPosts, 0, len(usersMap))
    for _, user := range usersMap {
        result = append(result, *user)
    }
    return result, nil
}

❌ Missing Connection Pool Configuration

❌ 未配置连接池

Wrong:

db, _ := sql.Open("postgres", dsn)
// Uses defaults: unlimited connections, no timeouts

Correct:

db, _ := sql.Open("postgres", dsn)

// Production-ready pool settings
db.SetMaxOpenConns(25)                      // Limit total connections
db.SetMaxIdleConns(5)                       // Limit idle connections
db.SetConnMaxLifetime(5 * time.Minute)      // Recycle old connections
db.SetConnMaxIdleTime(1 * time.Minute)      // Close idle connections

错误示例：

db, _ := sql.Open("postgres", dsn)
// 使用默认配置：无连接数限制，无超时

正确示例：

db, _ := sql.Open("postgres", dsn)

// 生产环境就绪的池配置
db.SetMaxOpenConns(25)                      // 限制总连接数
db.SetMaxIdleConns(5)                       // 限制空闲连接数
db.SetConnMaxLifetime(5 * time.Minute)      // 回收旧连接
db.SetConnMaxIdleTime(1 * time.Minute)      // 关闭空闲连接

❌ Ignoring Context Cancellation

❌ 忽略上下文取消

Wrong:

func SlowQuery(db *sql.DB) error {
    // No context - query runs until completion even if client disconnects
    rows, err := db.Query("SELECT * FROM huge_table")
    // ...
}

Correct:

func SlowQuery(ctx context.Context, db *sql.DB) error {
    // Context cancellation propagates to database
    rows, err := db.QueryContext(ctx, "SELECT * FROM huge_table")
    // If ctx is canceled, query is terminated
}

错误示例：

func SlowQuery(db *sql.DB) error {
    // 无上下文 - 即使客户端断开连接，查询仍会执行到完成
    rows, err := db.Query("SELECT * FROM huge_table")
    // ...
}

正确示例：

func SlowQuery(ctx context.Context, db *sql.DB) error {
    // 上下文取消会传递到数据库
    rows, err := db.QueryContext(ctx, "SELECT * FROM huge_table")
    // 如果ctx被取消，查询会终止
}

❌ Not Closing Rows

❌ 未关闭Rows

Wrong:

func GetUsers(db *sql.DB) ([]User, error) {
    rows, _ := db.Query("SELECT * FROM users")
    // Missing rows.Close() - connection leak!
    var users []User
    for rows.Next() {
        // ...
    }
    return users, nil
}

Correct:

func GetUsers(db *sql.DB) ([]User, error) {
    rows, err := db.Query("SELECT * FROM users")
    if err != nil {
        return nil, err
    }
    defer rows.Close() // CRITICAL: Always defer Close

    var users []User
    for rows.Next() {
        // ...
    }
    return users, rows.Err() // Check for iteration errors
}

错误示例：

func GetUsers(db *sql.DB) ([]User, error) {
    rows, _ := db.Query("SELECT * FROM users")
    // 缺少rows.Close() - 连接泄漏！
    var users []User
    for rows.Next() {
        // ...
    }
    return users, nil
}

正确示例：

func GetUsers(db *sql.DB) ([]User, error) {
    rows, err := db.Query("SELECT * FROM users")
    if err != nil {
        return nil, err
    }
    defer rows.Close() // 关键：必须延迟调用Close

    var users []User
    for rows.Next() {
        // ...
    }
    return users, rows.Err() // 检查迭代错误
}

❌ SQL Injection Vulnerability

❌ SQL注入漏洞

Wrong:

func FindUser(db *sql.DB, email string) (*User, error) {
    // NEVER concatenate user input into SQL!
    query := fmt.Sprintf("SELECT * FROM users WHERE email = '%s'", email)
    // Vulnerable to: ' OR '1'='1
    row := db.QueryRow(query)
    // ...
}

Correct:

func FindUser(db *sql.DB, email string) (*User, error) {
    // Use parameterized queries
    query := "SELECT * FROM users WHERE email = $1"
    row := db.QueryRow(query, email) // Safe
    // ...
}

错误示例：

func FindUser(db *sql.DB, email string) (*User, error) {
    // 绝对不要将用户输入拼接到SQL中！
    query := fmt.Sprintf("SELECT * FROM users WHERE email = '%s'", email)
    // 易受注入攻击：' OR '1'='1
    row := db.QueryRow(query)
    // ...
}

正确示例：

func FindUser(db *sql.DB, email string) (*User, error) {
    // 使用参数化查询
    query := "SELECT * FROM users WHERE email = $1"
    row := db.QueryRow(query, email) // 安全
    // ...
}

❌ Ignoring Transaction Errors

❌ 忽略事务错误

Wrong:

func UpdateUser(db *sql.DB, user *User) error {
    tx, _ := db.Begin()
    tx.Exec("UPDATE users SET name = $1 WHERE id = $2", user.Name, user.ID)
    tx.Commit() // Ignores errors - data may not be committed!
    return nil
}

Correct:

func UpdateUser(db *sql.DB, user *User) error {
    tx, err := db.Begin()
    if err != nil {
        return err
    }
    defer tx.Rollback() // Rollback if commit not reached

    _, err = tx.Exec("UPDATE users SET name = $1 WHERE id = $2", user.Name, user.ID)
    if err != nil {
        return err
    }

    return tx.Commit() // Check commit error
}

错误示例：

func UpdateUser(db *sql.DB, user *User) error {
    tx, _ := db.Begin()
    tx.Exec("UPDATE users SET name = $1 WHERE id = $2", user.Name, user.ID)
    tx.Commit() // 忽略错误 - 数据可能未提交！
    return nil
}

正确示例：

func UpdateUser(db *sql.DB, user *User) error {
    tx, err := db.Begin()
    if err != nil {
        return err
    }
    defer tx.Rollback() // 未提交则回滚

    _, err = tx.Exec("UPDATE users SET name = $1 WHERE id = $2", user.Name, user.ID)
    if err != nil {
        return err
    }

    return tx.Commit() // 检查提交错误
}

Connection Pooling Best Practices

连接池最佳实践

Tuning Parameters

调优参数

func OptimizeDatabasePool(db *sql.DB, config PoolConfig) {
    // MaxOpenConns: Total connections (in-use + idle)
    // Rule of thumb: (CPU cores * 2) + disk spindles
    // Cloud databases often limit connections (e.g., AWS RDS: 100-5000)
    db.SetMaxOpenConns(config.MaxOpen) // Example: 25 for small app

    // MaxIdleConns: Idle connections ready for reuse
    // Should be lower than MaxOpenConns
    // Too low: frequent reconnections (slow)
    // Too high: wasted resources
    db.SetMaxIdleConns(config.MaxIdle) // Example: 5-10

    // ConnMaxLifetime: Maximum age of any connection
    // Prevents stale connections to load balancers
    // Recommended: 5-15 minutes
    db.SetConnMaxLifetime(config.MaxLifetime)

    // ConnMaxIdleTime: Close idle connections after this duration
    // Saves resources during low traffic
    // Recommended: 1-5 minutes
    db.SetConnMaxIdleTime(config.MaxIdleTime)
}

type PoolConfig struct {
    MaxOpen      int
    MaxIdle      int
    MaxLifetime  time.Duration
    MaxIdleTime  time.Duration
}

// Example configurations
var (
    // Development: Low resource usage
    DevConfig = PoolConfig{
        MaxOpen:     10,
        MaxIdle:     2,
        MaxLifetime: 10 * time.Minute,
        MaxIdleTime: 2 * time.Minute,
    }

    // Production: High throughput
    ProdConfig = PoolConfig{
        MaxOpen:     25,
        MaxIdle:     10,
        MaxLifetime: 5 * time.Minute,
        MaxIdleTime: 1 * time.Minute,
    }

    // High-traffic API: Maximum performance
    HighTrafficConfig = PoolConfig{
        MaxOpen:     50,
        MaxIdle:     20,
        MaxLifetime: 5 * time.Minute,
        MaxIdleTime: 30 * time.Second,
    }
)

func OptimizeDatabasePool(db *sql.DB, config PoolConfig) {
    // MaxOpenConns: 总连接数（使用中 + 空闲）
    // 经验法则：(CPU核心数 * 2) + 磁盘主轴数
    // 云数据库通常限制连接数（如AWS RDS：100-5000）
    db.SetMaxOpenConns(config.MaxOpen) // 示例：小型应用设为25

    // MaxIdleConns: 可复用的空闲连接数
    // 应小于MaxOpenConns
    // 过小：频繁重连（慢）
    // 过大：资源浪费
    db.SetMaxIdleConns(config.MaxIdle) // 示例：5-10

    // ConnMaxLifetime: 连接最大存活时间
    // 防止连接到负载均衡器的连接过期
    // 推荐：5-15分钟
    db.SetConnMaxLifetime(config.MaxLifetime)

    // ConnMaxIdleTime: 空闲连接超时关闭时间
    // 低流量时节省资源
    // 推荐：1-5分钟
    db.SetConnMaxIdleTime(config.MaxIdleTime)
}

type PoolConfig struct {
    MaxOpen      int
    MaxIdle      int
    MaxLifetime  time.Duration
    MaxIdleTime  time.Duration
}

// 示例配置
var (
    // 开发环境：低资源占用
    DevConfig = PoolConfig{
        MaxOpen:     10,
        MaxIdle:     2,
        MaxLifetime: 10 * time.Minute,
        MaxIdleTime: 2 * time.Minute,
    }

    // 生产环境：高吞吐量
    ProdConfig = PoolConfig{
        MaxOpen:     25,
        MaxIdle:     10,
        MaxLifetime: 5 * time.Minute,
        MaxIdleTime: 1 * time.Minute,
    }

    // 高流量API：极致性能
    HighTrafficConfig = PoolConfig{
        MaxOpen:     50,
        MaxIdle:     20,
        MaxLifetime: 5 * time.Minute,
        MaxIdleTime: 30 * time.Second,
    }
)

Monitoring Connection Pool

监控连接池

func MonitorConnectionPool(db *sql.DB) {
    stats := db.Stats()

    fmt.Printf("Connection Pool Stats:\n")
    fmt.Printf("  Open Connections: %d\n", stats.OpenConnections)
    fmt.Printf("  In Use: %d\n", stats.InUse)
    fmt.Printf("  Idle: %d\n", stats.Idle)
    fmt.Printf("  Wait Count: %d\n", stats.WaitCount)          // Queries waited for connection
    fmt.Printf("  Wait Duration: %s\n", stats.WaitDuration)    // Total wait time
    fmt.Printf("  Max Idle Closed: %d\n", stats.MaxIdleClosed) // Connections closed due to idle
    fmt.Printf("  Max Lifetime Closed: %d\n", stats.MaxLifetimeClosed)

    // Alert if too many waits (need more connections)
    if stats.WaitCount > 100 {
        fmt.Println("WARNING: High wait count - consider increasing MaxOpenConns")
    }

    // Alert if many idle closures (pool too large)
    if stats.MaxIdleClosed > 1000 {
        fmt.Println("INFO: Many idle closures - consider reducing MaxIdleConns")
    }
}

func MonitorConnectionPool(db *sql.DB) {
    stats := db.Stats()

    fmt.Printf("连接池统计信息：\n")
    fmt.Printf("  打开连接数：%d\n", stats.OpenConnections)
    fmt.Printf("  使用中：%d\n", stats.InUse)
    fmt.Printf("  空闲：%d\n", stats.Idle)
    fmt.Printf("  等待次数：%d\n", stats.WaitCount)          // 等待连接的查询数
    fmt.Printf("  等待总时长：%s\n", stats.WaitDuration)    // 总等待时间
    fmt.Printf("  因空闲关闭的连接数：%d\n", stats.MaxIdleClosed) // 因空闲被关闭的连接
    fmt.Printf("  因生命周期关闭的连接数：%d\n", stats.MaxLifetimeClosed)

    // 等待次数过多时告警（需增加连接数）
    if stats.WaitCount > 100 {
        fmt.Println("警告：等待次数过多 - 考虑增加MaxOpenConns")
    }

    // 空闲关闭连接过多时告警（池配置过大）
    if stats.MaxIdleClosed > 1000 {
        fmt.Println("提示：大量空闲连接被关闭 - 考虑减少MaxIdleConns")
    }
}

Testing Database Code

数据库代码测试

Using sqlmock for Unit Tests

使用sqlmock进行单元测试

Installation:

bash

go get github.com/DATA-DOG/go-sqlmock

Example:

package repository_test

import (
    "context"
    "testing"

    "github.com/DATA-DOG/go-sqlmock"
    "github.com/jmoiron/sqlx"
    "github.com/stretchr/testify/assert"
)

func TestGetUserByID(t *testing.T) {
    // Create mock database
    db, mock, err := sqlmock.New()
    assert.NoError(t, err)
    defer db.Close()

    sqlxDB := sqlx.NewDb(db, "postgres")
    repo := NewPostgresUserRepository(sqlxDB)

    // Expected query and result
    rows := sqlmock.NewRows([]string{"id", "name", "email"}).
        AddRow(1, "Alice", "alice@example.com")

    mock.ExpectQuery("SELECT (.+) FROM users WHERE id = \\$1").
        WithArgs(1).
        WillReturnRows(rows)

    // Execute
    user, err := repo.GetByID(context.Background(), 1)

    // Assert
    assert.NoError(t, err)
    assert.Equal(t, "Alice", user.Name)
    assert.Equal(t, "alice@example.com", user.Email)
    assert.NoError(t, mock.ExpectationsWereMet())
}

func TestGetUserByID_NotFound(t *testing.T) {
    db, mock, err := sqlmock.New()
    assert.NoError(t, err)
    defer db.Close()

    sqlxDB := sqlx.NewDb(db, "postgres")
    repo := NewPostgresUserRepository(sqlxDB)

    mock.ExpectQuery("SELECT (.+) FROM users WHERE id = \\$1").
        WithArgs(999).
        WillReturnError(sql.ErrNoRows)

    user, err := repo.GetByID(context.Background(), 999)

    assert.Nil(t, user)
    assert.ErrorIs(t, err, ErrUserNotFound)
    assert.NoError(t, mock.ExpectationsWereMet())
}

安装：

bash

go get github.com/DATA-DOG/go-sqlmock

示例：

package repository_test

import (
    "context"
    "testing"

    "github.com/DATA-DOG/go-sqlmock"
    "github.com/jmoiron/sqlx"
    "github.com/stretchr/testify/assert"
)

func TestGetUserByID(t *testing.T) {
    // 创建Mock数据库
    db, mock, err := sqlmock.New()
    assert.NoError(t, err)
    defer db.Close()

    sqlxDB := sqlx.NewDb(db, "postgres")
    repo := NewPostgresUserRepository(sqlxDB)

    // 预期查询和结果
    rows := sqlmock.NewRows([]string{"id", "name", "email"}).
        AddRow(1, "Alice", "alice@example.com")

    mock.ExpectQuery("SELECT (.+) FROM users WHERE id = \\$1").
        WithArgs(1).
        WillReturnRows(rows)

    // 执行测试
    user, err := repo.GetByID(context.Background(), 1)

    // 断言
    assert.NoError(t, err)
    assert.Equal(t, "Alice", user.Name)
    assert.Equal(t, "alice@example.com", user.Email)
    assert.NoError(t, mock.ExpectationsWereMet())
}

func TestGetUserByID_NotFound(t *testing.T) {
    db, mock, err := sqlmock.New()
    assert.NoError(t, err)
    defer db.Close()

    sqlxDB := sqlx.NewDb(db, "postgres")
    repo := NewPostgresUserRepository(sqlxDB)

    mock.ExpectQuery("SELECT (.+) FROM users WHERE id = \\$1").
        WithArgs(999).
        WillReturnError(sql.ErrNoRows)

    user, err := repo.GetByID(context.Background(), 999)

    assert.Nil(t, user)
    assert.ErrorIs(t, err, ErrUserNotFound)
    assert.NoError(t, mock.ExpectationsWereMet())
}

Integration Tests with Real Database

使用真实数据库进行集成测试

// +build integration

package repository_test

import (
    "context"
    "testing"

    "github.com/jmoiron/sqlx"
    _ "github.com/lib/pq"
    "github.com/stretchr/testify/assert"
    "github.com/stretchr/testify/suite"
)

type UserRepositoryIntegrationSuite struct {
    suite.Suite
    db   *sqlx.DB
    repo *PostgresUserRepository
}

func (s *UserRepositoryIntegrationSuite) SetupSuite() {
    // Connect to test database
    db, err := sqlx.Connect("postgres", "postgres://test:test@localhost/testdb?sslmode=disable")
    s.Require().NoError(err)
    s.db = db
    s.repo = NewPostgresUserRepository(db)
}

func (s *UserRepositoryIntegrationSuite) TearDownSuite() {
    s.db.Close()
}

func (s *UserRepositoryIntegrationSuite) SetupTest() {
    // Clean database before each test
    _, err := s.db.Exec("TRUNCATE users RESTART IDENTITY CASCADE")
    s.Require().NoError(err)
}

func (s *UserRepositoryIntegrationSuite) TestCreateUser() {
    user := &User{Name: "Alice", Email: "alice@example.com"}

    err := s.repo.Create(context.Background(), user)

    s.NoError(err)
    s.NotZero(user.ID)
    s.NotZero(user.CreatedAt)
}

func (s *UserRepositoryIntegrationSuite) TestGetUserByID() {
    // Insert test data
    user := &User{Name: "Bob", Email: "bob@example.com"}
    s.repo.Create(context.Background(), user)

    // Test retrieval
    retrieved, err := s.repo.GetByID(context.Background(), user.ID)

    s.NoError(err)
    s.Equal(user.Name, retrieved.Name)
    s.Equal(user.Email, retrieved.Email)
}

func TestUserRepositoryIntegration(t *testing.T) {
    suite.Run(t, new(UserRepositoryIntegrationSuite))
}

Run integration tests:

bash

undefined

// +build integration

package repository_test

import (
    "context"
    "testing"

    "github.com/jmoiron/sqlx"
    _ "github.com/lib/pq"
    "github.com/stretchr/testify/assert"
    "github.com/stretchr/testify/suite"
)

type UserRepositoryIntegrationSuite struct {
    suite.Suite
    db   *sqlx.DB
    repo *PostgresUserRepository
}

func (s *UserRepositoryIntegrationSuite) SetupSuite() {
    // 连接测试数据库
    db, err := sqlx.Connect("postgres", "postgres://test:test@localhost/testdb?sslmode=disable")
    s.Require().NoError(err)
    s.db = db
    s.repo = NewPostgresUserRepository(db)
}

func (s *UserRepositoryIntegrationSuite) TearDownSuite() {
    s.db.Close()
}

func (s *UserRepositoryIntegrationSuite) SetupTest() {
    // 每个测试前清理数据库
    _, err := s.db.Exec("TRUNCATE users RESTART IDENTITY CASCADE")
    s.Require().NoError(err)
}

func (s *UserRepositoryIntegrationSuite) TestCreateUser() {
    user := &User{Name: "Alice", Email: "alice@example.com"}

    err := s.repo.Create(context.Background(), user)

    s.NoError(err)
    s.NotZero(user.ID)
    s.NotZero(user.CreatedAt)
}

func (s *UserRepositoryIntegrationSuite) TestGetUserByID() {
    // 插入测试数据
    user := &User{Name: "Bob", Email: "bob@example.com"}
    s.repo.Create(context.Background(), user)

    // 测试查询
    retrieved, err := s.repo.GetByID(context.Background(), user.ID)

    s.NoError(err)
    s.Equal(user.Name, retrieved.Name)
    s.Equal(user.Email, retrieved.Email)
}

func TestUserRepositoryIntegration(t *testing.T) {
    suite.Run(t, new(UserRepositoryIntegrationSuite))
}

运行集成测试：

bash

undefined

Skip integration tests by default

默认跳过集成测试

go test ./...

Run only integration tests

仅运行集成测试

go test -tags=integration ./...

undefined

go test -tags=integration ./...

undefined

Resources and Further Reading

资源与扩展阅读

Official Documentation

官方文档

database/sql Tutorial - Official Go database guide (2024)
sqlx Documentation - Illustrated guide to sqlx
pgx Documentation - PostgreSQL driver and toolkit
golang-migrate - Database migration tool

database/sql教程 - Go官方数据库指南（2024）
sqlx文档 - sqlx图解指南
pgx文档 - PostgreSQL驱动与工具包
golang-migrate - 数据库迁移工具

Best Practices

最佳实践

Go Database Best Practices - Alex Edwards (2024)
Connection Pool Tuning - Production configuration guide
Repository Pattern in Go - Three Dots Labs

Go数据库最佳实践 - Alex Edwards（2024）
连接池调优 - 生产环境配置指南
Go中的仓库模式 - Three Dots Labs

Migration Tools

迁移工具

golang-migrate CLI - Command-line tool
goose - Alternative with Go and SQL migrations
sql-migrate - Migration tool with sqlx integration

golang-migrate CLI - 命令行工具
goose - 支持Go和SQL迁移的替代工具
sql-migrate - 与sqlx集成的迁移工具

Testing

测试

go-sqlmock - SQL mock for unit tests
testcontainers-go - Docker containers for integration tests
dockertest - Ephemeral databases for testing

go-sqlmock - 用于单元测试的SQL Mock
testcontainers-go - 用于集成测试的Docker容器
dockertest - 用于测试的临时数据库

Advanced Topics

高级主题

Handling Database Errors - NULL handling patterns
PostgreSQL LISTEN/NOTIFY with pgx - Real-time notifications
Query Builders - Dynamic SQL generation
GORM - Full-featured ORM (alternative approach)

数据库错误处理 - NULL处理模式
使用pgx实现PostgreSQL LISTEN/NOTIFY - 实时通知
查询构建器 - 动态SQL生成
GORM - 全功能ORM（替代方案）

Summary

总结

Go database patterns prioritize simplicity, type safety, and performance:

Library Selection:

Start with
```
database/sql
```
for portability
Add
```
sqlx
```
for convenience and reduced boilerplate
Use
```
pgx
```
for PostgreSQL-specific high-performance applications

Core Patterns:

Repository pattern for testable data access layers
Context-aware queries for cancellation and timeouts
Proper transaction handling with defer rollback
Connection pooling tuned for production workloads

Migration Strategy:

Use
```
golang-migrate
```
for version-controlled schema evolution
Separate up/down migrations for safe rollbacks
Run migrations programmatically or via CLI

Avoid Common Pitfalls:

N+1 queries (use JOINs or batching)
Missing connection pool configuration
SQL injection (always use parameterized queries)
Not closing rows (defer rows.Close())
Ignoring context cancellation

Testing:

Unit tests with sqlmock for business logic
Integration tests with real databases for critical paths
Repository interfaces for dependency injection

By following these patterns, you'll build robust, performant, and maintainable database layers in Go.

Go数据库模式优先考虑简洁性、类型安全性和性能：

库选择：

从
```
database/sql
```
开始，保证可移植性
添加
```
sqlx
```
提升便捷性，减少样板代码
PostgreSQL专属高性能应用使用
```
pgx
```

核心模式：

仓库模式，实现可测试的数据访问层
上下文感知查询，支持取消和超时
正确的事务处理，延迟回滚
针对生产工作负载调优连接池

迁移策略：

使用
```
golang-migrate
```
进行版本化Schema演进
分离up/down迁移，支持安全回滚
程序化或通过CLI执行迁移

避免常见陷阱：

N+1查询（使用JOIN或批处理）
未配置连接池
SQL注入（始终使用参数化查询）
未关闭Rows（defer rows.Close()）
忽略上下文取消

测试：

使用sqlmock进行单元测试，验证业务逻辑
关键路径使用真实数据库进行集成测试
仓库接口支持依赖注入

遵循这些模式，你将构建出健壮、高性能且易于维护的Go数据库层。