網(wǎng)站首頁 編程語言 正文
1. go實現(xiàn)分布式鎖
通過 golang 實現(xiàn)一個簡單的分布式鎖，包括鎖續(xù)約、重試機制、singleflght機制的使用
1.1 redis_lock.go
package redis_lock
import (
"context"
_ "embed"
"errors"
"github.com/go-redis/redis/v9"
"github.com/google/uuid"
"golang.org/x/sync/singleflight"
"time"
)
// go:embed 可以直接解析出文件中的字符串
var (
//go:embed lua_unlock.lua
luaUnlock string
//go:embed refresh.lua
luaRefresh string
//go:embed lock.lua
luaLock string
//定義好兩個異常信息
ErrLockNotHold = errors.New("未持有鎖")
ErrFailedToPreemptLock = errors.New("加鎖失敗")
)
type Client struct {
//采用公共的接口，后續(xù)實例通過傳入的方式
client redis.Cmdable
// singleflight 用于在一個實例的多個攜程中只需要競爭出一個攜程
s singleflight.Group
}
func NewClient(c redis.Cmdable) *Client {
return &Client{
client: c,
}
}
func (c *Client) SingleflightLock(ctx context.Context,
key string,
expire time.Duration,
retry RetryStrategy,
timeout time.Duration) (*Lock, error) {
for {
flag := false
resCh := c.s.DoChan(key, func() (interface{}, error) {
flag = true
return c.Lock(ctx, key, expire, retry, timeout)
})
select {
case res := <-resCh:
if flag {
if res.Err != nil {
return nil, res.Err
}
//返回鎖對象
return res.Val.(*Lock), nil
}
case <-ctx.Done():
return nil, ctx.Err()
}
}
}
//Lock 加鎖方法，根據(jù)重試機制進行重新獲取
func (c *Client) Lock(ctx context.Context,
key string,
expire time.Duration,
retry RetryStrategy,
timeout time.Duration) (*Lock, error) {
var timer *time.Timer
defer func() {
if timer != nil {
timer.Stop()
}
}()
for {
//設(shè)置超時
lct, cancel := context.WithTimeout(ctx, timeout)
//獲取到uuid
value := uuid.New().String()
//執(zhí)行l(wèi)ua腳本進行加鎖
result, err := c.client.Eval(lct, luaLock, []string{key}, value, expire).Bool()
//用于主動釋放資源
cancel()
if err != nil && !errors.Is(err, context.DeadlineExceeded) {
return nil, err
}
if result {
return newLock(c.client, key, value), nil
}
//可以不傳重試機制
if retry != nil {
//通過重試機制獲取重試的策略
interval, ok := retry.Next()
if !ok {
//不用重試
return nil, ErrFailedToPreemptLock
}
if timer == nil {
timer = time.NewTimer(interval)
}
timer.Reset(interval)
select {
case <-timer.C: //睡眠時間超時了
return nil, ctx.Err()
case <-ctx.Done(): //整個調(diào)用的超時
return nil, ctx.Err()
}
}
}
}
// TryLock 嘗試加鎖
func (c *Client) TryLock(ctx context.Context, key string, expire time.Duration) (*Lock, error) {
return c.Lock(ctx, key, expire, nil, 0)
}
// NewLock 創(chuàng)建一個鎖結(jié)構(gòu)體
func newLock(client redis.Cmdable, key string, value string) *Lock {
return &Lock{
client: client,
key: key,
value: value,
unLockChan: make(chan struct{}, 1), //設(shè)置1個緩存數(shù)據(jù)，用于解鎖的信號量
}
}
// Lock 結(jié)構(gòu)體對象
type Lock struct {
client redis.Cmdable
key string
value string
expire time.Duration
//在解鎖成功之后發(fā)送信號來取消續(xù)約
unLockChan chan struct{}
}
// AutoRefresh 自動續(xù)約
func (l *Lock) AutoRefresh(interval time.Duration, timeout time.Duration) error {
//設(shè)計一個管道，如果失敗了，就發(fā)送數(shù)據(jù)到管道之中，通知進行重試
retry := make(chan struct{}, 1)
//方法返回時關(guān)閉close
defer close(retry)
ticker := time.NewTicker(interval)
for {
select {
//接收到結(jié)束的信號時，直接return
case <-l.unLockChan:
return nil
//監(jiān)聽重試的管道
case <-retry:
ctx, cancel := context.WithTimeout(context.Background(), timeout)
err := l.Refresh(ctx)
//主動調(diào)用釋放資源
cancel()
if err == context.DeadlineExceeded {
// 執(zhí)行重試往管道中發(fā)送一個信號
retry <- struct{}{}
continue
}
if err != nil {
return err
}
case <-ticker.C:
ctx, cancel := context.WithTimeout(context.Background(), timeout)
err := l.Refresh(ctx)
//主動調(diào)用釋放資源
cancel()
if err == context.DeadlineExceeded {
// 執(zhí)行重試往管道中發(fā)送一個信號
retry <- struct{}{}
continue
}
if err != nil {
return err
}
}
}
}
// Refresh 續(xù)約
func (l *Lock) Refresh(ctx context.Context) error {
//執(zhí)行l(wèi)ua腳本，對鎖進行續(xù)約
i, err := l.client.Eval(ctx, luaRefresh, []string{l.key}, l.value, l.expire.Milliseconds()).Int64()
if err == redis.Nil {
return ErrLockNotHold
}
if err != nil {
return err
}
if i == 0 {
return ErrLockNotHold
}
return nil
}
// Unlock 解鎖
func (l *Lock) Unlock(ctx context.Context) error {
//解鎖時，退出方法需要發(fā)送一個信號讓自動續(xù)約的goroutine停止
defer func() {
l.unLockChan <- struct{}{}
close(l.unLockChan)
}()
//判斷返回的結(jié)果
result, err := l.client.Eval(ctx, luaUnlock, []string{l.key}, l.value).Int64()
if err == redis.Nil {
return ErrLockNotHold
}
if err != nil {
return err
}
//lua腳本返回的結(jié)果如果為0，也是代表當前鎖不是自己的
if result == 0 {
return ErrLockNotHold
}
return nil
}1.2 retry.go
package redis_lock
import "time"
// RetryStrategy 重試策略
type RetryStrategy interface {
// Next 下一次重試的時間是多久，返回兩個參數(shù) time 時間，bool 是否直接重試
Next() (time.Duration, bool)
}
1.3 lock.lua
lua腳本原子化加鎖
--[[ 獲取到對應(yīng)的value是否跟當前的一樣 ]]
if redis.call("get", KEYS[1]) == ARGV[1]
then
-- 如果一樣直接對其時間進行續(xù)約
return redis.call("pexpire", KEYS[1], ARGV[2])
else
-- 如果不一樣調(diào)用setnx命令對其進行設(shè)置值
return redis.call("set", KEYS[1], ARGV[1], "NX", "PX", ARGV[2])
1.4 lua_unlock.lua
lua腳本原子化解鎖
if redis.call("get", KEYS[1]) == ARGV[1] then
-- 返回0，代表key不在
return redis.call("del", KEYS[1])
else
-- key在，但是值不對
return 0
end
1.5 refresh.lua
lua腳本續(xù)約
if redis.call("get", KEYS[1]) == ARGV[1] then
-- 返回0，代表key不在
return redis.call("pexpire", KEYS[1], ARGV[2])
else
-- key在，但是值不對
return 0
end
1.6 單元測試
使用go-mock工具生成本地的單元測試，不需要再單獨的搭建一個 redis 的服務(wù)端
項目根目錄下安裝mockgen工具
go install github.com/golang/mock/mockgen@latest
添加依賴
go get github.com/golang/mock/mockgen/model
生成redis客戶端接口
mockgen -package=mocks -destination=mocks/redis_cmdable.mock.go github.com/go-redis/redis/v9 Cmdable
- package：指定包
- destination：生成路徑名稱
- 剩下的是指定使用redis包下面的 Cmdable接口生成代碼
測試類
func TestClient_TryLock(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
testCase := []struct {
//測試的場景
name string
//輸入
key string
expiration time.Duration
//返回一個mock數(shù)據(jù)
mock func() redis.Cmdable
//期望的返回的錯誤值
wantError error
//期望返回的鎖
wantLock *Lock
}{
{
name: "locked",
key: "locked-key",
expiration: time.Minute,
mock: func() redis.Cmdable {
rdb := mocks.NewMockCmdable(ctrl)
res := redis.NewBoolResult(true, nil)
i := []interface{}{gomock.Any(), time.Minute}
rdb.EXPECT().Eval(gomock.Any(), luaLock, []string{"locked-key"}, i...).Return(res)
return rdb
},
wantLock: &Lock{
key: "locked-key",
},
},
}
for _, tc := range testCase {
t.Run(tc.name, func(t *testing.T) {
var c = NewClient(tc.mock())
l, err := c.TryLock(context.Background(), tc.key, tc.expiration)
assert.Equal(t, tc.wantError, err)
if err != nil {
return
}
//判斷返回的key是否跟期望的一樣
assert.Equal(t, tc.key, l.key)
assert.Equal(t, tc.wantLock.key, l.key)
assert.NotEmpty(t, l.value)
})
}
}
原文鏈接：https://blog.csdn.net/weixin_43915643/article/details/128232322
相關(guān)推薦
- 2022-11-11 C++?左值引用與一級指針示例詳解_C 語言
- 2023-06-04 pycharm中下載的包但是import還是無法使用/報紅的解決方法_python
- 2022-10-29 umi pro-layout : 某個頁面 禁用/移除 pro-layout ( 比如: 登錄頁不需
- 2022-11-15 Python正則表達式re.search()用法詳解_python
- 2022-07-07 圖解AVL樹數(shù)據(jù)結(jié)構(gòu)輸入與輸出及實現(xiàn)示例_C 語言
- 2022-09-23 python?pandas創(chuàng)建多層索引MultiIndex的6種方式_python
- 2023-04-06 Python求字符串的長度示例代碼_python
- 2022-04-20 appium中常見的幾種點擊方式_python
- 最近更新
-
- window11 系統(tǒng)安裝 yarn
- 超詳細win安裝深度學習環(huán)境2025年最新版（
- Linux 中運行的top命令 怎么退出？
- MySQL 中decimal 的用法？ 存儲小
- get 、set 、toString 方法的使
- @Resource和 @Autowired注解
- Java基礎(chǔ)操作-- 運算符，流程控制 Flo
- 1. Int 和Integer 的區(qū)別，Jav
- spring @retryable不生效的一種
- Spring Security之認證信息的處理
- Spring Security之認證過濾器
- Spring Security概述快速入門
- Spring Security之配置體系
- 【SpringBoot】SpringCache
- Spring Security之基于方法配置權(quán)
- redisson分布式鎖中waittime的設(shè)
- maven:解決release錯誤：Artif
- restTemplate使用總結(jié)
- Spring Security之安全異常處理
- MybatisPlus優(yōu)雅實現(xiàn)加密？
- Spring ioc容器與Bean的生命周期。
- 【探索SpringCloud】服務(wù)發(fā)現(xiàn)-Nac
- Spring Security之基于HttpR
- Redis 底層數(shù)據(jù)結(jié)構(gòu)-簡單動態(tài)字符串（SD
- arthas操作spring被代理目標對象命令
- Spring中的單例模式應(yīng)用詳解
- 聊聊消息隊列，發(fā)送消息的4種方式
- bootspring第三方資源配置管理
- GIT同步修改后的遠程分支
提供CDN加速