📖 数据库操作¶

package main

import (
    "database/sql"
    "fmt"
    "log"
    "time"

    _ "github.com/go-sql-driver/mysql" // MySQL 驱动
    // _ "github.com/lib/pq"           // PostgreSQL 驱动
    // _ "github.com/mattn/go-sqlite3"  // SQLite 驱动
)

func main() {
    // DSN 格式: user:password@protocol(host:port)/dbname?params
    dsn := "root:password@tcp(127.0.0.1:3306)/mydb?charset=utf8mb4&parseTime=True&loc=Local"

    db, err := sql.Open("mysql", dsn)
    if err != nil {
        log.Fatalf("打开数据库失败: %v", err)
    }
    defer db.Close()

    // sql.Open 不会真正连接，需要 Ping 验证
    if err := db.Ping(); err != nil {
        log.Fatalf("连接数据库失败: %v", err)
    }

    // ⚠️ 连接池配置（生产环境必须设置）
    db.SetMaxOpenConns(25)                  // 最大打开连接数
    db.SetMaxIdleConns(10)                  // 最大空闲连接数
    db.SetConnMaxLifetime(30 * time.Minute) // 连接最大存活时间
    db.SetConnMaxIdleTime(5 * time.Minute)  // 空闲连接最大存活时间

    log.Println("数据库连接成功")
}

// 插入
func insertUser(db *sql.DB, name, email string) (int64, error) {
    result, err := db.Exec(
        "INSERT INTO users (name, email, created_at) VALUES (?, ?, NOW())",
        name, email,  // ⚠️ 始终使用参数化查询，防止 SQL 注入
    )
    if err != nil {
        return 0, fmt.Errorf("插入失败: %w", err)
    }
    return result.LastInsertId()
}

// 查询单行
func getUserByID(db *sql.DB, id int64) (*User, error) {
    user := &User{}
    err := db.QueryRow(
        "SELECT id, name, email, created_at FROM users WHERE id = ?", id,
    ).Scan(&user.ID, &user.Name, &user.Email, &user.CreatedAt)

    if err == sql.ErrNoRows {
        return nil, fmt.Errorf("用户 %d 不存在", id)
    }
    if err != nil {
        return nil, fmt.Errorf("查询失败: %w", err)
    }
    return user, nil
}

// 查询多行
func listUsers(db *sql.DB, limit, offset int) ([]*User, error) {
    rows, err := db.Query(
        "SELECT id, name, email, created_at FROM users ORDER BY id DESC LIMIT ? OFFSET ?",
        limit, offset,
    )
    if err != nil {
        return nil, fmt.Errorf("查询失败: %w", err)
    }
    defer rows.Close() // ⚠️ 必须关闭，否则连接泄漏

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

    // ⚠️ 检查迭代过程中是否有错误
    if err := rows.Err(); err != nil {
        return nil, fmt.Errorf("迭代失败: %w", err)
    }
    return users, nil
}

// 更新
func updateUser(db *sql.DB, id int64, name string) (int64, error) {
    result, err := db.Exec(
        "UPDATE users SET name = ?, updated_at = NOW() WHERE id = ?",
        name, id,
    )
    if err != nil {
        return 0, fmt.Errorf("更新失败: %w", err)
    }
    return result.RowsAffected()
}

// 删除
func deleteUser(db *sql.DB, id int64) error {
    result, err := db.Exec("DELETE FROM users WHERE id = ?", id)
    if err != nil {
        return fmt.Errorf("删除失败: %w", err)
    }
    affected, _ := result.RowsAffected()
    if affected == 0 {
        return fmt.Errorf("用户 %d 不存在", id)
    }
    return nil
}

// 转账示例 — 正确的事务模式
func transfer(db *sql.DB, fromID, toID int64, amount float64) error {
    tx, err := db.Begin()
    if err != nil {
        return fmt.Errorf("开启事务失败: %w", err)
    }
    // ⚠️ defer rollback — 如果已 commit 则 rollback 是 no-op
    defer tx.Rollback()

    // 加锁查询余额
    var balance float64
    err = tx.QueryRow(
        "SELECT balance FROM accounts WHERE id = ? FOR UPDATE", fromID,
    ).Scan(&balance)
    if err != nil {
        return fmt.Errorf("查询余额失败: %w", err)
    }

    if balance < amount {
        return fmt.Errorf("余额不足: %.2f < %.2f", balance, amount)
    }

    // 扣款
    _, err = tx.Exec(
        "UPDATE accounts SET balance = balance - ? WHERE id = ?",
        amount, fromID,
    )
    if err != nil {
        return fmt.Errorf("扣款失败: %w", err)
    }

    // 入账
    _, err = tx.Exec(
        "UPDATE accounts SET balance = balance + ? WHERE id = ?",
        amount, toID,
    )
    if err != nil {
        return fmt.Errorf("入账失败: %w", err)
    }

    // 提交事务
    if err := tx.Commit(); err != nil {
        return fmt.Errorf("提交事务失败: %w", err)
    }

    return nil
}

