Golang教程|麻瓜教程

Appearance

07-性能优化与缓存 ​

随着业务量增长，数据库往往成为系统瓶颈。GORM 提供了多种性能优化手段：预加载、批量操作、游标查询、连接池调优、日志监控等。在高并发场景下，缓存是减轻数据库压力的关键。本篇将系统讲解 GORM 的查询与写入性能优化、日志配置以及 Redis 缓存集成。

查询性能优化 ​

预加载避免 N+1 ​

N+1 查询问题是性能杀手：1 次查询获取 N 条主记录，再 N 次查询获取关联数据，总共 N+1 次查询。使用 Preload 可以将关联数据一次性加载，减少到 2 次查询。

go
package main

import (
	"fmt"
	"log"
	"time"

	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
	"gorm.io/gorm/logger"
)

type User struct {
	ID    uint   `gorm:"primaryKey"`
	Name  string
	Posts []Post
}

type Post struct {
	ID     uint   `gorm:"primaryKey"`
	UserID uint
	Title  string
}

func main() {
	db, err := gorm.Open(sqlite.Open("perf.db"), &gorm.Config{
		Logger: logger.Default.LogMode(logger.Warn),
	})
	if err != nil {
		log.Fatal(err)
	}
	db.AutoMigrate(&User{}, &Post{})

	// 准备数据
	for i := 1; i <= 10; i++ {
		user := User{Name: fmt.Sprintf("user_%d", i)}
		db.Create(&user)
		for j := 1; j <= 5; j++ {
			db.Create(&Post{UserID: user.ID, Title: fmt.Sprintf("post_%d_%d", i, j)})
		}
	}

	// 错误做法：N+1 查询（10 用户 + 10 次查询 = 11 次）
	start := time.Now()
	var users []User
	db.Find(&users)
	for i := range users {
		db.Model(&users[i]).Related(&users[i].Posts)
	}
	fmt.Printf("N+1 用时: %v, 查询 %d 次\n", time.Since(start), 1+len(users))

	// 正确做法：Preload 预加载（2 次查询）
	start = time.Now()
	var users2 []User
	db.Preload("Posts").Find(&users2)
	fmt.Printf("Preload 用时: %v, 查询 2 次\n", time.Since(start))
	fmt.Printf("共加载 %d 用户和文章\n", len(users2))
}

Select 只查需要的字段 ​

避免 SELECT *，只查询业务需要的字段，减少数据传输和内存占用。

go
package main

import (
	"fmt"
	"log"

	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
)

type User struct {
	ID        uint   `gorm:"primaryKey"`
	Name      string `gorm:"type:varchar(50)"`
	Email     string `gorm:"type:varchar(150)"`
	Password  string `gorm:"type:varchar(100)"`
	Bio       string `gorm:"type:text"`
	Avatar    string `gorm:"type:varchar(255)"`
	Age       int
	City      string
}

func main() {
	db, err := gorm.Open(sqlite.Open("select.db"), &gorm.Config{})
	if err != nil {
		log.Fatal(err)
	}
	db.AutoMigrate(&User{})

	// 准备数据
	db.Create(&User{
		Name: "Tom", Email: "tom@x.com", Password: "secret",
		Bio: "很长的个人简介...", Avatar: "tom.png", Age: 25, City: "Beijing",
	})

	// 错误：SELECT * 查询所有字段（包括不需要的 Password、Bio）
	var users []User
	db.Find(&users)

	// 正确：只查需要的字段
	type UserBrief struct {
		ID   uint
		Name string
		Age  int
	}
	var briefs []UserBrief
	db.Model(&User{}).Select("id, name, age").Scan(&briefs)
	fmt.Printf("精简查询: %+v\n", briefs)

	// 列表场景：只查 id 和 name
	var names []struct {
		ID   uint
		Name string
	}
	db.Model(&User{}).Select("id, name").Scan(&names)
	fmt.Printf("仅 id+name: %+v\n", names)
}

使用索引 ​

确保查询条件字段有索引，可以极大提升查询速度。

go
package main

import (
	"fmt"
	"log"
	"time"

	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
)

type User struct {
	ID      uint   `gorm:"primaryKey"`
	Name    string `gorm:"type:varchar(50);index"`        // 普通索引
	Email   string `gorm:"type:varchar(150);uniqueIndex"` // 唯一索引
	City    string `gorm:"type:varchar(50);index:idx_city_age,priority:1"`
	Age     int    `gorm:"index:idx_city_age,priority:2"` // 复合索引
	Status  int    `gorm:"default:1;index"`
	CreatedAt time.Time
}

