【Go】缓存与分布式锁

缓存

本地缓存

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
package cache

import (
	"context"
	"errors"
	"time"
)

var (
	errKeyNotFound      = errors.New("cache: key 不存在")
	errOverCapacity     = errors.New("cache: 超过缓存最大容量")
	errFailedToSetCache = errors.New("cache: 设置键值对失败")
	errCacheClosed      = errors.New("cache: 已经被关闭")
	errKeyExpired       = errors.New("cache: 缓存失效")
)

type Cache interface {
	Get(ctx context.Context, key string) (any, error)
	Set(ctx context.Context, key string, val any, expiration time.Duration) error
	Delete(ctx context.Context, key string) error
	LoadAndDelete(ctx context.Context, key string) (any, error)
  Close(ctx context.Context) error
}

type item struct {
	val      any
	deadline time.Time
}

func (i *item) deadlineBefore(t time.Time) bool {
	return !i.deadline.IsZero() && i.deadline.Before(t)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
package cache

import (
	"context"
	"sync"
	"time"
)

type LocalCache struct {
	lock          sync.RWMutex
	data          map[string]*item
	close         chan struct{}
	closed        bool
	onEvicted     OnEvict
	cycleInterval time.Duration
}

type ParamCacheOption func(*LocalCache)
type OnEvict func(key string, val any)

func WithCycleInterval(interval time.Duration) ParamCacheOption {
	return func(cache *LocalCache) {
		cache.cycleInterval = interval
	}
}

func WithOnEvict(onEvict OnEvict) ParamCacheOption {
	return func(cache *LocalCache) {
		cache.onEvicted = onEvict
	}
}

func NewLocalCache(opts ...ParamCacheOption) Cache {
	lc := &LocalCache{
		lock:          sync.RWMutex{},
		data:          make(map[string]*item),
		close:         make(chan struct{}), // 初始化,否则调用 close 会阻塞掉
		cycleInterval: 10 * time.Second,
	}

	for _, opt := range opts {
		opt(lc)
	}

	lc.checkCycle()

	return lc
}

func (c *LocalCache) checkCycle() {
	go func() {
		ticker := time.NewTicker(c.cycleInterval)
		for {
			select {
			case now := <-ticker.C:
				for key, val := range c.data {
					func() {
						c.lock.Lock()
						defer c.lock.Unlock()
						if !val.deadline.IsZero() && val.deadline.Before(now) {
							c.delete(key)
						}
					}()
				}
			case <-c.close:
				close(c.close)
				return
			}
		}
	}()
}

func (c *LocalCache) delete(key string) {
	val, ok := c.data[key]
	if ok {
		delete(c.data, key)
		if c.onEvicted != nil {
			c.onEvicted(key, val.val)
		}

	}
}

func (c *LocalCache) Get(ctx context.Context, key string) (any, error) {
	c.lock.RLock()
	if c.closed {
		return nil, errCacheClosed
	}

	val, ok := c.data[key]
	c.lock.RUnlock()
	if !ok {
		return nil, errKeyNotFound
	}

	now := time.Now()
	// 别的 goroutine 设置值了
	if val.deadlineBefore(now) {
		c.lock.Lock()
		defer c.lock.Unlock()

		if c.closed {
			return nil, errCacheClosed
		}

		val, ok = c.data[key]
		if !ok {
			return nil, errKeyNotFound
		}

		if val.deadlineBefore(now) {
			c.delete(key)
			return nil, errKeyExpired
		}

	}

	return val.val, nil
}

func (c *LocalCache) Set(ctx context.Context, key string, val any, expiration time.Duration) error {
	c.lock.Lock()
	defer c.lock.Unlock()

	if c.closed {
		return errCacheClosed
	}

	var deadline time.Time
	if expiration > 0 {
		deadline = time.Now().Add(expiration)
	}

	c.data[key] = &item{
		val:      val,
		deadline: deadline,
	}

	return nil
}

func (c *LocalCache) Delete(ctx context.Context, key string) error {
	c.lock.Lock()
	defer c.lock.Unlock()
	if c.closed {
		return errCacheClosed
	}

	c.delete(key)

	return nil
}

func (c *LocalCache) Close(ctx context.Context) error {
	c.lock.Lock()
	defer c.lock.Unlock()
  
  if c.closed {
		return errCacheClosed
	}

	c.closed = true
	c.close <- struct{}{}
	if c.onEvicted != nil {
		for key, val := range c.data {
			c.onEvicted(key, val.val)
		}
	}

	c.data = nil

	return nil
}

func (c *LocalCache) LoadAndDelete(ctx context.Context, key string) (any, error) {
	c.lock.Lock()
	defer c.lock.Unlock()

	if c.closed {
		return nil, errCacheClosed
	}

	val, ok := c.data[key]
	if !ok {
		return nil, errKeyNotFound
	}

	c.delete(key)

	return val.val, nil
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
package cache

import (
	"context"
	"reflect"
	"testing"
	"time"
)

func TestLocalCache_Get(t *testing.T) {
	type args struct {
		ctx        context.Context
		key        string
		val        string
		sleepTime  time.Duration
		expiration time.Duration
		closed     bool
	}
	tests := []struct {
		name    string
		args    args
		want    any
		wantErr ErrCache
	}{
		{
			name: "Get Cache not expired",
			args: args{
				ctx:        context.Background(),
				key:        "key1",
				val:        "val1",
				sleepTime:  time.Millisecond * 10,
				expiration: time.Second * 10,
			},
			want:    "val1",
			wantErr: nil,
		},
		{
			name: "Get Cache expired",
			args: args{
				ctx:        context.Background(),
				key:        "key1",
				val:        "val1",
				sleepTime:  time.Second * 2,
				expiration: time.Second * 1,
			},
			want:    nil,
			wantErr: errKeyNotFound,
		},
		{
			name: "Get Cache expired",
			args: args{
				ctx:        context.Background(),
				key:        "key1",
				val:        "val1",
				sleepTime:  time.Second * 2,
				expiration: time.Second * 1,
				closed:     true,
			},
			want:    nil,
			wantErr: errCacheClosed,
		},
	}

	c := NewCache(WithCycleInterval(time.Second))
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			c.Set(tt.args.ctx, tt.args.key, tt.args.val, tt.args.expiration)
			time.Sleep(tt.args.sleepTime)
			if tt.args.closed {
				c.Close(tt.args.ctx)
			}
			got, err := c.Get(tt.args.ctx, tt.args.key)
			if err != tt.wantErr {
				t.Errorf("Get() error = %v, wantErr %v", err, tt.wantErr)
				return
			}
			if !reflect.DeepEqual(got, tt.want) {
				t.Errorf("Get() got = %v, want %v", got, tt.want)
			}
		})
	}
}

