Go学习笔记-Database操作

2025.3.2 2025.9.21 编程语言 4744 10 分钟

数据库操作是现代应用程序的核心功能。Go语言提供了database/sql包作为标准的数据库接口，支持多种数据库驱动，提供了连接池、事务、预处理语句等功能。

数据库基础操作

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285


package main

import (
    "database/sql"
    "fmt"
    "log"
    "time"
    
    _ "github.com/mattn/go-sqlite3" // SQLite驱动
)

// 1. 用户结构体
type User struct {
    ID        int       `json:"id"`
    Name      string    `json:"name"`
    Email     string    `json:"email"`
    Age       int       `json:"age"`
    CreatedAt time.Time `json:"created_at"`
    UpdatedAt time.Time `json:"updated_at"`
}

// 2. 数据库连接和初始化
func initDatabase() *sql.DB {
    // 连接SQLite数据库
    db, err := sql.Open("sqlite3", "./test.db")
    if err != nil {
        log.Fatal("连接数据库失败:", err)
    }
    
    // 测试连接
    if err := db.Ping(); err != nil {
        log.Fatal("数据库连接测试失败:", err)
    }
    
    // 设置连接池参数
    db.SetMaxOpenConns(25)                 // 最大打开连接数
    db.SetMaxIdleConns(25)                 // 最大空闲连接数
    db.SetConnMaxLifetime(5 * time.Minute) // 连接最大生存时间
    
    fmt.Println("数据库连接成功")
    return db
}

// 3. 创建表
func createTables(db *sql.DB) {
    createUserTable := `
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE NOT NULL,
        age INTEGER,
        created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
        updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
    );
    `
    
    _, err := db.Exec(createUserTable)
    if err != nil {
        log.Fatal("创建用户表失败:", err)
    }
    
    // 创建索引
    createEmailIndex := `CREATE INDEX IF NOT EXISTS idx_users_email ON users(email);`
    _, err = db.Exec(createEmailIndex)
    if err != nil {
        log.Fatal("创建邮箱索引失败:", err)
    }
    
    fmt.Println("数据表创建成功")
}

// 4. 插入数据
func insertUser(db *sql.DB, user User) (int64, error) {
    query := `
    INSERT INTO users (name, email, age, created_at, updated_at)
    VALUES (?, ?, ?, ?, ?)
    `
    
    now := time.Now()
    result, err := db.Exec(query, user.Name, user.Email, user.Age, now, now)
    if err != nil {
        return 0, fmt.Errorf("插入用户失败: %v", err)
    }
    
    id, err := result.LastInsertId()
    if err != nil {
        return 0, fmt.Errorf("获取插入ID失败: %v", err)
    }
    
    return id, nil
}

// 5. 查询单个用户
func getUserByID(db *sql.DB, id int) (*User, error) {
    query := `
    SELECT id, name, email, age, created_at, updated_at
    FROM users
    WHERE id = ?
    `
    
    var user User
    row := db.QueryRow(query, id)
    
    err := row.Scan(&user.ID, &user.Name, &user.Email, &user.Age, 
        &user.CreatedAt, &user.UpdatedAt)
    if err != nil {
        if err == sql.ErrNoRows {
            return nil, fmt.Errorf("用户不存在")
        }
        return nil, fmt.Errorf("查询用户失败: %v", err)
    }
    
    return &user, nil
}

// 6. 查询多个用户
func getAllUsers(db *sql.DB) ([]User, error) {
    query := `
    SELECT id, name, email, age, created_at, updated_at
    FROM users
    ORDER BY created_at DESC
    `
    
    rows, err := db.Query(query)
    if err != nil {
        return nil, fmt.Errorf("查询用户列表失败: %v", err)
    }
    defer rows.Close()
    
    var users []User
    for rows.Next() {
        var user User
        err := rows.Scan(&user.ID, &user.Name, &user.Email, &user.Age,
            &user.CreatedAt, &user.UpdatedAt)
        if err != nil {
            return nil, fmt.Errorf("扫描用户数据失败: %v", err)
        }
        users = append(users, user)
    }
    
    // 检查遍历过程中的错误
    if err := rows.Err(); err != nil {
        return nil, fmt.Errorf("遍历用户数据失败: %v", err)
    }
    
    return users, nil
}