func main() {
	db, err := gorm.Open(sqlite.Open("indexed.db"), &gorm.Config{})
	if err != nil {
		log.Fatal(err)
	}
	db.AutoMigrate(&User{})

	// 准备 1000 条数据
	for i := 1; i <= 1000; i++ {
		db.Create(&User{
			Name:      fmt.Sprintf("user_%d", i),
			Email:     fmt.Sprintf("user_%d@x.com", i),
			City:      []string{"Beijing", "Shanghai", "Shenzhen"}[i%3],
			Age:       20 + i%50,
			Status:    i % 2,
			CreatedAt: time.Now(),
		})
	}

	// 查询1：使用索引（city 在复合索引前缀）
	start := time.Now()
	var users []User
	db.Where("city = ?", "Beijing").Find(&users)
	fmt.Printf("索引查询 city: %v, %d 条\n", time.Since(start), len(users))

	// 查询2：使用复合索引
	start = time.Now()
	db.Where("city = ? AND age > ?", "Beijing", 30).Find(&users)
	fmt.Printf("复合索引查询: %v, %d 条\n", time.Since(start), len(users))

	// 查询3：唯一索引
	start = time.Now()
	var user User
	db.Where("email = ?", "user_500@x.com").First(&user)
	fmt.Printf("唯一索引查询: %v\n", time.Since(start))

	// 反例：不在索引前缀的查询（age 在复合索引第二位，单独查询无法使用）
	start = time.Now()
	db.Where("age > ?", 50).Find(&users)
	fmt.Printf("非前缀查询 age: %v, %d 条\n", time.Since(start), len(users))
}

分批查询：FindInBatches ​

处理大量数据时，一次性加载到内存会导致 OOM。FindInBatches 分批读取数据，每批处理指定数量。

go
package main

import (
	"fmt"
	"log"

	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
)

type User struct {
	ID    uint `gorm:"primaryKey"`
	Name  string
	Email string
}

func main() {
	db, err := gorm.Open(sqlite.Open("batch.db"), &gorm.Config{})
	if err != nil {
		log.Fatal(err)
	}
	db.AutoMigrate(&User{})

	// 准备 1000 条数据
	users := make([]User, 1000)
	for i := range users {
		users[i] = User{Name: fmt.Sprintf("user_%d", i), Email: fmt.Sprintf("user_%d@x.com", i)}
	}
	db.CreateInBatches(users, 100)

	// 分批查询处理
	var allUsers []User
	result := db.Where("id > ?", 0).FindInBatches(&allUsers, 100, func(tx *gorm.DB, batch int) error {
		fmt.Printf("处理第 %d 批, %d 条记录\n", batch, len(allUsers))

		// 对每批数据进行处理
		for _, u := range allUsers {
			_ = u.Name // 业务处理...
		}

		// 可以返回错误中止处理
		// return errors.New("中止处理")
		return nil
	})
	fmt.Printf("总处理 %d 条, 错误: %v\n", result.RowsAffected, result.Error)
}

游标查询：Rows 和 Scan ​

对于超大结果集，Rows 提供游标式读取，逐行处理，内存占用极低。

go
package main

import (
	"fmt"
	"log"

	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
)

type User struct {
	ID    uint `gorm:"primaryKey"`
	Name  string
	Email string
}

func main() {
	db, err := gorm.Open(sqlite.Open("rows.db"), &gorm.Config{})
	if err != nil {
		log.Fatal(err)
	}
	db.AutoMigrate(&User{})

	for i := 1; i <= 100; i++ {
		db.Create(&User{Name: fmt.Sprintf("user_%d", i), Email: fmt.Sprintf("user_%d@x.com", i)})
	}

	// 方式1：Rows + Scan 逐行处理
	rows, err := db.Model(&User{}).Where("id > ?", 0).Rows()
	if err != nil {
		log.Fatal(err)
	}
	defer rows.Close()

	count := 0
	for rows.Next() {
		var user User
		if err := db.ScanRows(rows, &user); err != nil {
			log.Fatal(err)
		}
		count++
		if count <= 3 {
			fmt.Printf("读取: %s\n", user.Name)
		}
	}
	fmt.Printf("共读取 %d 条\n", count)

	// 方式2：Pluck 流式读取单列
	var names []string
	rows2, _ := db.Model(&User{}).Select("name").Rows()
	defer rows2.Close()
	for rows2.Next() {
		var name string
		rows2.Scan(&name)
		names = append(names, name)
	}
	fmt.Printf("Pluck 读取 %d 个名字\n", len(names))
}