func TestLocalCache_Set(t *testing.T) {
	type args struct {
		ctx        context.Context
		key        string
		val        string
		expiration time.Duration
		closed     bool
	}
	tests := []struct {
		name    string
		args    args
		want    any
		wantErr ErrCache
	}{
		{
			name: "Set Cache",
			args: args{
				ctx:        context.Background(),
				key:        "key1",
				val:        "val1",
				expiration: time.Second * 10,
			},
			want:    "val1",
			wantErr: nil,
		},
		{
			name: "Set Cache closed",
			args: args{
				ctx:        context.Background(),
				key:        "key1",
				val:        "val1",
				closed:     true,
				expiration: time.Second * 10,
			},
			want:    "val1",
			wantErr: errCacheClosed,
		},
	}
	c := NewCache(WithCycleInterval(time.Second))
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if tt.args.closed {
				c.Close(tt.args.ctx)
			}
			err := c.Set(tt.args.ctx, tt.args.key, tt.args.val, tt.args.expiration)
			if err != tt.wantErr {
				t.Errorf("Get() error = %v, wantErr %v", err, tt.wantErr)
				return
			}
		})
	}
}

func TestLocalCache_Delete(t *testing.T) {
	type args struct {
		ctx        context.Context
		key        string
		val        string
		expiration time.Duration
		closed     bool
		unset      bool
	}
	tests := []struct {
		name    string
		args    args
		want    any
		wantErr ErrCache
	}{
		{
			name: "Delete Cache",
			args: args{
				ctx:        context.Background(),
				key:        "key1",
				val:        "val1",
				expiration: time.Second * 10,
			},
			wantErr: nil,
		},
		{
			name: "Delete Cache not exist",
			args: args{
				ctx:        context.Background(),
				key:        "key1",
				val:        "val1",
				unset:      true,
				expiration: time.Second * 10,
			},
			wantErr: nil,
		},
		{
			name: "Delete Cache closed",
			args: args{
				ctx:        context.Background(),
				key:        "key1",
				val:        "val1",
				closed:     true,
				expiration: time.Second * 10,
			},
			wantErr: errCacheClosed,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			c := NewCache(WithCycleInterval(time.Second))
			if !tt.args.unset {
				c.Set(tt.args.ctx, tt.args.key, tt.args.val, 0)
			}
			if tt.args.closed {
				c.Close(tt.args.ctx)
			}

			err := c.Delete(tt.args.ctx, tt.args.key)
			if err != tt.wantErr {
				t.Errorf("Delete() error = %v, wantErr %v", err, tt.wantErr)
				return
			}
		})
	}
}

