Golang 基于 MySQL 实现分布式读写锁

业务场景与核心需求

对于刚上线且暂不引入其他中间件的项目，可以直接复用现有的 MySQL 数据库来实现分布式读写锁。该方案适用于以下分布式环境：

• 分布式的多节点运行：不同的服务实例运行在不同的机器或容器节点中。
• 读操作的共享性：允许同时存在多个 Goroutine 对同一资源进行读操作，无需阻塞。
• 写操作的互斥性：同一时刻仅允许一个 Goroutine 对资源执行写操作。
• 读写互斥：当资源处于写锁状态时，禁止任何读操作；反之亦然（读锁不影响读，但会阻塞写）。

什么是分布式读写锁

在单节点环境中，Golang 原生的 sync.Mutex 足以处理并发安全问题。然而，当应用部署在分布式架构下时，单机锁无法跨进程通信，此时需要依赖数据库或外部缓存构建分布式锁。分布式读写锁相比单纯的排他锁，粒度更细，提供了更灵活的状态管理：

• 读模式（ReadLock）：资源共享，多个客户端可并行获取。
• 写模式（WriteLock）：资源互斥，独占访问。
• 无锁模式（Unlock）：空闲状态，接受任何请求。

访问原则与控制逻辑

分布式读写锁遵循以下严格的访问控制原则：

读锁机制详解

• 获取条件：只要当前状态不是“写锁”，即可尝试获取读锁。
• 共享策略：在“读锁”状态下，后续请求仍可获取读锁，系统会自动累加计数器。
• 阻塞逻辑：只要有读锁存在，写锁请求必须等待所有读锁解锁后，才能转为写锁状态。

写锁机制详解

• 获取条件：只有当前状态严格为“无锁”时，才允许获取写锁。
• 独占策略：一旦获取写锁，将阻塞后续所有的读锁和写锁请求。

GORM 连接与数据表设计

第1步：配置数据库连接与结构体

首先需要通过 GORM 建立连接并定义模型。核心在于设计包含锁状态标记和计数器字段的结构体。

package main

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

var (
	db *gorm.DB

	// 配置信息
	dbUsername = "kele"
	dbPassword = "baishi2020"
	dbHost     = "127.0.0.1:7306"
	dbDatabase = "lingmo"

	// 状态常量
	stateReadLock  = "ReadLock"
	stateWriteLock = "WriteLock"
	stateUnlock    = "Unlock"
)

// RWLock 定义分布式读写锁的核心字段
type RWLock struct {
	LockMark      string `gorm:"default:'Unlock'"` // 锁状态标识
	ReadLockCount uint32 `gorm:"default:0"`        // 活跃读锁计数器
	LockReason    string                        // 加锁原因记录
}

// Stock 模拟库存业务数据表
type Stock struct {
	gorm.Model
	RWLock
	Count int64
}

func (Stock) TableName() string {
	return "stocks"
}

// 初始化数据库连接与迁移
func init() {
	dsn := fmt.Sprintf("%s:%s@tcp(%s)/%s?charset=utf8mb4&parseTime=True&loc=Local", dbUsername, dbPassword, dbHost, dbDatabase)

	mysqlConfig := mysql.Config{DSN: dsn}
	gormConfig := &gorm.Config{Logger: logger.Default.LogMode(logger.Info)}

	var err error
	if db, err = gorm.Open(mysql.New(mysqlConfig), gormConfig); err != nil {
		panic(err)
	}

	// 设置引擎为 InnoDB 以支持事务与行锁
	db.Set("db:table_options", "ENGINE = InnoDB DEFAULT CHARSET = utf8")

	// 自动创建/迁移表结构
	if err = db.AutoMigrate(&Stock{}); err != nil {
		panic(err)
	}
}

func main() {
	// 初始化测试数据
	if result := db.Model(&Stock{}).Save(&Stock{Model: gorm.Model{}, RWLock: RWLock{}, Count: 10}); result.Error != nil {
		panic(result.Error)
	}
}

在上述结构中：LockMark 用于标识当前是读锁、写锁还是无锁；ReadLockCount 用于统计当前并发的读请求数量，是实现共享读的关键；LockReason 记录触发锁的原因，便于调试与审计。

第2步：实现读锁 (RLock/RUnlock)

实现读锁时，关键在于利用 SQL 表达式原子性地增加计数器，并通过条件过滤排除写锁状态。