写入性能优化 ​

批量插入：CreateInBatches ​

CreateInBatches 分批插入，避免单条插入的开销，也避免 SQL 过长。

go
package main

import (
	"fmt"
	"log"
	"time"

	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
)

type User struct {
	ID    uint `gorm:"primaryKey"`
	Name  string
	Email string
}

func main() {
	db, err := gorm.Open(sqlite.Open("insert.db"), &gorm.Config{})
	if err != nil {
		log.Fatal(err)
	}
	db.AutoMigrate(&User{})

	// 准备 1000 条数据
	users := make([]User, 1000)
	for i := range users {
		users[i] = User{Name: fmt.Sprintf("user_%d", i), Email: fmt.Sprintf("user_%d@x.com", i)}
	}

	// 方式1：逐条插入（慢）
	start := time.Now()
	for i := 0; i < 100; i++ { // 只测 100 条对比
		db.Create(&users[i])
	}
	fmt.Printf("逐条插入 100 条: %v\n", time.Since(start))

	// 方式2：一次性批量插入
	start = time.Now()
	db.Create(&users[100:1000])
	fmt.Printf("批量插入 900 条: %v\n", time.Since(start))

	// 方式3：CreateInBatches 分批插入（推荐）
	db.Where("1 = 1").Delete(&User{})
	users2 := make([]User, 1000)
	for i := range users2 {
		users2[i] = User{Name: fmt.Sprintf("user_%d", i), Email: fmt.Sprintf("user_%d@x.com", i)}
	}
	start = time.Now()
	db.CreateInBatches(users2, 100) // 每批 100 条
	fmt.Printf("CreateInBatches 1000 条: %v\n", time.Since(start))
}

批量更新：使用 Case When ​

GORM 没有直接的批量更新不同值的 API，但可以用 Case When 构造 SQL：

go
package main

import (
	"fmt"
	"log"

	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
)

type Product struct {
	ID    uint    `gorm:"primaryKey"`
	Name  string
	Price float64
	Stock int
}

func main() {
	db, err := gorm.Open(sqlite.Open("case.db"), &gorm.Config{})
	if err != nil {
		log.Fatal(err)
	}
	db.AutoMigrate(&Product{})

	// 准备数据
	products := []Product{
		{Name: "A", Price: 10, Stock: 100},
		{Name: "B", Price: 20, Stock: 200},
		{Name: "C", Price: 30, Stock: 300},
	}
	db.Create(&products)

	// 需求：根据 ID 批量更新不同价格
	// 等价 SQL: UPDATE products SET price = CASE id WHEN 1 THEN 11 WHEN 2 THEN 22 WHEN 3 THEN 33 END WHERE id IN (1,2,3)
	updates := []struct {
		ID    uint
		Price float64
	}{
		{ID: 1, Price: 11},
		{ID: 2, Price: 22},
		{ID: 3, Price: 33},
	}

	// 构建 CASE WHEN 表达式
	caseStmt := "CASE id"
	ids := make([]interface{}, len(updates))
	for i, u := range updates {
		caseStmt += fmt.Sprintf(" WHEN %d THEN %f", u.ID, u.Price)
		ids[i] = u.ID
	}
	caseStmt += " END"

	// 执行批量更新
	result := db.Model(&Product{}).
		Where("id IN ?", updates[0:1]). // 简化演示
		Update("price", gorm.Expr(caseStmt))
	_ = result

	// 更简单的批量更新方式：循环 Update
	for _, u := range updates {
		db.Model(&Product{}).Where("id = ?", u.ID).Update("price", u.Price)
	}

	// 验证
	var all []Product
	db.Find(&all)
	for _, p := range all {
		fmt.Printf("  %s: price=%.0f\n", p.Name, p.Price)
	}
}

Upsert：冲突时更新 ​

Upsert（Update or Insert）在主键或唯一键冲突时执行更新，避免先查后写的两次操作。

go
package main