func TestLocalCache_LoadAndDelete(t *testing.T) {
	type args struct {
		ctx        context.Context
		key        string
		val        string
		expiration time.Duration
		closed     bool
		unset      bool
	}
	tests := []struct {
		name    string
		args    args
		want    any
		wantErr ErrCache
	}{
		{
			name: "LoadAndDelete Cache",
			args: args{
				ctx:        context.Background(),
				key:        "key1",
				val:        "val1",
				expiration: time.Second * 10,
			},
			want:    "val1",
			wantErr: nil,
		},
		{
			name: "LoadAndDelete Cache not exist",
			args: args{
				ctx:        context.Background(),
				key:        "key1",
				val:        "val1",
				unset:      true,
				expiration: time.Second * 10,
			},
			want:    nil,
			wantErr: errKeyNotFound,
		},
		{
			name: "Delete Cache closed",
			args: args{
				ctx:        context.Background(),
				key:        "key1",
				val:        "val1",
				closed:     true,
				expiration: time.Second * 10,
			},
			want:    nil,
			wantErr: errCacheClosed,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			c := NewCache(WithCycleInterval(time.Second))
			if !tt.args.unset {
				c.Set(tt.args.ctx, tt.args.key, tt.args.val, 0)
			}
			if tt.args.closed {
				c.Close(tt.args.ctx)
			}

			got, err := c.LoadAndDelete(tt.args.ctx, tt.args.key)
			if err != tt.wantErr {
				t.Errorf("Delete() error = %v, wantErr %v", err, tt.wantErr)
				return
			}

			if !reflect.DeepEqual(got, tt.want) {
				t.Errorf("Get() got = %v, want %v", got, tt.want)
			}
		})
	}
}

func TestLocalCache_Close(t *testing.T) {
	tests := []struct {
		name    string
		key     string
		wantVal string
	}{
		{
			name: "Test close",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			c := NewCache()
			err := c.Close(context.Background())
			if err != nil {
				t.Errorf("Close() error = %v", err)
			}
		})
	}
}

控制缓存内存

限制缓存key数量
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
package cache

import (
	"context"
	"sync/atomic"
	"time"
)

type MaxCntCache struct {
	*LocalCache
  mutex  sync.Mutex
	cnt    int32
	maxCnt int32
}

func NewMaxCntCache(c *LocalCache, maxCnt int32) Cache {
	mc := &MaxCntCache{
		LocalCache: c,
		maxCnt:     maxCnt,
	}

	origin := c.onEvicted
	c.onEvicted = func(key string, val any) {
		atomic.AddInt32(&mc.cnt, -1)
		if origin != nil {
			origin(key, val)
		}
	}

	return mc
}

