golang使用sqlite3，开启wal模式，并发读写

news2025/2/9 7:33:11

因为sqlite是基于文件的，所以默认情况下，sqlite是不支持并发读写的，即写操作会阻塞其他操作，同时sqlite也很容易就产生死锁。

但是作为一个使用广泛的离线数据库，从sqlite3.7.0版本开始（SQLite Release 3.7.0 On 2010-07-21），sqlite引入了更常见的WAL机制来解决页面的读写并发问题。但是sqlite的实现特点决定了其并发能力较低。

SELECT sqlite_version();
3.8.8

开启了WAL模式之后，sqlite就会生成三个文件test.db, test.db-shm, test.db-wal。在WAL模式下支持一写多读。

当临时文件的内容达到一定的量，sqlite会进行一次落盘。

PRAGMA wal_autocheckpoint=5000;

pagesize默认设置的是4k，autocheckpoint设置5000，表示5000个page的数据量，会进行一下checkpoint，也就是20M。

查询日志模式：PRAGMA journal_mode;

设置日志模式：PRAGMA journal_mode=WAL;

示例

CREATE TABLE "users" (
"id"  INTEGER,
"name"  TEXT,
"age"  INTEGER,
"created_at"  TEXT,
"updated_at"  TEXT
);

使用Go的gorm来操作sqlite3

package go_sqlite

import (
	"fmt"
	"strconv"
	"sync"

	"time"

	"database/sql"

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

var dbfile = "demos/go_sqlite/test.db"

func Run() {
	gormDB, sqlDB, err := InitDB()
	if err != nil {
		panic(err)
	}
	defer sqlDB.Close()

	users := []User{}
	for i := 0; i < 1000; i++ {
		user := User{
			Name:      "user_" + strconv.Itoa(i),
			Age:       uint8(i % 100),
			CreatedAt: time.Now().Unix(),
			UpdatedAt: time.Now().Unix(),
		}
		users = append(users, user)
	}
	err = BatchInsertUsers(gormDB, users)
	if err != nil {
		panic(err)
	}

	users, err = GetUsers(gormDB)
	if err != nil {
		panic(err)
	}

	fmt.Println(len(users))
	fmt.Println(users[0])
}

type User struct {
	ID        uint
	Name      string
	Age       uint8
	CreatedAt int64
	UpdatedAt int64
}

func InitDB() (*gorm.DB, *sql.DB, error) {
	gormDB, err := gorm.Open(sqlite.Open(dbfile), &gorm.Config{})
	if err != nil {
		return nil, nil, err
	}
	sqlDB, _ := gormDB.DB()
	gormDB.Exec("PRAGMA journal_mode=WAL;")
	sqlDB.SetMaxIdleConns(10)
	sqlDB.SetMaxOpenConns(100)
	sqlDB.SetConnMaxLifetime(time.Hour)

	return gormDB, sqlDB, nil
}

func BatchInsertUsers(gormDB *gorm.DB, users []User) error {
	batchSize := 100
	batchCount := (len(users) + batchSize - 1) / batchSize
	for i := 0; i < batchCount; i++ {
		start := i * batchSize
		end := (i + 1) * batchSize
		if end > len(users) {
			end = len(users)
		}
		batch := users[start:end]
		tx := gormDB.Begin()
		if err := tx.Error; err != nil {
			return err
		}
		if err := tx.Create(&batch).Error; err != nil {
			tx.Rollback()
			return err
		}
		if err := tx.Commit().Error; err != nil {
			return err
		}
	}
	return nil
}

func GetUsers(gormDB *gorm.DB) ([]User, error) {
	var users []User
	err := gormDB.Find(&users).Error
	if err != nil {
		return nil, err
	}
	return users, nil
}

并发测试

var wg sync.WaitGroup

func Run2() {
	gormDB, err := gorm.Open(sqlite.Open(dbfile), &gorm.Config{})
	if err != nil {
		panic("failed to connect database")
	}
	gormDB.Exec("PRAGMA journal_mode=WAL;")
	sqlDB, _ := gormDB.DB()
	sqlDB.SetMaxIdleConns(10)
	sqlDB.SetMaxOpenConns(100)

	wg.Add(2000)

	// 并发写入 1000 条数据
	for i := 0; i < 1000; i++ {
		go func(i int) {
			defer wg.Done()
			err := gormDB.Transaction(func(tx *gorm.DB) error {
				user := User{Name: fmt.Sprintf("user_%d", i)}
				result := tx.Create(&user)
				return result.Error
			})
			if err != nil {
				fmt.Printf("failed to write data: %v\n", err)
			}
		}(i)
	}

	// 并发读取数据
	for i := 0; i < 1000; i++ {
		go func() {
			defer wg.Done()
			var users []User
			err := gormDB.Transaction(func(tx *gorm.DB) error {
				result := tx.Find(&users)
				return result.Error
			})
			if err != nil {
				fmt.Printf("failed to read data: %v\n", err)
			} else {
				fmt.Printf("read %d records\n", len(users))
			}
		}()
	}

	wg.Wait()

	fmt.Println("done")
}