// 7. 更新用户
func updateUser(db *sql.DB, id int, user User) error {
    query := `
    UPDATE users
    SET name = ?, email = ?, age = ?, updated_at = ?
    WHERE id = ?
    `
    
    result, err := db.Exec(query, user.Name, user.Email, user.Age, 
        time.Now(), id)
    if err != nil {
        return fmt.Errorf("更新用户失败: %v", err)
    }
    
    rowsAffected, err := result.RowsAffected()
    if err != nil {
        return fmt.Errorf("获取影响行数失败: %v", err)
    }
    
    if rowsAffected == 0 {
        return fmt.Errorf("用户不存在或数据未变更")
    }
    
    return nil
}

// 8. 删除用户
func deleteUser(db *sql.DB, id int) error {
    query := `DELETE FROM users WHERE id = ?`
    
    result, err := db.Exec(query, id)
    if err != nil {
        return fmt.Errorf("删除用户失败: %v", err)
    }
    
    rowsAffected, err := result.RowsAffected()
    if err != nil {
        return fmt.Errorf("获取影响行数失败: %v", err)
    }
    
    if rowsAffected == 0 {
        return fmt.Errorf("用户不存在")
    }
    
    return nil
}

// 9. 基本操作示例
func basicDatabaseOperations() {
    fmt.Println("=== 基本数据库操作 ===")
    
    // 初始化数据库
    db := initDatabase()
    defer db.Close()
    
    // 创建表
    createTables(db)
    
    // 插入用户
    users := []User{
        {Name: "张三", Email: "zhangsan@example.com", Age: 30},
        {Name: "李四", Email: "lisi@example.com", Age: 25},
        {Name: "王五", Email: "wangwu@example.com", Age: 35},
    }
    
    fmt.Println("插入用户:")
    for _, user := range users {
        id, err := insertUser(db, user)
        if err != nil {
            log.Printf("插入用户失败: %v", err)
            continue
        }
        fmt.Printf("  插入用户成功，ID: %d\n", id)
    }
    
    // 查询所有用户
    fmt.Println("\n查询所有用户:")
    allUsers, err := getAllUsers(db)
    if err != nil {
        log.Printf("查询用户失败: %v", err)
        return
    }
    
    for _, user := range allUsers {
        fmt.Printf("  ID: %d, 姓名: %s, 邮箱: %s, 年龄: %d\n",
            user.ID, user.Name, user.Email, user.Age)
    }
    
    // 查询单个用户
    fmt.Println("\n查询单个用户 (ID=1):")
    user, err := getUserByID(db, 1)
    if err != nil {
        log.Printf("查询用户失败: %v", err)
    } else {
        fmt.Printf("  用户信息: %+v\n", user)
    }
    
    // 更新用户
    fmt.Println("\n更新用户 (ID=1):")
    updateData := User{
        Name:  "张三(已更新)",
        Email: "zhangsan_updated@example.com",
        Age:   31,
    }
    
    err = updateUser(db, 1, updateData)
    if err != nil {
        log.Printf("更新用户失败: %v", err)
    } else {
        fmt.Println("  用户更新成功")
        
        // 验证更新
        updatedUser, _ := getUserByID(db, 1)
        fmt.Printf("  更新后用户信息: %+v\n", updatedUser)
    }
    
    // 删除用户
    fmt.Println("\n删除用户 (ID=3):")
    err = deleteUser(db, 3)
    if err != nil {
        log.Printf("删除用户失败: %v", err)
    } else {
        fmt.Println("  用户删除成功")
    }
    
    // 查看最终用户列表
    fmt.Println("\n最终用户列表:")
    finalUsers, _ := getAllUsers(db)
    for _, user := range finalUsers {
        fmt.Printf("  ID: %d, 姓名: %s, 邮箱: %s, 年龄: %d\n",
            user.ID, user.Name, user.Email, user.Age)
    }
}

func main() {
    basicDatabaseOperations()
}

预处理语句和事务

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291


package main

import (
    "database/sql"
    "fmt"
    "log"
    "time"
    
    _ "github.com/mattn/go-sqlite3"
)

