feat: init project with distributed lock implementation
- Add DistributedLock with Acquire/Release/Refresh using Lua scripts - Add AcquireWithRetry with configurable retry interval and max retries - Add integration tests for all lock operations - Add .env for Redis connection (gitignored)
This commit is contained in:
+113
@@ -0,0 +1,113 @@
|
||||
package lock
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrLockNotAcquired = errors.New("lock not acquired")
|
||||
ErrLockNotHeld = errors.New("lock not held by this owner")
|
||||
)
|
||||
|
||||
// DistributedLock 基于 Redis 的分布式锁
|
||||
type DistributedLock struct {
|
||||
client *redis.Client
|
||||
key string
|
||||
owner string // 唯一标识，用于安全释放锁
|
||||
ttl time.Duration
|
||||
}
|
||||
|
||||
// New 创建一个分布式锁实例
|
||||
// - client: Redis 客户端
|
||||
// - key: 锁的 key
|
||||
// - owner: 持有者唯一标识（如 UUID）
|
||||
// - ttl: 锁的过期时间，防止死锁
|
||||
func New(client *redis.Client, key string, owner string, ttl time.Duration) *DistributedLock {
|
||||
return &DistributedLock{
|
||||
client: client,
|
||||
key: key,
|
||||
owner: owner,
|
||||
ttl: ttl,
|
||||
}
|
||||
}
|
||||
|
||||
// Acquire 尝试获取锁（非阻塞）
|
||||
func (l *DistributedLock) Acquire(ctx context.Context) (bool, error) {
|
||||
ok, err := l.client.SetNX(ctx, l.key, l.owner, l.ttl).Result()
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return ok, nil
|
||||
}
|
||||
|
||||
// AcquireWithRetry 带重试的获取锁
|
||||
// - retryInterval: 重试间隔
|
||||
// - maxRetries: 最大重试次数，0 表示无限重试
|
||||
func (l *DistributedLock) AcquireWithRetry(ctx context.Context, retryInterval time.Duration, maxRetries int) (bool, error) {
|
||||
for i := 0; maxRetries == 0 || i < maxRetries; i++ {
|
||||
acquired, err := l.Acquire(ctx)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if acquired {
|
||||
return true, nil
|
||||
}
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return false, ctx.Err()
|
||||
case <-time.After(retryInterval):
|
||||
}
|
||||
}
|
||||
return false, ErrLockNotAcquired
|
||||
}
|
||||
|
||||
// Release 释放锁（仅当自己持有时才释放，使用 Lua 脚本保证原子性）
|
||||
func (l *DistributedLock) Release(ctx context.Context) error {
|
||||
script := redis.NewScript(`
|
||||
if redis.call("GET", KEYS[1]) == ARGV[1] then
|
||||
return redis.call("DEL", KEYS[1])
|
||||
else
|
||||
return 0
|
||||
end
|
||||
`)
|
||||
|
||||
result, err := script.Run(ctx, l.client, []string{l.key}, l.owner).Int64()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if result == 0 {
|
||||
return ErrLockNotHeld
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Refresh 续期锁（延长 TTL，仅当自己持有时才续期）
|
||||
func (l *DistributedLock) Refresh(ctx context.Context) (bool, error) {
|
||||
script := redis.NewScript(`
|
||||
if redis.call("GET", KEYS[1]) == ARGV[1] then
|
||||
return redis.call("PEXPIRE", KEYS[1], ARGV[2])
|
||||
else
|
||||
return 0
|
||||
end
|
||||
`)
|
||||
|
||||
result, err := script.Run(ctx, l.client, []string{l.key}, l.owner, l.ttl.Milliseconds()).Int64()
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return result == 1, nil
|
||||
}
|
||||
|
||||
// IsHeld 检查锁是否被持有（不一定是自己）
|
||||
func (l *DistributedLock) IsHeld(ctx context.Context) (bool, error) {
|
||||
val, err := l.client.Exists(ctx, l.key).Result()
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return val > 0, nil
|
||||
}
|
||||
@@ -0,0 +1,172 @@
|
||||
package lock
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/joho/godotenv"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
func setupClient(t *testing.T) *redis.Client {
|
||||
t.Helper()
|
||||
|
||||
// 尝试加载 .env（兼容从项目根目录或子目录运行）
|
||||
_ = godotenv.Load("../../.env")
|
||||
_ = godotenv.Load("../.env")
|
||||
_ = godotenv.Load(".env")
|
||||
|
||||
url := os.Getenv("REDIS_URL")
|
||||
if url == "" {
|
||||
t.Skip("REDIS_URL not set, skipping integration test")
|
||||
}
|
||||
|
||||
opt, err := redis.ParseURL(url)
|
||||
if err != nil {
|
||||
t.Fatalf("Invalid REDIS_URL: %v", err)
|
||||
}
|
||||
|
||||
client := redis.NewClient(opt)
|
||||
if err := client.Ping(context.Background()).Err(); err != nil {
|
||||
t.Fatalf("Redis connection failed: %v", err)
|
||||
}
|
||||
|
||||
t.Cleanup(func() { client.Close() })
|
||||
return client
|
||||
}
|
||||
|
||||
func TestAcquireAndRelease(t *testing.T) {
|
||||
client := setupClient(t)
|
||||
ctx := context.Background()
|
||||
|
||||
l := New(client, "test-lock-acquire", "owner-1", 10*time.Second)
|
||||
|
||||
// 清理可能残留的 key
|
||||
_ = l.Release(ctx)
|
||||
|
||||
// 第一次获取应该成功
|
||||
ok, err := l.Acquire(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("Acquire error: %v", err)
|
||||
}
|
||||
if !ok {
|
||||
t.Fatal("Expected to acquire lock")
|
||||
}
|
||||
|
||||
// 释放
|
||||
if err := l.Release(ctx); err != nil {
|
||||
t.Fatalf("Release error: %v", err)
|
||||
}
|
||||
|
||||
// 释放后再次获取应该成功
|
||||
ok, err = l.Acquire(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("Acquire after release error: %v", err)
|
||||
}
|
||||
if !ok {
|
||||
t.Fatal("Expected to acquire lock after release")
|
||||
}
|
||||
|
||||
_ = l.Release(ctx)
|
||||
}
|
||||
|
||||
func TestAcquireConflict(t *testing.T) {
|
||||
client := setupClient(t)
|
||||
ctx := context.Background()
|
||||
|
||||
l1 := New(client, "test-lock-conflict", "owner-1", 10*time.Second)
|
||||
l2 := New(client, "test-lock-conflict", "owner-2", 10*time.Second)
|
||||
|
||||
_ = l1.Release(ctx)
|
||||
_ = l2.Release(ctx)
|
||||
|
||||
// owner-1 获取锁
|
||||
ok, err := l1.Acquire(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("Acquire error: %v", err)
|
||||
}
|
||||
if !ok {
|
||||
t.Fatal("Expected owner-1 to acquire lock")
|
||||
}
|
||||
|
||||
// owner-2 获取同一把锁应该失败
|
||||
ok, err = l2.Acquire(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("Acquire error: %v", err)
|
||||
}
|
||||
if ok {
|
||||
t.Fatal("Expected owner-2 to fail acquiring lock")
|
||||
}
|
||||
|
||||
// owner-2 释放别人的锁应该失败
|
||||
err = l2.Release(ctx)
|
||||
if err != ErrLockNotHeld {
|
||||
t.Fatalf("Expected ErrLockNotHeld, got: %v", err)
|
||||
}
|
||||
|
||||
// owner-1 正常释放
|
||||
if err := l1.Release(ctx); err != nil {
|
||||
t.Fatalf("Release error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRefresh(t *testing.T) {
|
||||
client := setupClient(t)
|
||||
ctx := context.Background()
|
||||
|
||||
l := New(client, "test-lock-refresh", "owner-1", 5*time.Second)
|
||||
_ = l.Release(ctx)
|
||||
|
||||
ok, err := l.Acquire(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("Acquire error: %v", err)
|
||||
}
|
||||
if !ok {
|
||||
t.Fatal("Expected to acquire lock")
|
||||
}
|
||||
|
||||
// 续期
|
||||
ok, err = l.Refresh(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("Refresh error: %v", err)
|
||||
}
|
||||
if !ok {
|
||||
t.Fatal("Expected refresh to succeed")
|
||||
}
|
||||
|
||||
_ = l.Release(ctx)
|
||||
}
|
||||
|
||||
func TestAcquireWithRetry(t *testing.T) {
|
||||
client := setupClient(t)
|
||||
ctx := context.Background()
|
||||
|
||||
l1 := New(client, "test-lock-retry", "owner-1", 10*time.Second)
|
||||
l2 := New(client, "test-lock-retry", "owner-2", 10*time.Second)
|
||||
|
||||
_ = l1.Release(ctx)
|
||||
|
||||
// owner-1 先获取锁
|
||||
ok, _ := l1.Acquire(ctx)
|
||||
if !ok {
|
||||
t.Fatal("Expected owner-1 to acquire lock")
|
||||
}
|
||||
|
||||
// owner-2 带重试，100ms 后 owner-1 释放，owner-2 应该能拿到
|
||||
go func() {
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
_ = l1.Release(ctx)
|
||||
}()
|
||||
|
||||
ok, err := l2.AcquireWithRetry(ctx, 50*time.Millisecond, 10)
|
||||
if err != nil {
|
||||
t.Fatalf("AcquireWithRetry error: %v", err)
|
||||
}
|
||||
if !ok {
|
||||
t.Fatal("Expected owner-2 to acquire lock after retry")
|
||||
}
|
||||
|
||||
_ = l2.Release(ctx)
|
||||
}
|
||||
Reference in New Issue
Block a user