import (
	"fmt"
	"log"

	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

type User struct {
	ID    uint   `gorm:"primaryKey"`
	Name  string `gorm:"type:varchar(50)"`
	Email string `gorm:"type:varchar(150);uniqueIndex"`
	Age   int
}

func main() {
	db, err := gorm.Open(sqlite.Open("upsert.db"), &gorm.Config{})
	if err != nil {
		log.Fatal(err)
	}
	db.AutoMigrate(&User{})

	// 第一次插入
	user := User{Name: "Tom", Email: "tom@x.com", Age: 25}
	db.Create(&user)

	// Upsert：邮箱冲突时什么都不做
	newUser := User{Name: "Tom2", Email: "tom@x.com", Age: 30}
	result := db.Clauses(clause.OnConflict{DoNothing: true}).Create(&newUser)
	fmt.Printf("DoNothing 影响行数: %d\n", result.RowsAffected)

	// Upsert：邮箱冲突时更新指定字段
	newUser2 := User{Name: "Tom3", Email: "tom@x.com", Age: 35}
	result = db.Clauses(clause.OnConflict{
		Columns:   []clause.Column{{Name: "email"}}, // 冲突检测列
		DoUpdates: clause.AssignmentColumns([]string{"name", "age"}), // 更新这些列
	}).Create(&newUser2)
	fmt.Printf("UpdateAll 影响行数: %d\n", result.RowsAffected)

	// 验证
	var found User
	db.Where("email = ?", "tom@x.com").First(&found)
	fmt.Printf("最终: name=%s, age=%d\n", found.Name, found.Age)

	// Upsert：自定义更新表达式
	newUser3 := User{Name: "Tom4", Email: "tom@x.com", Age: 40}
	db.Clauses(clause.OnConflict{
		Columns: []clause.Column{{Name: "email"}},
		DoUpdates: clause.Assignments(map[string]interface{}{
			"age": gorm.Expr("age + ?", 1), // 年龄 +1
		}),
	}).Create(&newUser3)
	db.Where("email = ?", "tom@x.com").First(&found)
	fmt.Printf("自定义表达式后 age=%d\n", found.Age)
}

Map 更新避免零值问题 ​

如前所述，结构体更新会忽略零值字段。使用 map[string]interface{} 可以解决。

go
package main

import (
	"fmt"
	"log"

	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
)

type User struct {
	ID     uint   `gorm:"primaryKey"`
	Name   string
	Status int `gorm:"default:1"`
	Age    int
}

func main() {
	db, err := gorm.Open(sqlite.Open("map.db"), &gorm.Config{})
	if err != nil {
		log.Fatal(err)
	}
	db.AutoMigrate(&User{})

	user := User{Name: "Tom", Status: 1, Age: 25}
	db.Create(&user)

	// 错误：结构体更新零值被忽略
	db.Model(&user).Updates(User{Status: 0, Age: 30})
	var found User
	db.First(&found, user.ID)
	fmt.Printf("结构体更新后: status=%d (未更新), age=%d\n", found.Status, found.Age)

	// 正确：Map 更新包含零值
	db.Model(&user).Updates(map[string]interface{}{
		"status": 0,
		"age":    30,
	})
	db.First(&found, user.ID)
	fmt.Printf("Map 更新后: status=%d, age=%d\n", found.Status, found.Age)
}

连接池调优 ​

连接池配置直接影响并发性能。根据业务负载调整四个参数：

go
package main

import (
	"fmt"
	"log"
	"sync"
	"time"

	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
)

type User struct {
	ID   uint `gorm:"primaryKey"`
	Name string
}

func main() {
	db, err := gorm.Open(sqlite.Open("pool.db"), &gorm.Config{})
	if err != nil {
		log.Fatal(err)
	}
	db.AutoMigrate(&User{})

	sqlDB, _ := db.DB()

	// 连接池参数调优
	sqlDB.SetMaxIdleConns(20)              // 最大空闲连接（建议 10-20）
	sqlDB.SetMaxOpenConns(100)             // 最大打开连接（根据 DB 配置）
	sqlDB.SetConnMaxLifetime(30 * time.Minute) // 连接最大存活时间
	sqlDB.SetConnMaxIdleTime(5 * time.Minute)  // 连接最大空闲时间

	// 模拟并发查询
	var wg sync.WaitGroup
	start := time.Now()
	for i := 0; i < 50; i++ {
		wg.Add(1)
		go func(n int) {
			defer wg.Done()
			user := User{Name: fmt.Sprintf("user_%d", n)}
			db.Create(&user)
		}(i)
	}
	wg.Wait()
	fmt.Printf("50 并发写入用时: %v\n", time.Since(start))

	// 查看连接池状态
	stats := sqlDB.Stats()
	fmt.Printf("当前连接: %d, 空闲: %d, 等待: %d\n",
		stats.InUse, stats.Idle, stats.WaitCount)
}

GORM 日志配置 ​

日志级别 ​

GORM 提供四个日志级别：Silent、Error、Warn、Info。

go
package main

import (
	"log"
	"os"
	"time"

	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
	"gorm.io/gorm/logger"
)

type User struct {
	ID   uint `gorm:"primaryKey"`
	Name string
}

func main() {
	// 自定义 Logger
	newLogger := logger.New(
		log.New(os.Stdout, "\r\n", log.LstdFlags),
		logger.Config{
			SlowThreshold:             200 * time.Millisecond, // 慢查询阈值
			LogLevel:                  logger.Info,             // 日志级别
			IgnoreRecordNotFoundError: true,                     // 忽略 ErrRecordNotFound
			Colorful:                  true,                     // 彩色输出
		},
	)

	db, err := gorm.Open(sqlite.Open("log.db"), &gorm.Config{
		Logger: newLogger,
	})
	if err != nil {
		log.Fatal(err)
	}
	db.AutoMigrate(&User{})

	// 所有 SQL 都会被打印（Info 级别）
	db.Create(&User{Name: "Tom"})

	var user User
	db.First(&user, 1)

	// 运行时动态调整日志级别
	db.Debug().First(&user, 1) // 临时开启详细日志

	// 针对单个会话设置
	// db.Session(&gorm.Session{Logger: ...}).
}

自定义 Logger ​

实现 logger.Interface 可以将日志输出到任意目标（如文件、ELK）。

go
package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
	"gorm.io/gorm/logger"
)