// 通用事务辅助函数
func withTransaction(db *sql.DB, fn func(tx *sql.Tx) error) error {
    tx, err := db.Begin()
    if err != nil {
        return err
    }
    defer tx.Rollback()

    if err := fn(tx); err != nil {
        return err
    }
    return tx.Commit()
}

// 使用
func createOrderTx(db *sql.DB, order *Order) error {
    return withTransaction(db, func(tx *sql.Tx) error {
        // 插入订单
        result, err := tx.Exec("INSERT INTO orders (...) VALUES (...)", ...)
        if err != nil {
            return err
        }
        orderID, _ := result.LastInsertId()

        // 扣减库存
        _, err = tx.Exec("UPDATE products SET stock = stock - ? WHERE id = ?",
            order.Quantity, order.ProductID)
        return err
    })
}

// 批量操作时使用 Prepared Statement 提升性能
func batchInsertUsers(db *sql.DB, users []User) error {
    stmt, err := db.Prepare(
        "INSERT INTO users (name, email) VALUES (?, ?)",
    )
    if err != nil {
        return fmt.Errorf("预编译失败: %w", err)
    }
    defer stmt.Close()

    for _, u := range users {
        _, err := stmt.Exec(u.Name, u.Email)
        if err != nil {
            return fmt.Errorf("插入用户 %s 失败: %w", u.Name, err)
        }
    }
    return nil
}

import (
    "gorm.io/gorm"
    "gorm.io/gorm/logger"
    "gorm.io/driver/mysql"
)

func initDB() (*gorm.DB, error) {
    dsn := "root:password@tcp(127.0.0.1:3306)/mydb?charset=utf8mb4&parseTime=True&loc=Local"

    db, err := gorm.Open(mysql.Open(dsn), &gorm.Config{
        Logger: logger.Default.LogMode(logger.Info), // 打印 SQL
        // 禁用默认事务（单条操作不需要事务时可提升 30% 性能）
        SkipDefaultTransaction: true,
        // 命名策略
        NamingStrategy: schema.NamingStrategy{
            TablePrefix:   "t_",  // 表前缀
            SingularTable: true,  // 禁用表名复数
        },
    })
    if err != nil {
        return nil, err
    }

    // 获取底层 sql.DB 配置连接池
    sqlDB, _ := db.DB()
    sqlDB.SetMaxOpenConns(25)
    sqlDB.SetMaxIdleConns(10)
    sqlDB.SetConnMaxLifetime(30 * time.Minute)

    return db, nil
}

import "gorm.io/gorm"

// User 模型
type User struct {
    gorm.Model                       // 内嵌: ID, CreatedAt, UpdatedAt, DeletedAt
    Name     string   `gorm:"type:varchar(100);not null;index"`
    Email    string   `gorm:"type:varchar(255);uniqueIndex"`
    Age      int      `gorm:"default:0"`
    Role     string   `gorm:"type:varchar(20);default:'user'"`
    Profile  Profile  `gorm:"foreignKey:UserID"` // Has One
    Orders   []Order  `gorm:"foreignKey:UserID"` // Has Many
}

type Profile struct {
    ID     uint   `gorm:"primaryKey"`
    UserID uint   `gorm:"uniqueIndex"`
    Bio    string `gorm:"type:text"`
    Avatar string `gorm:"type:varchar(255)"`
}

type Order struct {
    ID        uint      `gorm:"primaryKey"`
    UserID    uint      `gorm:"index"`
    Product   string    `gorm:"type:varchar(200)"`
    Amount    float64   `gorm:"type:decimal(10,2)"`
    Status    string    `gorm:"type:varchar(20);default:'pending'"`
    CreatedAt time.Time
}

// 自动迁移
func autoMigrate(db *gorm.DB) error {
    return db.AutoMigrate(&User{}, &Profile{}, &Order{})
}

// 创建
func createUser(db *gorm.DB) {
    user := User{
        Name:  "Alice",
        Email: "alice@example.com",
        Age:   25,
        Profile: Profile{
            Bio: "Go 开发者",
        },
    }
    // Create 会自动创建关联的 Profile
    result := db.Create(&user)
    fmt.Printf("ID: %d, 影响行数: %d, 错误: %v\n",
        user.ID, result.RowsAffected, result.Error)
}