// 1. 预处理语句示例
func preparedStatements(db *sql.DB) {
    fmt.Println("=== 预处理语句 ===")
    
    // 准备插入语句
    insertStmt, err := db.Prepare(`
        INSERT INTO users (name, email, age, created_at, updated_at)
        VALUES (?, ?, ?, ?, ?)
    `)
    if err != nil {
        log.Fatal("准备插入语句失败:", err)
    }
    defer insertStmt.Close()
    
    // 准备查询语句
    selectStmt, err := db.Prepare(`
        SELECT id, name, email, age FROM users WHERE age > ?
    `)
    if err != nil {
        log.Fatal("准备查询语句失败:", err)
    }
    defer selectStmt.Close()
    
    // 批量插入数据
    users := []User{
        {Name: "预处理用户1", Email: "prep1@example.com", Age: 28},
        {Name: "预处理用户2", Email: "prep2@example.com", Age: 32},
        {Name: "预处理用户3", Email: "prep3@example.com", Age: 24},
    }
    
    fmt.Println("使用预处理语句批量插入:")
    now := time.Now()
    for _, user := range users {
        result, err := insertStmt.Exec(user.Name, user.Email, user.Age, now, now)
        if err != nil {
            log.Printf("插入失败: %v", err)
            continue
        }
        
        id, _ := result.LastInsertId()
        fmt.Printf("  插入用户: %s (ID: %d)\n", user.Name, id)
    }
    
    // 使用预处理语句查询
    fmt.Println("\n使用预处理语句查询年龄大于25的用户:")
    rows, err := selectStmt.Query(25)
    if err != nil {
        log.Printf("查询失败: %v", err)
        return
    }
    defer rows.Close()
    
    for rows.Next() {
        var user User
        err := rows.Scan(&user.ID, &user.Name, &user.Email, &user.Age)
        if err != nil {
            log.Printf("扫描数据失败: %v", err)
            continue
        }
        fmt.Printf("  ID: %d, 姓名: %s, 年龄: %d\n", user.ID, user.Name, user.Age)
    }
}

// 2. 事务处理
func transactionExample(db *sql.DB) {
    fmt.Println("\n=== 事务处理 ===")
    
    // 开始事务
    tx, err := db.Begin()
    if err != nil {
        log.Fatal("开始事务失败:", err)
    }
    
    // 使用defer确保事务被正确处理
    defer func() {
        if p := recover(); p != nil {
            tx.Rollback()
            panic(p) // 重新抛出panic
        } else if err != nil {
            tx.Rollback()
        } else {
            err = tx.Commit()
        }
    }()
    
    // 在事务中执行多个操作
    fmt.Println("在事务中执行操作:")
    
    // 操作1: 插入用户
    insertQuery := `
        INSERT INTO users (name, email, age, created_at, updated_at)
        VALUES (?, ?, ?, ?, ?)
    `
    now := time.Now()
    result, err := tx.Exec(insertQuery, "事务用户1", "tx1@example.com", 29, now, now)
    if err != nil {
        log.Printf("事务中插入用户1失败: %v", err)
        return
    }
    
    user1ID, _ := result.LastInsertId()
    fmt.Printf("  插入用户1成功，ID: %d\n", user1ID)
    
    // 操作2: 插入另一个用户
    result, err = tx.Exec(insertQuery, "事务用户2", "tx2@example.com", 33, now, now)
    if err != nil {
        log.Printf("事务中插入用户2失败: %v", err)
        return
    }
    
    user2ID, _ := result.LastInsertId()
    fmt.Printf("  插入用户2成功，ID: %d\n", user2ID)
    
    // 操作3: 更新第一个用户
    updateQuery := `UPDATE users SET age = ? WHERE id = ?`
    _, err = tx.Exec(updateQuery, 30, user1ID)
    if err != nil {
        log.Printf("事务中更新用户失败: %v", err)
        return
    }
    
    fmt.Printf("  更新用户1年龄成功\n")
    
    // 如果所有操作都成功，事务将在defer中被提交
    fmt.Println("  事务中的所有操作都成功，准备提交")
}

// 3. 事务回滚示例
func transactionRollbackExample(db *sql.DB) {
    fmt.Println("\n=== 事务回滚示例 ===")
    
    // 获取操作前的用户数量
    var countBefore int
    db.QueryRow("SELECT COUNT(*) FROM users").Scan(&countBefore)
    fmt.Printf("操作前用户数量: %d\n", countBefore)
    
    // 开始事务
    tx, err := db.Begin()
    if err != nil {
        log.Fatal("开始事务失败:", err)
    }
    
    // 故意制造一个会失败的事务
    defer func() {
        if err != nil {
            tx.Rollback()
            fmt.Println("  事务已回滚")
        } else {
            tx.Commit()
            fmt.Println("  事务已提交")
        }
    }()
    
    // 操作1: 插入一个正常用户
    insertQuery := `
        INSERT INTO users (name, email, age, created_at, updated_at)
        VALUES (?, ?, ?, ?, ?)
    `
    now := time.Now()
    _, err = tx.Exec(insertQuery, "回滚测试用户", "rollback@example.com", 25, now, now)
    if err != nil {
        log.Printf("插入用户失败: %v", err)
        return
    }
    fmt.Println("  插入用户成功")
    
    // 操作2: 尝试插入重复邮箱的用户（会失败）
    _, err = tx.Exec(insertQuery, "重复邮箱用户", "rollback@example.com", 26, now, now)
    if err != nil {
        log.Printf("插入重复邮箱用户失败: %v", err)
        return // 这会触发回滚
    }
    
    // 这行代码不会执行，因为上面的操作会失败
    fmt.Println("  所有操作成功")
    
    // 检查操作后的用户数量
    var countAfter int
    db.QueryRow("SELECT COUNT(*) FROM users").Scan(&countAfter)
    fmt.Printf("操作后用户数量: %d\n", countAfter)
    
    if countAfter == countBefore {
        fmt.Println("  确认事务已回滚，用户数量未变化")
    }
}