// 自定义 Logger
type CustomLogger struct {
	logger.Config
	fileLogger *log.Logger
}

func NewCustomLogger() *CustomLogger {
	file, _ := os.OpenFile("gorm.log", os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
	return &CustomLogger{
		Config: logger.Config{
			SlowThreshold: 200 * time.Millisecond,
			LogLevel:      logger.Warn,
			Colorful:      false,
		},
		fileLogger: log.New(file, "", log.LstdFlags),
	}
}

func (l *CustomLogger) LogMode(level logger.LogLevel) logger.Interface {
	newLogger := *l
	newLogger.LogLevel = level
	return &newLogger
}

func (l *CustomLogger) Info(ctx context.Context, msg string, data ...interface{}) {
	if l.LogLevel >= logger.Info {
		l.fileLogger.Printf("[INFO] "+msg, data...)
	}
}

func (l *CustomLogger) Warn(ctx context.Context, msg string, data ...interface{}) {
	if l.LogLevel >= logger.Warn {
		l.fileLogger.Printf("[WARN] "+msg, data...)
	}
}

func (l *CustomLogger) Error(ctx context.Context, msg string, data ...interface{}) {
	if l.LogLevel >= logger.Error {
		l.fileLogger.Printf("[ERROR] "+msg, data...)
	}
}

func (l *CustomLogger) Trace(ctx context.Context, begin time.Time, fc func() (string, int64), err error) {
	elapsed := time.Since(begin)
	sql, rows := fc()
	if err != nil {
		l.fileLogger.Printf("[ERROR] %s (%d rows) %v", sql, rows, err)
	} else if elapsed > l.SlowThreshold {
		l.fileLogger.Printf("[SLOW] %s (%d rows) %v", sql, rows, elapsed)
	}
}

type User struct {
	ID   uint `gorm:"primaryKey"`
	Name string
}

func main() {
	db, err := gorm.Open(sqlite.Open("customlog.db"), &gorm.Config{
		Logger: NewCustomLogger(),
	})
	if err != nil {
		log.Fatal(err)
	}
	db.AutoMigrate(&User{})

	db.Create(&User{Name: "Tom"})
	var user User
	db.First(&user, 1)
	fmt.Println("操作完成，请查看 gorm.log 文件")
}

慢查询日志 ​

通过设置 SlowThreshold 自动记录慢查询：

go
package main

import (
	"log"
	"os"
	"time"

	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
	"gorm.io/gorm/logger"
)

type User struct {
	ID   uint `gorm:"primaryKey"`
	Name string
}

func main() {
	newLogger := logger.New(
		log.New(os.Stdout, "\r\n", log.LstdFlags),
		logger.Config{
			SlowThreshold: 100 * time.Millisecond, // 超过 100ms 视为慢查询
			LogLevel:      logger.Warn,             // Warn 级别会打印慢查询
			Colorful:      true,
		},
	)

	db, err := gorm.Open(sqlite.Open("slow.db"), &gorm.Config{
		Logger: newLogger,
	})
	if err != nil {
		log.Fatal(err)
	}
	db.AutoMigrate(&User{})

	// 准备大量数据
	for i := 0; i < 10000; i++ {
		db.Create(&User{Name: "user"})
	}

	// 慢查询（无索引 + 大量数据）
	var users []User
	db.Where("name = ?", "user").Find(&users)
	// 控制台会输出慢查询警告
}

缓存集成 ​

基于 Redis 的查询缓存 ​

高并发场景下，将热点数据缓存到 Redis 可以极大降低数据库压力。

go
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"time"

	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
)