// 批量创建
func batchCreate(db *gorm.DB) {
    users := []User{
        {Name: "Bob", Email: "bob@example.com"},
        {Name: "Charlie", Email: "charlie@example.com"},
    }
    db.CreateInBatches(users, 100) // 每批 100 条
}

// 查询
func queryUsers(db *gorm.DB) {
    // 根据主键查询
    var user User
    db.First(&user, 1)                     // SELECT * FROM users WHERE id = 1
    db.First(&user, "email = ?", "alice@example.com") // 条件查询

    // 查询列表
    var users []User
    db.Where("age > ?", 18).
        Order("created_at DESC").
        Limit(10).
        Offset(0).
        Find(&users)

    // 带预加载的查询（解决 N+1 问题）
    db.Preload("Profile").
        Preload("Orders", func(db *gorm.DB) *gorm.DB {
            return db.Where("status = ?", "paid").Order("created_at DESC")
        }).
        Find(&users)

    // 选择特定字段
    type UserDTO struct {
        ID    uint
        Name  string
        Email string
    }
    var dtos []UserDTO
    db.Model(&User{}).
        Select("id, name, email").
        Where("role = ?", "admin").
        Scan(&dtos)

    // 统计
    var count int64
    db.Model(&User{}).Where("age > ?", 18).Count(&count)
}

// 更新
func updateUsers(db *gorm.DB) {
    // 更新单个字段
    db.Model(&User{}).Where("id = ?", 1).Update("name", "NewAlice")

    // 更新多个字段
    db.Model(&User{}).Where("id = ?", 1).Updates(map[string]any{
        "name": "NewAlice",
        "age":  26,
    })

    // 结构体更新（零值字段不更新）
    db.Model(&User{}).Where("id = ?", 1).Updates(User{Name: "NewAlice", Age: 26})

    // 使用 Select 强制更新零值字段
    db.Model(&User{}).Where("id = ?", 1).
        Select("name", "age").
        Updates(User{Name: "NewAlice", Age: 0}) // Age=0 也会被更新
}

// 删除
func deleteUsers(db *gorm.DB) {
    // 软删除（设置 deleted_at 字段）
    db.Delete(&User{}, 1)

    // 永久删除
    db.Unscoped().Delete(&User{}, 1)

    // 查询包含已软删除的记录
    var users []User
    db.Unscoped().Where("name = ?", "Alice").Find(&users)
}

// Scope — 可复用的查询条件
func ActiveUsers(db *gorm.DB) *gorm.DB {
    return db.Where("status = ?", "active")
}

func Paginate(page, pageSize int) func(db *gorm.DB) *gorm.DB {
    return func(db *gorm.DB) *gorm.DB {
        offset := (page - 1) * pageSize
        return db.Offset(offset).Limit(pageSize)
    }
}

// 使用
db.Scopes(ActiveUsers, Paginate(1, 20)).Find(&users)

// 原生 SQL
db.Raw("SELECT * FROM users WHERE age > ? AND role = ?", 18, "admin").Scan(&users)

// 子查询
subQuery := db.Model(&Order{}).Select("user_id").Where("amount > ?", 100)
db.Where("id IN (?)", subQuery).Find(&users)

// Joins
type Result struct {
    UserName  string
    OrderCount int
}
var results []Result
db.Model(&User{}).
    Select("users.name as user_name, COUNT(orders.id) as order_count").
    Joins("LEFT JOIN orders ON orders.user_id = users.id").
    Group("users.id").
    Having("COUNT(orders.id) > ?", 0).
    Scan(&results)

// 手动事务
func createOrderGORM(db *gorm.DB, order *Order) error {
    return db.Transaction(func(tx *gorm.DB) error {
        // 创建订单
        if err := tx.Create(order).Error; err != nil {
            return err // 返回 error 自动 rollback
        }

        // 扣减库存
        result := tx.Model(&Product{}).
            Where("id = ? AND stock >= ?", order.ProductID, order.Quantity).
            Update("stock", gorm.Expr("stock - ?", order.Quantity))

        if result.RowsAffected == 0 {
            return fmt.Errorf("库存不足")
        }

        return nil // 返回 nil 自动 commit
    })
}

// 嵌套事务（SavePoint）
db.Transaction(func(tx *gorm.DB) error {
    tx.Create(&user1)

    tx.Transaction(func(tx2 *gorm.DB) error {
        tx2.Create(&user2)
        return errors.New("回滚内层事务") // 只回滚到 SavePoint
    })

    tx.Create(&user3) // user1 和 user3 会被提交
    return nil
})

