在 Go 中实现基于 MySQL 的分布式读写锁，可以通过数据库来协调不同服务或节点对共享资源的访问。在这种模式下，数据库充当锁的存储和协调者，所有的操作通过对数据库表的读写操作来控制锁的获取与释放。
假设我们创建一个名为 distributed_locks 的表，包含以下字段：
Life is like a journey
lock_name：锁的名称，标识哪个资源被锁定。lock_type：锁的类型，可以是 read 或 write。lock_owner：持锁的节点标识，用于区分不同的请求者。lock_time：锁的时间戳，用于判断锁是否超时。CREATE TABLE distributed_locks (
lock_name VARCHAR(255) PRIMARY KEY,
lock_type ENUM('read', 'write'),
lock_owner VARCHAR(255),
lock_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
以下是一个简单的 Go 示例，展示了如何使用 MySQL 实现分布式读写锁。
package main
import (
"database/sql"
"fmt"
"log"
"time"
_ "github.com/go-sql-driver/mysql"
)
var db *sql.DB
func init() {
var err error
// 假设你的 MySQL 数据库用户名和密码是 root/root，数据库名称是 locks
dsn := "root:root@tcp(localhost:3306)/locks"
db, err = sql.Open("mysql", dsn)
if err != nil {
log.Fatalf("Failed to connect to database: %v", err)
}
// 验证数据库连接
if err := db.Ping(); err != nil {
log.Fatalf("Failed to ping database: %v", err)
}
}
func closeDB() {
if err := db.Close(); err != nil {
log.Fatalf("Failed to close DB: %v", err)
}
}
func acquireReadLock(lockName, nodeName string) error {
// 检查是否有写锁
var existingLockType string
var existingLockOwner string
err := db.QueryRow("SELECT lock_type, lock_owner FROM distributed_locks WHERE lock_name = ?", lockName).Scan(&existingLockType, &existingLockOwner)
if err == nil && existingLockType == "write" {
return fmt.Errorf("write lock already held by %s", existingLockOwner)
}
// 尝试插入一个读锁
_, err = db.Exec("INSERT INTO distributed_locks (lock_name, lock_type, lock_owner) VALUES (?, 'read', ?) ON DUPLICATE KEY UPDATE lock_owner = ?", lockName, nodeName, nodeName)
if err != nil {
return fmt.Errorf("failed to acquire read lock: %v", err)
}
return nil
}
func acquireWriteLock(lockName, nodeName string) error {
// 检查是否有读锁或写锁
var existingLockType string
var existingLockOwner string
err := db.QueryRow("SELECT lock_type, lock_owner FROM distributed_locks WHERE lock_name = ?", lockName).Scan(&existingLockType, &existingLockOwner)
if err == nil && (existingLockType == "read" || existingLockType == "write") {
return fmt.Errorf("lock already held by %s", existingLockOwner)
}
// 尝试插入一个写锁
_, err = db.Exec("INSERT INTO distributed_locks (lock_name, lock_type, lock_owner) VALUES (?, 'write', ?) ON DUPLICATE KEY UPDATE lock_owner = ?", lockName, nodeName, nodeName)
if err != nil {
return fmt.Errorf("failed to acquire write lock: %v", err)
}
return nil
}
func releaseLock(lockName, nodeName string) error {
// 删除锁表中当前节点持有的锁
_, err := db.Exec("DELETE FROM distributed_locks WHERE lock_name = ? AND lock_owner = ?", lockName, nodeName)
if err != nil {
return fmt.Errorf("failed to release lock: %v", err)
}
return nil
}
func main() {
defer closeDB()
lockName := "my_resource"
nodeName := "node1" // 假设每个节点都有唯一的名字
// 请求写锁
err := acquireWriteLock(lockName, nodeName)
if err != nil {
log.Println("Failed to acquire write lock:", err)
} else {
log.Println("Write lock acquired")
// 执行写操作...
time.Sleep(3 * time.Second) // 模拟写操作
// 释放写锁
err := releaseLock(lockName, nodeName)
if err != nil {
log.Println("Failed to release lock:", err)
} else {
log.Println("Write lock released")
}
}
// 请求读锁
err = acquireReadLock(lockName, nodeName)
if err != nil {
log.Println("Failed to acquire read lock:", err)
} else {
log.Println("Read lock acquired")
// 执行读操作...
time.Sleep(1 * time.Second) // 模拟读操作
// 释放读锁
err := releaseLock(lockName, nodeName)
if err != nil {
log.Println("Failed to release lock:", err)
} else {
log.Println("Read lock released")
}
}
}
这种实现方式适用于分布式系统中，不同节点需要协作访问共享资源的场景。然而，数据库本身并不是为了实现分布式锁而设计的，因此需要特别小心性能和死锁问题。