// 模拟 Redis 客户端
type RedisClient struct {
	data map[string]string
}

func NewRedisClient() *RedisClient {
	return &RedisClient{data: make(map[string]string)}
}

func (r *RedisClient) Get(ctx context.Context, key string) (string, error) {
	if v, ok := r.data[key]; ok {
		return v, nil
	}
	return "", fmt.Errorf("not found")
}

func (r *RedisClient) Set(ctx context.Context, key string, value string, ttl time.Duration) error {
	r.data[key] = value
	return nil
}

func (r *RedisClient) Del(ctx context.Context, key string) error {
	delete(r.data, key)
	return nil
}

type User struct {
	ID    uint   `gorm:"primaryKey"`
	Name  string
	Email string
}

// UserCacheService 带缓存的用户服务
type UserCacheService struct {
	db    *gorm.DB
	redis *RedisClient
}

func NewUserCacheService(db *gorm.DB, redis *RedisClient) *UserCacheService {
	return &UserCacheService{db: db, redis: redis}
}

func (s *UserCacheService) GetByID(ctx context.Context, id uint) (*User, error) {
	key := fmt.Sprintf("user:%d", id)

	// 1. 先查缓存
	if data, err := s.redis.Get(ctx, key); err == nil {
		var user User
		if err := json.Unmarshal([]byte(data), &user); err == nil {
			fmt.Println("命中缓存")
			return &user, nil
		}
	}

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

	// 3. 写入缓存
	data, _ := json.Marshal(user)
	s.redis.Set(ctx, key, string(data), 5*time.Minute)

	return &user, nil
}

func (s *UserCacheService) Update(ctx context.Context, user *User) error {
	// 更新数据库
	if err := s.db.Save(user).Error; err != nil {
		return err
	}
	// 失效缓存
	key := fmt.Sprintf("user:%d", user.ID)
	return s.redis.Del(ctx, key)
}

func main() {
	db, err := gorm.Open(sqlite.Open("cache.db"), &gorm.Config{})
	if err != nil {
		log.Fatal(err)
	}
	db.AutoMigrate(&User{})

	// 创建测试用户
	user := User{Name: "Tom", Email: "tom@x.com"}
	db.Create(&user)

	redis := NewRedisClient()
	svc := NewUserCacheService(db, redis)
	ctx := context.Background()

	// 第一次查询：未命中缓存
	u1, _ := svc.GetByID(ctx, user.ID)
	fmt.Printf("第一次: %s\n", u1.Name)

	// 第二次查询：命中缓存
	u2, _ := svc.GetByID(ctx, user.ID)
	fmt.Printf("第二次: %s\n", u2.Name)

	// 更新后缓存失效
	u2.Name = "Tom Updated"
	svc.Update(ctx, u2)

	// 第三次查询：缓存已失效，重新加载
	u3, _ := svc.GetByID(ctx, user.ID)
	fmt.Printf("第三次: %s\n", u3.Name)
}

缓存失效策略 ​

常见的缓存失效策略：

  1. Cache-Aside：读时缓存，写时失效（如上例）
  2. Write-Through：写时同步更新缓存
  3. Write-Behind：写时只更新缓存，异步刷新数据库
  4. TTL 过期：设置过期时间，定期失效
go
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"sync"
	"time"

	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
)

type Product struct {
	ID    uint    `gorm:"primaryKey"`
	Name  string
	Price float64
	Stock int
}

type RedisClient struct {
	data map[string]string
	mu   sync.RWMutex
}

func NewRedis() *RedisClient {
	return &RedisClient{data: make(map[string]string)}
}

func (r *RedisClient) Get(key string) (string, bool) {
	r.mu.RLock()
	defer r.mu.RUnlock()
	v, ok := r.data[key]
	return v, ok
}

func (r *RedisClient) Set(key, value string, ttl time.Duration) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.data[key] = value
}

func (r *RedisClient) Del(key string) {
	r.mu.Lock()
	defer r.mu.Unlock()
	delete(r.data, key)
}

type ProductService struct {
	db    *gorm.DB
	redis *RedisClient
}

func NewProductService(db *gorm.DB, redis *RedisClient) *ProductService {
	return &ProductService{db: db, redis: redis}
}