func (c *MaxCntCache) Set(ctx context.Context, key string, val any, expiration time.Duration) error {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	_, err := c.LocalCache.Get(ctx, key)
	if err != nil && err != errKeyNotFound {
		return err
	}

	if err == errKeyNotFound {
		cnt := atomic.AddInt32(&c.cnt, 1)
		if cnt > c.maxCnt {
			atomic.AddInt32(&c.cnt, -1)
			return errOverCapacity
		}
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
package cache

import (
	"context"
	"testing"
	"time"
)

func TestMaxCntCache_Set(t *testing.T) {
	type args struct {
		ctx        context.Context
		key        string
		val        string
		expiration time.Duration
		closed     bool
		maxCnt     int32
	}
	tests := []struct {
		name    string
		args    args
		wantErr ErrCache
	}{
		{
			name: "Set Cache",
			args: args{
				ctx:        context.Background(),
				key:        "key1",
				val:        "val1",
				expiration: time.Second * 10,
				maxCnt:     100,
			},
			wantErr: nil,
		},
		{
			name: "Set Cache closed",
			args: args{
				ctx:        context.Background(),
				key:        "key1",
				val:        "val1",
				closed:     true,
				expiration: time.Second * 10,
				maxCnt:     100,
			},
			wantErr: errCacheClosed,
		},
		{
			name: "Set Cache over capacity",
			args: args{
				ctx:        context.Background(),
				key:        "key1",
				val:        "val1",
				closed:     true,
				expiration: time.Second * 10,
				maxCnt:     0,
			},
			wantErr: errOverCapacity,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			cc := NewLocalCache(WithCycleInterval(time.Second))
			c := NewMaxCntCache(cc, tt.args.maxCnt)
			if tt.args.closed {
				c.Close(tt.args.ctx)
			}
			err := c.Set(tt.args.ctx, tt.args.key, tt.args.val, tt.args.expiration)
			if err != tt.wantErr {
				t.Errorf("Get() error = %v, wantErr %v", err, tt.wantErr)
				return
			}
		})
	}
}
限制内存
缓存淘汰

缓存模式

cache-aside

  • 把cache当作普通数据源

  • 更新cache和DB都依赖开发者自己写代码

    • 业务代码决策是否从DB取数据
    • 同步或者异步读取并写入数据
    • 采用singleflight

同步读取数据到缓存

异步存数据到缓存

异步读取数据到缓存

如果并发情况下，可能会引起数据不一致问题

read-through
  • 业务代码从缓存中读取数据，cache不命中是读取数据
    • cache决策是否从DB读取数据
    • 同步或者异步读取并写入数据
    • 采用singleflight
  • 写数据时，业务代码需要自己写DB和cache

同步读取数据到缓存

异步存数据到缓存

异步读取数据到缓存（一般不会用）
write-through
  • 业务代码数据只写cache，cache自己更新DB
    • cache决定同步或者异步写到DB或者cache
    • cache决定先写DB还是cache,一般先写DB
  • 业务代码读取未命中，需要手动从DB获取并写入缓存

同步写数据

异步写数据到缓存

异步写数据
write-back
  • 写操作时直接写缓存，读操作时也是直接读缓存
  • 缓存过期时，将缓存数据写入DB（onEvicted回调，刷新数据到DB）
  • 数据可能会丢失（缓存过期刷数据到DB之前,缓存宕机）

refresh-ahead
  • 依赖cdc
  • cache或者cannel这一类的工具监听数据变更后，更新数据到缓存
  • 读数据cache未命中时，还是需要刷新缓存，也存在并发问题

Redis缓存

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
package cache

import (
	"context"
	"strings"
	"time"

	_ "github.com/golang/mock/mockgen/model"
	"github.com/pkg/errors"
	"github.com/redis/go-redis/v9"
)

type RedisCache struct {
	client redis.Cmdable
}

func NewRedisClient(addr string) (client redis.Cmdable) {
	return redis.NewClient(&redis.Options{
		Addr:     addr,
		Password: "",
		DB:       0,
	})
}

func NewRedisCache(client redis.Cmdable) *RedisCache {
	return &RedisCache{
		client: client,
	}
}

func (c *RedisCache) Get(ctx context.Context, key string) (any, error) {

	return c.client.Get(ctx, key).Result()
}

func (c *RedisCache) Set(ctx context.Context, key string, val any, expiration time.Duration) error {
	msg, err := c.client.Set(ctx, key, val, expiration).Result()
	if err != nil {
		return err
	}

	if strings.ToLower(msg) != "ok" {
		return errors.Wrapf(err, "返回信息 %s", msg)
	}

	return nil
}

func (c *RedisCache) Delete(ctx context.Context, key string) error {
	_, err := c.client.Del(ctx, key).Result()
	return err
}

func (c *RedisCache) LoadAndDelete(ctx context.Context, key string) (any, error) {
	return c.client.GetDel(ctx, key).Result()
}

func (c *RedisCache) Close(ctx context.Context) error {
	return nil
}

生成redis_mock.go文件

1
2
go install github.com/golang/mock/mockgen@v1.6.0
mockgen -package=mocks -destination=mocks/redis_cmdable.mock.go github.com/redis/go-redis/v9 Cmdable
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
package cache

import (
	"context"
	"reflect"
	"testing"
	"time"

	"github.com/golang/mock/gomock"
	"github.com/redis/go-redis/v9"
	"github.com/stretchr/testify/assert"

	"offer/pkg/cache/mocks"
)

func TestRedisCache_Set(t *testing.T) {
	controller := gomock.NewController(t)
	defer controller.Finish()
	testCases := []struct {
		name       string
		mock       func() redis.Cmdable
		key        string
		val        string
		expiration time.Duration
		wantErr    error
	}{
		{
			name: "OK",
			mock: func() redis.Cmdable {
				cmdable := mocks.NewMockCmdable(controller)
				cmd := redis.NewStatusCmd(context.Background())
				cmd.SetVal("OK")
				cmdable.EXPECT().Set(gomock.Any(), "key1", "val1", 30*time.Second).Return(cmd)
				return cmdable
			},
			key:        "key1",
			val:        "val1",
			expiration: 30 * time.Second,
			wantErr:    nil,
		},
	}

	for _, c := range testCases {
		t.Run(c.name, func(t *testing.T) {
			rc := NewRedisCache(c.mock())
			err := rc.Set(context.Background(), c.key, c.val, c.expiration)
			assert.Equal(t, c.wantErr, err)
		})
	}
}

func TestRedisCache_Get(t *testing.T) {
	controller := gomock.NewController(t)
	defer controller.Finish()
	testCases := []struct {
		name    string
		mock    func() redis.Cmdable
		key     string
		want    any
		wantErr error
	}{
		{
			name: "OK",
			mock: func() redis.Cmdable {
				cmdable := mocks.NewMockCmdable(controller)
				cmd := redis.NewStringCmd(context.Background())
				cmd.SetVal("val1")
				cmdable.EXPECT().Get(gomock.Any(), "key1").Return(cmd)
				return cmdable
			},
			key:     "key1",
			want:    "val1",
			wantErr: nil,
		},
	}
	for _, c := range testCases {
		t.Run(c.name, func(t *testing.T) {
			ctx := context.Background()
			rc := NewRedisCache(c.mock())
			got, err := rc.Get(ctx, c.key)
			if c.wantErr != err {
				t.Errorf("Get() error = %v, wantErr %v", err, c.wantErr)
				return
			}
			if !reflect.DeepEqual(got, c.want) {
				t.Errorf("Get() got = %v, want %v", got, c.want)
			}
		})
	}
}

缓存异常

穿透

cache没有，DB也没有

击穿

并发请求同一个缓存数据，cache没有，DB压力大

雪崩

cache出现了错误，不能正常工作了，所有的请求都会达到DB

分布式锁

几个Redis分布式锁库

redsync

redsync获取锁时使用命令SETNX，释放锁通过lua脚本实现

go-zero

Go-zero实现的只是简单的加锁和释放锁，并且这两个操作均是基于lua脚本实现。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
var (
	lockScript = NewScript(`if redis.call("GET", KEYS[1]) == ARGV[1] then
    redis.call("SET", KEYS[1], ARGV[1], "PX", ARGV[2])
    return "OK"
else
    return redis.call("SET", KEYS[1], ARGV[1], "NX", "PX", ARGV[2])
end`)
	delScript = NewScript(`if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("DEL", KEYS[1])
else
    return 0
end`)
)

// A RedisLock is a redis lock.
type RedisLock struct {
	store   *Redis
	seconds uint32
	key     string
	id      string
}

// NewRedisLock returns a RedisLock.
func NewRedisLock(store *Redis, key string) *RedisLock {
	return &RedisLock{
		store: store,
		key:   key,
		id:    stringx.Randn(randomLen),
	}
}

// AcquireCtx acquires the lock with the given ctx.
func (rl *RedisLock) AcquireCtx(ctx context.Context) (bool, error) {
	seconds := atomic.LoadUint32(&rl.seconds)
	resp, err := rl.store.ScriptRunCtx(ctx, lockScript, []string{rl.key}, []string{
		rl.id, strconv.Itoa(int(seconds)*millisPerSecond + tolerance),
	})
	if err == red.Nil {
		return false, nil
	} else if err != nil {
		logx.Errorf("Error on acquiring lock for %s, %s", rl.key, err.Error())
		return false, err
	} else if resp == nil {
		return false, nil
	}

	reply, ok := resp.(string)
	if ok && reply == "OK" {
		return true, nil
	}

	logx.Errorf("Unknown reply when acquiring lock for %s: %v", rl.key, resp)
	return false, nil
}

// ReleaseCtx releases the lock with the given ctx.
func (rl *RedisLock) ReleaseCtx(ctx context.Context) (bool, error) {
	resp, err := rl.store.ScriptRunCtx(ctx, delScript, []string{rl.key}, []string{rl.id})
	if err != nil {
		return false, err
	}

	reply, ok := resp.(int64)
	if !ok {
		return false, nil
	}

	return reply == 1, nil
}

// ScriptRunCtx is the implementation of *redis.Script run command.
func (s *Redis) ScriptRunCtx(ctx context.Context, script *Script, keys []string, args ...any) (val any, err error) {
	err = s.brk.DoWithAcceptable(func() error {
		conn, err := getRedis(s)
		if err != nil {
			return err
		}

		val, err = script.Run(ctx, conn, keys, args...).Result()
		return err
	}, acceptable)
	return
}

Redis分布式锁

Redis分布式自旋锁

Golang
#Go #Cache #Lock
【Go】缓存与分布式锁
https://weitrue.github.io/2024/03/21/golang-cache/
作者
Pony W
发布于
2024年3月21日
许可协议