// 4. 连接池监控
func connectionPoolMonitoring(db *sql.DB) {
    fmt.Println("\n=== 连接池监控 ===")
    
    // 获取数据库统计信息
    stats := db.Stats()
    
    fmt.Printf("连接池统计信息:\n")
    fmt.Printf("  最大打开连接数: %d\n", stats.MaxOpenConnections)
    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("  等待连接总时间: %v\n", stats.WaitDuration)
    fmt.Printf("  最大空闲关闭数: %d\n", stats.MaxIdleClosed)
    fmt.Printf("  最大生存时间关闭数: %d\n", stats.MaxLifetimeClosed)
}

// 5. 批量操作
func batchOperations(db *sql.DB) {
    fmt.Println("\n=== 批量操作 ===")
    
    // 开始事务进行批量插入
    tx, err := db.Begin()
    if err != nil {
        log.Fatal("开始事务失败:", err)
    }
    defer tx.Rollback() // 如果没有显式提交，则回滚
    
    // 准备批量插入语句
    stmt, err := tx.Prepare(`
        INSERT INTO users (name, email, age, created_at, updated_at)
        VALUES (?, ?, ?, ?, ?)
    `)
    if err != nil {
        log.Fatal("准备语句失败:", err)
    }
    defer stmt.Close()
    
    // 批量插入数据
    batchSize := 1000
    fmt.Printf("批量插入 %d 条记录:\n", batchSize)
    
    start := time.Now()
    now := time.Now()
    
    for i := 0; i < batchSize; i++ {
        name := fmt.Sprintf("批量用户%d", i+1)
        email := fmt.Sprintf("batch%d@example.com", i+1)
        age := 20 + (i % 50) // 年龄在20-69之间
        
        _, err := stmt.Exec(name, email, age, now, now)
        if err != nil {
            log.Printf("批量插入第%d条记录失败: %v", i+1, err)
            return
        }
        
        if (i+1)%100 == 0 {
            fmt.Printf("  已插入 %d 条记录\n", i+1)
        }
    }
    
    // 提交事务
    err = tx.Commit()
    if err != nil {
        log.Printf("提交事务失败: %v", err)
        return
    }
    
    duration := time.Since(start)
    fmt.Printf("批量插入完成，耗时: %v\n", duration)
    fmt.Printf("平均每条记录耗时: %v\n", duration/time.Duration(batchSize))
    
    // 验证插入结果
    var count int
    db.QueryRow("SELECT COUNT(*) FROM users").Scan(&count)
    fmt.Printf("当前数据库中总用户数: %d\n", count)
}

func main() {
    // 初始化数据库
    db := initDatabase()
    defer db.Close()
    
    // 创建表
    createTables(db)
    
    // 演示各种数据库操作
    preparedStatements(db)
    transactionExample(db)
    transactionRollbackExample(db)
    connectionPoolMonitoring(db)
    batchOperations(db)
}

总结

基本操作：
- sql.Open() - 连接数据库
- db.Exec() - 执行SQL语句
- db.Query() / db.QueryRow() - 查询数据
- rows.Scan() - 扫描结果到变量
预处理语句：
- db.Prepare() - 准备SQL语句
- 提高性能和安全性
- 防止SQL注入攻击
- 适合重复执行的操作
事务处理：
- db.Begin() - 开始事务
- tx.Commit() - 提交事务
- tx.Rollback() - 回滚事务
- 确保数据一致性
连接池管理：
- SetMaxOpenConns() - 最大连接数
- SetMaxIdleConns() - 最大空闲连接数
- SetConnMaxLifetime() - 连接生存时间
- 监控连接池状态
最佳实践：
- 正确处理错误和资源清理
- 使用预处理语句防止SQL注入
- 合理使用事务保证数据一致性
- 监控和调优连接池参数
- 批量操作提高性能

作者：JerryWang1996
链接：https://wjinlei.github.io/posts/code-go/code-go-021-database/
许可：CC BY-NC-SA 4.0

go