Skip to content

Commit fadb855

Browse files
keepchenkeepchen
committed
utils:redis锁优化增强
1 parent 12b51ea commit fadb855

File tree

2 files changed

+131
-30
lines changed

2 files changed

+131
-30
lines changed

‎utils/redislock.go‎

Lines changed: 85 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"context"
55
"fmt"
66
"os"
7+
"strings"
78
"sync"
89
"time"
910

@@ -62,8 +63,8 @@ type IRedisLocker interface {
6263
//
6364
// # Note
6465
//
65-
// 该方法会阻塞住线程直到解锁成功 或者 触发ctx.Done()
66-
UnlockWithContext(ctx context.Context, key string)
66+
// 该方法会阻塞住线程直到解锁有结果 或者 触发ctx.Done()
67+
UnlockWithContext(ctx context.Context, key string) bool
6768
}
6869

6970
var _ IRedisLocker = &redisLockerImpl{}
@@ -104,7 +105,7 @@ type cancelControl struct {
104105
}
105106

106107
type stateListeners struct {
107-
mux *sync.Mutex
108+
mux *sync.RWMutex
108109
listeners map[string]*cancelControl
109110
}
110111

@@ -113,7 +114,7 @@ var (
113114
redisExecuteTimeout = time.Second * 3
114115
retryInterval = time.Millisecond * 100
115116
renewalCheckInterval = time.Second * 1
116-
states = &stateListeners{mux: &sync.Mutex{}, listeners: make(map[string]*cancelControl)}
117+
states = &stateListeners{mux: &sync.RWMutex{}, listeners: make(map[string]*cancelControl)}
117118
)
118119