// GetProductMulti 多级缓存：本地缓存 -> Redis -> DB
func (s *ProductService) GetProduct(id uint) (*Product, error) {
	key := fmt.Sprintf("product:%d", id)

	// 1. Redis 缓存
	if data, ok := s.redis.Get(key); ok {
		var p Product
		if err := json.Unmarshal([]byte(data), &p); err == nil {
			fmt.Println("Redis 命中")
			return &p, nil
		}
	}

	// 2. 数据库
	fmt.Println("查询数据库")
	var p Product
	if err := s.db.First(&p, id).Error; err != nil {
		return nil, err
	}

	// 3. 写入 Redis
	data, _ := json.Marshal(p)
	s.redis.Set(key, string(data), 10*time.Minute)
	return &p, nil
}

// UpdateProduct 更新并失效缓存
func (s *ProductService) UpdateProduct(p *Product) error {
	if err := s.db.Save(p).Error; err != nil {
		return err
	}
	// 失效缓存
	s.redis.Del(fmt.Sprintf("product:%d", p.ID))
	return nil
}

func main() {
	db, err := gorm.Open(sqlite.Open("strategy.db"), &gorm.Config{})
	if err != nil {
		log.Fatal(err)
	}
	db.AutoMigrate(&Product{})

	// 准备数据
	product := Product{Name: "iPhone", Price: 7999, Stock: 100}
	db.Create(&product)

	svc := NewProductService(db, NewRedis())

	// 查询流程演示
	p1, _ := svc.GetProduct(product.ID)
	fmt.Printf("第一次: %s 价格=%.0f\n", p1.Name, p1.Price)

	p2, _ := svc.GetProduct(product.ID)
	fmt.Printf("第二次: %s 价格=%.0f\n", p2.Name, p2.Price)

	// 更新后缓存失效
	p2.Price = 6999
	svc.UpdateProduct(p2)

	p3, _ := svc.GetProduct(product.ID)
	fmt.Printf("更新后: %s 价格=%.0f\n", p3.Name, p3.Price)

	_ = context.Background()
}

完整示例：高并发商品查询 ​

下面用一个完整的高并发商品查询示例，综合演示缓存和批量查询：

go
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"sync"
	"time"

	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
	"gorm.io/gorm/logger"
)

type Product struct {
	ID          uint    `gorm:"primaryKey"`
	Name        string  `gorm:"type:varchar(200);index"`
	Price       float64
	Stock       int
	CategoryID  uint    `gorm:"index"`
	Status      int     `gorm:"default:1;index"`
	Description string  `gorm:"type:text"`
	CreatedAt   time.Time
}

type Category struct {
	ID   uint   `gorm:"primaryKey"`
	Name string `gorm:"type:varchar(100)"`
}

// 模拟 Redis
type Redis struct {
	data map[string]string
	mu   sync.RWMutex
}

func NewRedis() *Redis {
	return &Redis{data: make(map[string]string)}
}

func (r *Redis) Get(key string) (string, bool) {
	r.mu.RLock()
	defer r.mu.RUnlock()
	v, ok := r.data[key]
	return v, ok
}

func (r *Redis) Set(key, value string, ttl time.Duration) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.data[key] = value
}

func (r *Redis) Del(key string) {
	r.mu.Lock()
	defer r.mu.Unlock()
	delete(r.data, key)
}

// ProductService 商品服务
type ProductService struct {
	db    *gorm.DB
	redis *Redis
}

func NewProductService(db *gorm.DB, redis *Redis) *ProductService {
	return &ProductService{db: db, redis: redis}
}

// GetByID 单商品查询（带缓存）
func (s *ProductService) GetByID(ctx context.Context, id uint) (*Product, error) {
	key := fmt.Sprintf("product:%d", id)
	if data, ok := s.redis.Get(key); ok {
		var p Product
		if json.Unmarshal([]byte(data), &p) == nil {
			return &p, nil
		}
	}

	var p Product
	if err := s.db.First(&p, id).Error; err != nil {
		return nil, err
	}
	data, _ := json.Marshal(p)
	s.redis.Set(key, string(data), 5*time.Minute)
	return &p, nil
}

// GetByIDs 批量查询（带缓存）
func (s *ProductService) GetByIDs(ctx context.Context, ids []uint) ([]Product, error) {
	// 1. 先从缓存批量获取
	result := make([]Product, 0, len(ids))
	missIDs := make([]uint, 0)
	for _, id := range ids {
		key := fmt.Sprintf("product:%d", id)
		if data, ok := s.redis.Get(key); ok {
			var p Product
			if json.Unmarshal([]byte(data), &p) == nil {
				result = append(result, p)
				continue
			}
		}
		missIDs = append(missIDs, id)
	}

	// 2. 缓存未命中的批量查询数据库
	if len(missIDs) > 0 {
		var dbProducts []Product
		if err := s.db.Where("id IN ?", missIDs).Find(&dbProducts).Error; err != nil {
			return nil, err
		}
		// 写入缓存
		for i := range dbProducts {
			key := fmt.Sprintf("product:%d", dbProducts[i].ID)
			data, _ := json.Marshal(dbProducts[i])
			s.redis.Set(key, string(data), 5*time.Minute)
		}
		result = append(result, dbProducts...)
	}
	return result, nil
}