import "github.com/jmoiron/sqlx"

// sqlx 在 database/sql 基础上增加了结构体映射

func initSqlx() (*sqlx.DB, error) {
    db, err := sqlx.Connect("mysql",
        "root:password@tcp(127.0.0.1:3306)/mydb?parseTime=True")
    if err != nil {
        return nil, err
    }
    return db, nil
}

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

// 查询单行 — 直接映射到结构体
func getUser(db *sqlx.DB, id int64) (*User, error) {
    var user User
    err := db.Get(&user, "SELECT * FROM users WHERE id = ?", id)
    return &user, err
}

// 查询多行
func listUsers(db *sqlx.DB) ([]User, error) {
    var users []User
    err := db.Select(&users, "SELECT * FROM users ORDER BY id DESC LIMIT 100")
    return users, err
}

// 命名参数
func searchUsers(db *sqlx.DB, name string, minAge int) ([]User, error) {
    var users []User
    query := "SELECT * FROM users WHERE name LIKE :name AND age >= :min_age"
    rows, err := db.NamedQuery(query, map[string]any{
        "name":    "%" + name + "%",
        "min_age": minAge,
    })
    if err != nil {
        return nil, err
    }
    defer rows.Close()

    for rows.Next() {
        var u User
        if err := rows.StructScan(&u); err != nil {
            return nil, err
        }
        users = append(users, u)
    }
    return users, nil
}

// 批量插入
func batchInsert(db *sqlx.DB, users []User) error {
    _, err := db.NamedExec(
        "INSERT INTO users (name, email) VALUES (:name, :email)", users)
    return err
}

// In 查询
func getUsersByIDs(db *sqlx.DB, ids []int64) ([]User, error) {
    query, args, err := sqlx.In(
        "SELECT * FROM users WHERE id IN (?)", ids)
    if err != nil {
        return nil, err
    }
    query = db.Rebind(query) // 适配不同数据库的占位符

    var users []User
    err = db.Select(&users, query, args...)
    return users, err
}

import (
    "context"
    "time"
    "github.com/redis/go-redis/v9"
)

func initRedis() *redis.Client {
    rdb := redis.NewClient(&redis.Options{
        Addr:         "localhost:6379",
        Password:     "",
        DB:           0,
        PoolSize:     20,            // 连接池大小
        MinIdleConns: 5,             // 最小空闲连接数
        DialTimeout:  5 * time.Second,
        ReadTimeout:  3 * time.Second,
        WriteTimeout: 3 * time.Second,
    })

    ctx := context.Background()
    if err := rdb.Ping(ctx).Err(); err != nil {
        log.Fatalf("Redis 连接失败: %v", err)
    }
    return rdb
}

// 基本操作
func basicOps(rdb *redis.Client) {
    ctx := context.Background()

    // String
    rdb.Set(ctx, "user:1:name", "Alice", 10*time.Minute)
    name, _ := rdb.Get(ctx, "user:1:name").Result()

    // 不存在时设置（分布式锁常用）
    ok, _ := rdb.SetNX(ctx, "lock:order:123", "holder-1", 30*time.Second).Result()
    if ok {
        // 获取锁成功
        defer rdb.Del(ctx, "lock:order:123")
    }

    // Hash
    rdb.HSet(ctx, "user:1", map[string]any{
        "name":  "Alice",
        "email": "alice@example.com",
        "age":   25,
    })
    result, _ := rdb.HGetAll(ctx, "user:1").Result()

    // List
    rdb.LPush(ctx, "queue:tasks", "task1", "task2")
    task, _ := rdb.RPop(ctx, "queue:tasks").Result()

    // Set
    rdb.SAdd(ctx, "tags:go", "web", "concurrent", "static")
    members, _ := rdb.SMembers(ctx, "tags:go").Result()

    // Sorted Set
    rdb.ZAdd(ctx, "leaderboard", redis.Z{Score: 100, Member: "Alice"})
    rdb.ZAdd(ctx, "leaderboard", redis.Z{Score: 85, Member: "Bob"})
    top, _ := rdb.ZRevRangeWithScores(ctx, "leaderboard", 0, 9).Result()

    _, _, _, _, _ = name, result, task, members, top // 使用变量
}