func (s Stock) RLock(db *gorm.DB, lockReason string) error {
	// 条件：指定ID且当前不是写锁状态
	condition := "(id = ?) AND (lock_mark != ?)"
	fields := map[string]interface{}{
		"lock_mark":       stateReadLock,
		"read_lock_count": gorm.Expr("read_lock_count + ?", 1), // 原子自增
		"lock_reason":     lockReason,
	}

	result := db.Model(&Stock{}).Where(condition, s.ID, stateWriteLock).Updates(fields)
	if result.Error != nil {
		return result.Error
	}
	if result.RowsAffected == 0 {
		return fmt.Errorf("failed to rlock Stock, RowsAffected=0 (likely blocked by write lock)")
	}

	return nil
}

func (s Stock) RUnlock(db *gorm.DB, unlockReason string) error {
	// 使用原生 SQL 处理复杂的 IF 逻辑：判断计数归零后再清除状态
	sql := fmt.Sprintf(`
	UPDATE stocks 
	SET read_lock_count=IF(read_lock_count>0,read_lock_count-1,0), 
	    lock_mark=IF(read_lock_count<1, 'Unlock', 'ReadLock'),
	    lock_reason = '%s' 
	WHERE id = %d AND lock_mark='%s'`, unlockReason, s.ID, stateReadLock)
	
	result := db.Exec(sql)
	if result.Error != nil {
		return result.Error
	}
	if result.RowsAffected == 0 {
		return fmt.Errorf("failed to RUnlock Stock, RowsAffected=0")
	}

	return nil
}

// 调用示例
func testRUnlock(s Stock) {
	_ = s.RLock(db, "readLock_reason_1")
	_ = s.RLock(db, "readLock_reason_2")
	// 并发读成功...
	_ = s.RUnlock(db, "readLock_unlock_1")
	_ = s.RUnlock(db, "readLock_unlock_2")
}

生成的 SQL 分析：

• 加读锁：如果当前状态非 WriteLock，则直接更新为 ReadLock 并将计数+1。
• 解除读锁：若计数减1后仍大于0，状态维持 ReadLock；若计数归零，则状态转为 Unlock。

第3步：实现写锁 (WLock/WUnlock)

写锁具有排他性，要求当前必须是完全的无锁状态（Unlock），且成功后需强制重置读锁计数。

func (s Stock) WLock(db *gorm.DB, lockReason string) error {
	// 只有完全无锁状态才能获取写锁
	condition := "(id = ?) AND (lock_mark = ?)"
	fields := map[string]interface{}{
		"lock_mark":       stateWriteLock,
		"read_lock_count": 0,             // 写锁介入，清除读计数
		"lock_reason":     lockReason,
	}

	result := db.Model(&Stock{}).Where(condition, s.ID, stateUnlock).Updates(fields)
	if result.Error != nil {
		return result.Error
	}
	if result.RowsAffected == 0 {
		return fmt.Errorf("failed to WLock Stock, likely blocked by existing locks")
	}

	return nil
}

func (s Stock) WUnlock(db *gorm.DB, unlockReason string) error {
	condition := "(id = ?) AND (lock_mark = ?)"
	fields := map[string]interface{}{
		"lock_mark":       stateUnlock,
		"read_lock_count": 0,
		"lock_reason":     unlockReason,
	}

	result := db.Model(&Stock{}).Where(condition, s.ID, stateWriteLock).Updates(fields)
	if result.Error != nil {
		return result.Error
	}
	if result.RowsAffected == 0 {
		return fmt.Errorf("failed to WUnlock Stock, RowsAffected=0")
	}

	return nil
}

// 调用示例
func testWLock(s Stock) {
	_ = s.WLock(db, "writeLock_reason_1")
	// 进行独占写操作...
	_ = s.WUnlock(db, "unlockWrite_reason_1")
}

生成的 SQL 分析：

• 加写锁：严格匹配 LockMark='Unlock'，成功后将 LockMark 改为 WriteLock，Count 清零。
• 解除写锁：仅在 LockMark='WriteLock' 时执行回退。

总结

虽然市面上存在 Etcd、Redis 等成熟的分布式协调组件，但在轻量级或初期场景中，利用 Golang 配合 MySQL 实现简单的分布式读写锁是一种务实的选择。该方法避免了额外的基础设施维护成本，借助数据库的行级锁特性保障了跨节点并发下的数据一致性与正确性。

以上就是基于 Golang 和 MySQL 实现分布式读写锁的详细内容，更多关于 Golang 高并发控制与数据库锁设计的资料请关注本站其它相关文章！