// Search 搜索商品（分批 + Select）
func (s *ProductService) Search(ctx context.Context, keyword string, page, size int) ([]Product, int64, error) {
	if page <= 0 {
		page = 1
	}
	if size <= 0 || size > 100 {
		size = 10
	}
	offset := (page - 1) * size

	query := s.db.Model(&Product{}).
		Select("id, name, price, stock, category_id").
		Where("status = ? AND name LIKE ?", 1, "%"+keyword+"%")

	var total int64
	query.Count(&total)

	var products []Product
	err := query.Order("id desc").Offset(offset).Limit(size).Find(&products).Error
	return products, total, err
}

func main() {
	db, err := gorm.Open(sqlite.Open("ecommerce_perf.db"), &gorm.Config{
		Logger: logger.Default.LogMode(logger.Error),
	})
	if err != nil {
		log.Fatal(err)
	}
	db.AutoMigrate(&Product{}, &Category{})

	// 准备数据
	db.Create(&Category{Name: "电子产品"})
	for i := 1; i <= 100; i++ {
		db.Create(&Product{
			Name:        fmt.Sprintf("商品_%d", i),
			Price:       float64(10 + i),
			Stock:       100,
			CategoryID:  1,
			Status:      1,
			Description: "商品描述",
			CreatedAt:   time.Now(),
		})
	}

	svc := NewProductService(db, NewRedis())
	ctx := context.Background()

	// 1. 单商品查询（带缓存）
	fmt.Println("=== 单商品查询 ===")
	start := time.Now()
	p, _ := svc.GetByID(ctx, 1)
	fmt.Printf("首次查询: %s (%v)\n", p.Name, time.Since(start))

	start = time.Now()
	p, _ = svc.GetByID(ctx, 1)
	fmt.Printf("缓存命中: %s (%v)\n", p.Name, time.Since(start))

	// 2. 批量查询
	fmt.Println("\n=== 批量查询 ===")
	start = time.Now()
	products, _ := svc.GetByIDs(ctx, []uint{1, 2, 3, 4, 5})
	fmt.Printf("批量首次: %d 个商品 (%v)\n", len(products), time.Since(start))

	start = time.Now()
	products, _ = svc.GetByIDs(ctx, []uint{1, 2, 3, 4, 5})
	fmt.Printf("批量缓存: %d 个商品 (%v)\n", len(products), time.Since(start))

	// 3. 搜索
	fmt.Println("\n=== 搜索商品 ===")
	start = time.Now()
	results, total, _ := svc.Search(ctx, "商品", 1, 10)
	fmt.Printf("搜索 '商品': %d 条, 当前页 %d 条 (%v)\n", total, len(results), time.Since(start))

	// 4. 分批处理
	fmt.Println("\n=== 分批处理 ===")
	var allProducts []Product
	db.FindInBatches(&allProducts, 20, func(tx *gorm.DB, batch int) error {
		fmt.Printf("处理第 %d 批: %d 条\n", batch, len(allProducts))
		return nil
	})
}

小结 ​

本篇系统讲解了 GORM 的性能优化与缓存：

  1. 预加载避免 N+1：Preload 将 N+1 查询降为 2 次
  2. Select 精简字段：避免 SELECT *，减少数据传输
  3. 使用索引：确保查询条件命中索引
  4. FindInBatches 分批查询：避免大结果集 OOM
  5. Rows 游标查询：逐行处理超大结果集
  6. CreateInBatches 批量插入：远快于逐条插入
  7. Case When 批量更新：一条 SQL 更新多行不同值
  8. Upsert 冲突更新：避免先查后写
  9. Map 更新解决零值问题：精确控制更新字段
  10. 连接池调优：四个参数合理配置
  11. 日志配置：日志级别、自定义 Logger、慢查询日志
  12. Redis 缓存：Cache-Aside 模式、多级缓存、批量查询缓存

下一篇我们将学习 GORM 最佳实践与架构模式，掌握企业级应用的架构设计。

Copyright © 2025-2026 京公网安备 11010102000001号 京ICP备2024078128号-1