// 缓存模式 — Cache Aside
func getUserWithCache(rdb *redis.Client, db *gorm.DB, userID uint) (*User, error) {
    ctx := context.Background()
    cacheKey := fmt.Sprintf("user:%d", userID)

    // 1. 先查缓存
    data, err := rdb.Get(ctx, cacheKey).Bytes()
    if err == nil {
        var user User
        json.Unmarshal(data, &user)
        return &user, nil
    }

    if err != redis.Nil {
        log.Printf("Redis 错误: %v", err) // 日志记录但不返回错误
    }

    // 2. 缓存未命中，查数据库
    var user User
    if err := db.First(&user, userID).Error; err != nil {
        return nil, err
    }

    // 3. 写入缓存（设置过期时间 + 随机偏移防止缓存雪崩）
    ttl := 10*time.Minute + time.Duration(rand.Intn(60))*time.Second
    data, _ = json.Marshal(user)
    rdb.Set(ctx, cacheKey, data, ttl)

    return &user, nil
}

// Pipeline — 批量操作减少网络往返
func batchGet(rdb *redis.Client, keys []string) map[string]string {
    ctx := context.Background()
    pipe := rdb.Pipeline()

    cmds := make(map[string]*redis.StringCmd)
    for _, key := range keys {
        cmds[key] = pipe.Get(ctx, key)
    }
    pipe.Exec(ctx)

    result := make(map[string]string)
    for key, cmd := range cmds {
        if val, err := cmd.Result(); err == nil {
            result[key] = val
        }
    }
    return result
}

// 分布式锁（简易版）
type RedisLock struct {
    client *redis.Client
    key    string
    value  string
    ttl    time.Duration
}

func NewRedisLock(client *redis.Client, key string, ttl time.Duration) *RedisLock {
    return &RedisLock{
        client: client,
        key:    "lock:" + key,
        value:  uuid.New().String(),
        ttl:    ttl,
    }
}

func (l *RedisLock) Lock(ctx context.Context) (bool, error) {
    return l.client.SetNX(ctx, l.key, l.value, l.ttl).Result()
}

func (l *RedisLock) Unlock(ctx context.Context) error {
    // Lua 脚本保证原子性：只有持有者才能释放
    script := redis.NewScript(`
        if redis.call("GET", KEYS[1]) == ARGV[1] then
            return redis.call("DEL", KEYS[1])
        else
            return 0
        end
    `)
    _, err := script.Run(ctx, l.client, []string{l.key}, l.value).Result()
    return err
}

// 仓库接口
type UserRepository interface {
    GetByID(ctx context.Context, id uint) (*User, error)
    GetByEmail(ctx context.Context, email string) (*User, error)
    List(ctx context.Context, page, pageSize int) ([]*User, int64, error)
    Create(ctx context.Context, user *User) error
    Update(ctx context.Context, user *User) error
    Delete(ctx context.Context, id uint) error
}

// GORM 实现
type gormUserRepo struct {
    db *gorm.DB
}

func NewGormUserRepo(db *gorm.DB) UserRepository {
    return &gormUserRepo{db: db}
}

func (r *gormUserRepo) GetByID(ctx context.Context, id uint) (*User, error) {
    var user User
    err := r.db.WithContext(ctx).
        Preload("Profile").
        First(&user, id).Error
    if errors.Is(err, gorm.ErrRecordNotFound) {
        return nil, ErrUserNotFound
    }
    return &user, err
}

func (r *gormUserRepo) List(ctx context.Context, page, pageSize int) ([]*User, int64, error) {
    var users []*User
    var total int64

    db := r.db.WithContext(ctx).Model(&User{})

    if err := db.Count(&total).Error; err != nil {
        return nil, 0, err
    }

    offset := (page - 1) * pageSize
    err := db.Offset(offset).Limit(pageSize).
        Order("created_at DESC").
        Find(&users).Error

    return users, total, err
}

func (r *gormUserRepo) Create(ctx context.Context, user *User) error {
    return r.db.WithContext(ctx).Create(user).Error
}

func (r *gormUserRepo) Update(ctx context.Context, user *User) error {
    return r.db.WithContext(ctx).Save(user).Error
}

func (r *gormUserRepo) Delete(ctx context.Context, id uint) error {
    return r.db.WithContext(ctx).Delete(&User{}, id).Error
}

// ✅ 正确 — 参数化
db.Query("SELECT * FROM users WHERE name = ?", name)

// ❌ 危险 — SQL 注入
db.Query("SELECT * FROM users WHERE name = '" + name + "'")

rows, err := db.Query(...)
if err != nil { return err }
defer rows.Close()  // ⚠️ 不关闭会导致连接泄漏

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
db.QueryRowContext(ctx, "SELECT ...")

// 使用 Select 指定要更新的字段，避免零值被忽略
db.Model(&user).Select("Name", "Age").Updates(User{Name: "new", Age: 0})