119120
// TryLock redis锁-尝试上锁
@@ -142,6 +143,10 @@ func (rl *redisLockerImpl) TryLock(key string) bool {
142143
//
143144
// 该方法会立即返回锁定成功与否的结果
144145
func (rl *redisLockerImpl) TryLockWithContext(ctx context.Context, key string) bool {
146+
if !canDoLockPreflight(key) {
147+
return false
148+
}
149+
145150
lockOk, lockErr := rl.client.SetNX(ctx, key, lockerValue(), lockTTL).Result()
146151
if lockErr != nil {
147152
fmt.Printf("[Go-Sail] <redisLock> key: %s lock err: %v\n", key, lockErr)
@@ -174,6 +179,7 @@ func (rl *redisLockerImpl) Lock(ctx context.Context, key string) {
174179
//第一次锁定失败，进行重试操作
175180
if !lockOk || lockErr != nil {
176181
retryTicker := time.NewTicker(retryInterval)
182+
defer retryTicker.Stop()
177183

178184
LOOP:
179185
for {
@@ -202,9 +208,19 @@ func (rl *redisLockerImpl) Unlock(key string) bool {
202208
ctx, cancel := withRedisExecuteTimeout()
203209
defer cancel()
204210

211+
//持有者一致性检测(如果获取失败，也认为不符合一致性)
212+
lv, err := rl.client.Get(ctx, key).Result()
213+
if err != nil {
214+
fmt.Printf("[Go-Sail] <redisLock> key: %s unlock get key err: %v\n", key, err)
215+
return false
216+
}
217+
if !holderConsistencyDetection(lv) {
218+
return false
219+
}
220+
205221
unlockOk, unlockErr := rl.client.Del(ctx, key).Result()
206222
if unlockErr != nil {
207-
fmt.Printf("[Go-Sail] <redisLock> key: %s unlock error: %v\n", key, unlockErr)
223+
fmt.Printf("[Go-Sail] <redisLock> key: %s unlock delete key error: %v\n", key, unlockErr)
208224
}
209225

210226
//清理内存数据并终止自动续期
@@ -218,11 +234,21 @@ func (rl *redisLockerImpl) Unlock(key string) bool {
218234
// UnlockWithContext redis锁-解锁
219235
//
220236
// using Del
221-
func (rl *redisLockerImpl) UnlockWithContext(ctx context.Context, key string) {
237+
func (rl *redisLockerImpl) UnlockWithContext(ctx context.Context, key string) bool {
238+
//持有者一致性检测(如果获取失败，也认为不符合一致性)
239+
lv, err := rl.client.Get(ctx, key).Result()
240+
if err != nil {
241+
fmt.Printf("[Go-Sail] <redisLock> key: %s unlock with context get key err: %v\n", key, err)
242+
return false
243+
}
244+
if !holderConsistencyDetection(lv) {
245+
return false
246+
}
222247
unlockOk, unlockErr := rl.client.Del(ctx, key).Result()
223248

224249
if unlockOk != 1 || unlockErr != nil {
225250
ticker := time.NewTicker(retryInterval)
251+
defer ticker.Stop()
226252

227253
LOOP:
228254
for {
@@ -240,6 +266,8 @@ func (rl *redisLockerImpl) UnlockWithContext(ctx context.Context, key string) {
240266

241267
//清理内存数据并终止自动续期
242268
rl.clearListenerAndStopAutoRenewal(key)
269+
270+
return unlockOk == 1
243271
}
244272

245273
// 自动续期
@@ -274,26 +302,35 @@ type keyAndCtrl struct {
274302
// 2.使用redis pipeline减少RTT
275303
func (rl *redisLockerImpl) startRenewalScheduler() {
276304
doRenewalRound := func() {
305+
if rl.client == nil {
306+
fmt.Println("[Go-Sail] <redisLock> renewal task not emit cause redis client is nil")
307+
return
308+
}
309+
277310
states.mux.Lock()
278311
if len(states.listeners) == 0 {
279312
states.mux.Unlock()
280313
//避免空转锁定占用
281314
return
282315
}
283-
keys := make([]*keyAndCtrl, 0, len(states.listeners))
316+
processingKeys := make([]*keyAndCtrl, 0, len(states.listeners))
284317
for key, ctrl := range states.listeners {
285-
keys = append(keys, &keyAndCtrl{key: key, ctrl: ctrl})
318+
processingKeys = append(processingKeys, &keyAndCtrl{key: key, ctrl: ctrl})
286319
}
287320
states.mux.Unlock()
288321

289322
ctx, cancel := withRedisExecuteTimeout()
290323
defer cancel()
324+
325+
validKeys := make([]*keyAndCtrl, 0, len(processingKeys))
326+
invalidKeys := make([]*keyAndCtrl, 0, len(processingKeys))
291327
cmds, pipeErr := rl.client.Pipelined(ctx, func(pipe redisLib.Pipeliner) error {
292-
for index := range keys {
293-
if keys[index].ctrl.ctx.Err() != nil {
294-
delete(states.listeners, keys[index].key)
328+
for index := range processingKeys {
329+
if processingKeys[index].ctrl.ctx.Err() != nil {
330+
invalidKeys = append(invalidKeys, processingKeys[index])
295331
} else {
296-
pipe.ExpireXX(ctx, keys[index].key, lockTTL)
332+
pipe.Expire(ctx, processingKeys[index].key, lockTTL)
333+
validKeys = append(validKeys, processingKeys[index])
297334
}
298335
}
299336
return nil
@@ -308,12 +345,17 @@ func (rl *redisLockerImpl) startRenewalScheduler() {
308345
for index := range cmds {
309346
if expOk, expErr := cmds[index].(*redisLib.BoolCmd).Result(); !expOk || expErr != nil {
310347
if expErr != nil {
311-
fmt.Printf("[Go-Sail] <redisLock> key: %s renewal err: %v\n", keys[index].key, expErr)
348+
fmt.Printf("[Go-Sail] <redisLock> key: %s renewal err: %v\n", validKeys[index].key, expErr)
312349
}
313-
keys[index].ctrl.cancel() //续期失败也要清理掉
314-
delete(states.listeners, keys[index].key)
350+
validKeys[index].ctrl.cancel() //续期失败也要清理掉
351+
delete(states.listeners, validKeys[index].key)
315352
}
316353
}
354+
//清理已经过期的
355+
for index := range invalidKeys {
356+
invalidKeys[index].ctrl.cancel() //保险的再次调用以确保触发ctx.Done
357+
delete(states.listeners, invalidKeys[index].key)
358+
}
317359
states.mux.Unlock()
318360
}
319361

@@ -327,6 +369,16 @@ func (rl *redisLockerImpl) startRenewalScheduler() {
327369
}()
328370
}
329371

372+
// 预检是否可以执行锁定任务
373+
//
374+
// 此操作属于本地(堆栈)检测
375+
func canDoLockPreflight(key string) bool {
376+
states.mux.RLock()
377+
defer states.mux.RUnlock()
378+
_, exist := states.listeners[key]
379+
return !exist
380+
}
381+
330382
// redis操作超时控制
331383
func withRedisExecuteTimeout() (context.Context, context.CancelFunc) {
332384
return context.WithTimeout(context.Background(), redisExecuteTimeout)
@@ -335,9 +387,26 @@ func withRedisExecuteTimeout() (context.Context, context.CancelFunc) {
335387
var (
336388
hostname, _ = os.Hostname() //主机名称
337389
ip, _ = IP().GetLocal() //主机ip
390+
processId = os.Getpid() //进程id
338391
)
339392

340393
// 锁的持有者信息
341394
func lockerValue() string {
342-
return fmt.Sprintf("lockedAt:%s@%s(%s)", hostname, ip, time.Now().Format("2006-01-02T15:04:05.000000Z"))
395+
return fmt.Sprintf("lockedAt:%s@%s<%d>(%s)",
396+
hostname, ip, processId, time.Now().Format("2006-01-02T15:04:05.000000Z"))
397+
}
398+
399+
// 持有者一致性检测
400+
//
401+
// # 注意：
402+
//
403+
// 一致性检测以【机器主机名+ip+进程id】为判断依据，
404+
//
405+
// 这样设计为的是锁只能被【持有者自己】释放，若进程
406+
//
407+
// down掉，堆栈中的自动维护信息会被释放，
408+
//
409+
// 因此即便是重新启动获取到了相同的进程号，也不受影响。
410+
func holderConsistencyDetection(lockerValue string) bool {
411+
return strings.HasPrefix(lockerValue, fmt.Sprintf("lockedAt:%s@%s<%d>(", hostname, ip, processId))
343412
}

‎utils/redislock_test.go‎

Lines changed: 46 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@ package utils
22

33
import (
44
"context"
5+
"fmt"
6+
"sync"
57
"testing"
68
"time"
79

@@ -16,6 +18,35 @@ func TestRedisLockerImplLockerValue(t *testing.T) {
1618
}
1719
}
1820

21+
func TestHolderConsistencyDetection(t *testing.T) {
22+
t.Run("holderConsistencyDetection", func(t *testing.T) {
23+
realLv := fmt.Sprintf("lockedAt:%s@%s<%d>(", hostname, ip, processId)
24+
assert.Equal(t, true, holderConsistencyDetection(realLv))
25+
fakeLv := fmt.Sprintf("lockedAt:%s@%s<%d>(", hostname, ip, 0)
26+
assert.Equal(t, false, holderConsistencyDetection(fakeLv))
27+
})
28+
}
29+
30+
func TestCanDoLockPreflight(t *testing.T) {
31+
t.Run("canDoLockPreflight", func(t *testing.T) {
32+
clearState()
33+
34+
var testKey = "canDoLockPreflight-testKey"
35+
assert.Equal(t, true, canDoLockPreflight(testKey))
36+
37+
states.mux.Lock()
38+
states.listeners[testKey] = &cancelControl{}
39+
states.mux.Unlock()
40+
41+
assert.Equal(t, false, canDoLockPreflight(testKey))
42+
43+
//clear
44+
states.mux.Lock()
45+
delete(states.listeners, testKey)
46+
states.mux.Unlock()
47+
})
48+
}
49+
1950
func TestRedisLockPanic(t *testing.T) {
2051
t.Run("RedisLockPanic-Init-Panic", func(t *testing.T) {
2152
if redis.GetInstance() == nil && redis.GetClusterInstance() == nil {
@@ -85,12 +116,12 @@ func TestRedisLock(t *testing.T) {
85116
redis.InitRedis(conf)
86117

87118
t.Run("TryLock", func(t *testing.T) {
88-
key := "go-sail-redisLocker-TryLock"
119+
key := "go-sail-redisLocker-TryLock-01"
89120
t.Log(RedisLocker().TryLock(key))
90121
assert.Equal(t, false, RedisLocker().TryLock(key))
91122
time.Sleep(time.Second * 15)
92123
assert.Equal(t, false, RedisLocker().TryLock(key))
93-
RedisLocker().Unlock(key)
124+
t.Log(RedisLocker().Unlock(key))
94125
assert.Equal(t, true, RedisLocker().TryLock(key))
95126
})
96127

@@ -104,8 +135,7 @@ func TestRedisLock(t *testing.T) {
104135
t.Run("Unlock", func(t *testing.T) {
105136
key := "go-sail-redisLocker-Unlock"
106137
t.Log(RedisLocker().TryLock(key))
107-
RedisLocker().Unlock(key)
108-
assert.Equal(t, true, RedisLocker().TryLock(key))
138+
assert.Equal(t, true, RedisLocker().Unlock(key))
109139
})
110140

111141
t.Run("Unlock-WithContext", func(t *testing.T) {
@@ -150,31 +180,29 @@ func TestRedisClusterLock(t *testing.T) {
150180

151181
t.Run("TryLock", func(t *testing.T) {
152182
key := "go-sail-redisLocker-TryLock"
153-
t.Log(RedisLocker().TryLock(key))
154-
assert.Equal(t, false, RedisLocker().TryLock(key))
183+
t.Log(RedisLocker(redis.GetClusterInstance()).TryLock(key))
184+
assert.Equal(t, false, RedisLocker(redis.GetClusterInstance()).TryLock(key))
155185
})
156186

157187
t.Run("Lock", func(t *testing.T) {
158188
key := "go-sail-redisLocker-Lock"
159189
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
160190
defer cancel()
161-
RedisLocker().Lock(ctx, key)
191+
RedisLocker(redis.GetClusterInstance()).Lock(ctx, key)
162192
})
163193

164194
t.Run("Unlock", func(t *testing.T) {
165-
key := "go-sail-redisLocker-Unlock"
166-
t.Log(RedisLocker().TryLock(key))
167-
RedisLocker().Unlock(key)
168-
assert.Equal(t, true, RedisLocker().TryLock(key))
195+
key := "go-sail-redisLocker-cluster-Unlock"
196+
t.Log(RedisLocker(redis.GetClusterInstance()).TryLock(key))
197+
assert.Equal(t, true, RedisLocker(redis.GetClusterInstance()).Unlock(key))
169198
})
170199

171200
t.Run("Unlock-WithContext", func(t *testing.T) {
172201
key := "go-sail-redisLocker-UnlockWithContext"
173-
t.Log(RedisLocker().TryLock(key))
202+
t.Log(RedisLocker(redis.GetClusterInstance()).TryLock(key))
174203
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
175204
defer cancel()
176-
RedisLocker().UnlockWithContext(ctx, key)
177-
assert.Equal(t, true, RedisLocker().TryLock(key))
205+
assert.Equal(t, true, RedisLocker(redis.GetClusterInstance()).UnlockWithContext(ctx, key))
178206
})
179207
}
180208

@@ -204,3 +232,7 @@ func TestStartRenewalScheduler(t *testing.T) {
204232
_ = redisClient.Close()
205233
})
206234
}
235+
236+
func clearState() {
237+
states = &stateListeners{mux: &sync.RWMutex{}, listeners: make(map[string]*cancelControl)}
238+
}

0 commit comments

Comments
 (0)