通過原子操作實(shí)現(xiàn) redis 鎖

redis 內(nèi)部是通過 key/value 的形式存儲(chǔ)的，核心原理是設(shè)置一個(gè)唯一的 key，如果這個(gè) key 存在，說明有服務(wù)在使用

具體實(shí)現(xiàn)方式：

• 首先判斷 redis 中是否存在某個(gè) key，并且為某個(gè)值
• 如果這個(gè) key 不存在，說明當(dāng)前沒有服務(wù)在使用，設(shè)置 key
• 如果這個(gè) key 存在，說明當(dāng)前有服務(wù)在使用，就等待一段時(shí)間，然后再次判斷這個(gè) key 是否存在

如下圖所示

這種情況有沒有問題呢？

如果在單體應(yīng)用的場(chǎng)景下，這種方式是可行的；但是在分布式場(chǎng)景下，這種方式就不可行了

因?yàn)樵诜植际綀?chǎng)景下，redis 是多個(gè)服務(wù)共享的，如果多個(gè)服務(wù)同時(shí)判斷 key 不存在，那么就會(huì)同時(shí)設(shè)置 key，就會(huì)導(dǎo)致多個(gè)服務(wù)同時(shí)執(zhí)行，這不是我們想要的結(jié)果

為什么這樣做會(huì)有問題？

因?yàn)?get 和 set 操作不是原子操作，你先要做操作 get，然后在操作 set，這個(gè)過程中

這就會(huì)導(dǎo)致當(dāng)?shù)谝慌_(tái)服務(wù)在執(zhí)行 get 時(shí)，發(fā)現(xiàn) key 不存在，然后進(jìn)行 set，這個(gè)時(shí)候 set 可能還沒有完成，第二臺(tái)服務(wù)執(zhí)行了 get，發(fā)現(xiàn) key 不存在，然后進(jìn)行 set，這個(gè)時(shí)候就會(huì)導(dǎo)致多個(gè)服務(wù)同時(shí)執(zhí)行，這就不是原子操作了

原子操作的意思是：一次性執(zhí)行，不會(huì)被打斷

這個(gè)怎么做呢？

redis 提供了一個(gè) setnx 的方法，作用是如果 key 不存在，就設(shè)置 key，設(shè)置成功返回 1，設(shè)置失敗返回 0

這就將 get 和 set 的邏輯合二為一了，保證原子性了

如下圖所示：

當(dāng)我們了解了原理之后，看下人家是不是這樣實(shí)現(xiàn)的，以 redsync 為例，先來看它使用，從入口函數(shù)一步步往下追

rs := redsync.New(pool)
mutexname := "my-global-mutex"
mutex := rs.NewMutex(mutexname)
if err := mutex.Lock(); err != nil {
  panic(err)
}
if ok, err := mutex.Unlock(); !ok || err != nil {
  panic("unlock failed")
}

從上面代碼可以看到，它先調(diào)用 NewMutex 創(chuàng)建了一個(gè) mutex，然后調(diào)用 mutex.Lock() 方法

NewMutex 是初始化函數(shù)，用來初始化一系列的參數(shù)，

比較重要的有：

• name：redis 中的 key
• genValueFunc：生成 key 的函數(shù)，保證唯一性
• expiry：key 過期的時(shí)間
• tries：嘗試的次數(shù)，可能會(huì)拿不到鎖，所以要嘗試多次
• delayFunc：延遲時(shí)間(睡眠時(shí)間)，可能會(huì)拿不到鎖，就需要等一會(huì)再嘗試
• quorum：大多數(shù)節(jié)點(diǎn)，這個(gè)是用來做分布式鎖的，如果有 5 個(gè)節(jié)點(diǎn)，那么這里的大多數(shù)是 3 個(gè)節(jié)點(diǎn)

m := &Mutex{
  name:   name,
  expiry: 8 * time.Second,
  tries:  32,
  delayFunc: func(tries int) time.Duration {
    return time.Duration(rand.Intn(maxRetryDelayMilliSec-minRetryDelayMilliSec)+minRetryDelayMilliSec) * time.Millisecond
  },
  genValueFunc:  genValue,
  driftFactor:   0.01,
  timeoutFactor: 0.05,
  quorum:        len(r.pools)/2 + 1,
  pools:         r.pools,
}

初始化結(jié)束之后，調(diào)用 m.Lock() 上鎖，m.Lock() 方法中調(diào)用 m.LockContext() 方法，

LockContext 是核心方法，里面會(huì)做很多事情，這一步我們關(guān)心它是怎么上鎖的，通過搜索發(fā)現(xiàn)，上鎖的方法是 m.acquire()，其源碼是：

func (m *Mutex) acquire(ctx context.Context, pool redis.Pool, value string) (bool, error) {
  conn, err := pool.Get(ctx)
  if err != nil {
    return false, err
  }
  defer conn.Close()
  reply, err := conn.SetNX(m.name, value, m.expiry)
  if err != nil {
    return false, err
  }
  return reply, nil
}

在這里我們清晰的看到調(diào)用 SetNX 方法

通過過期時(shí)間防止死鎖

這樣做完之后，還有一個(gè)問題需要解決

如果正在操作 redis 的服務(wù)掛了，那么這個(gè) key 就會(huì)一直存在，其他服務(wù)就會(huì)等待，這樣就造成了死鎖

解決這個(gè)問題就是設(shè)置過期時(shí)間，如果服務(wù)掛了，過期時(shí)間到了，key 就會(huì)自動(dòng)刪除，其他服務(wù)就可以繼續(xù)使用了

通過源代碼我們可以看到它設(shè)置了一個(gè)過期時(shí)間 expiry

reply, err := conn.SetNX(m.name, value, m.expiry)

這個(gè)過期時(shí)間是怎么來的呢？

剛剛在入口函數(shù)中，我們看到了 NewMutex 函數(shù)，它初始化了一個(gè) expiry，這個(gè) expiry 就是過期時(shí)間：expiry: 8 * time.Second，它默認(rèn)設(shè)置的是 8 秒

到這里就有疑問了，如果我的服務(wù)執(zhí)行時(shí)間超過 8 秒怎么辦？，不就達(dá)不到鎖的效果了？

我們很快就會(huì)想到，在過期前刷新下過期時(shí)間不就行了？

確實(shí) redsync 也考慮到了這個(gè)問題，它提供了一個(gè) Extend 方法，用來刷新過期時(shí)間

m.Extent() 方法調(diào)用 m.ExtendContext() 方法，在 m.ExtendContext() 方法中調(diào)用 m.touch() 方法

func (m *Mutex) Extend() (bool, error) {
  return m.ExtendContext(nil)
}
func (m *Mutex) ExtendContext(ctx context.Context) (bool, error) {
  // ... 省略其他代碼
  m.touch(ctx, pool, m.value, int(m.expiry/time.Millisecond))
  // ... 省略其他代碼
}
func (m *Mutex) touch(ctx context.Context, pool redis.Pool, value string, expiry int) (bool, error) {
  // ... 省略其他代碼
	conn, err := pool.Get(ctx)
  conn.Eval(touchScript, m.name, value, expiry)
}

在 m.touch() 方法中我們看到它調(diào)用 redis 提供的 Eval 方法，可以執(zhí)行一段 lua 腳本，腳本的內(nèi)容如下：

var touchScript = redis.NewScript(1, `
  if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("PEXPIRE", KEYS[1], ARGV[2])
  else
    return 0
  end
`)

它為什么要這樣做呢？

不就是把過期時(shí)間刷新下嗎？為什么要寫 lua 呢

這里我們需要了解下 redis 的 lua 腳本，redis 的 lua 腳本是原子性的，它可以保證一段腳本的執(zhí)行是原子性的

這樣就可以保證刷新過期時(shí)間的操作是原子性的，不會(huì)出現(xiàn)刷新過期時(shí)間失敗的情況

如果我們用 go 語言去續(xù)期的需要三步：

• 先獲取到 key 的值
• 判斷 redis 中的值是不是你傳進(jìn)來的值
• 如果是的話，續(xù)期

這樣的話，這樣的話就不具備原子性了，任何一步都有失敗的可能，所以 redsync 選擇了 lua 腳本

我們?cè)谑褂?m.Extend() 續(xù)期時(shí)，需要用協(xié)程去做

那 redsync 為什么不自動(dòng)續(xù)期呢？

如果做自動(dòng)續(xù)期的話，當(dāng)前正在操作的服務(wù)如果 hung 住了，那么就會(huì)不停的續(xù)期，造成其他服務(wù)無法進(jìn)來，所以 redsync 將續(xù)期的功能交給了使用者

防止被其他服務(wù)刪除

鎖只能被持有該鎖的服務(wù)刪除，不能被其他服務(wù)刪除

如果保證鎖只能被持有該鎖的服務(wù)刪除，那么就需要在 setnx 的時(shí)候，給 key 設(shè)置一個(gè)唯一的值，這個(gè)值可以是 uuid，這樣就可以保證鎖只能被持有該鎖的服務(wù)刪除

我們看下 redsync 源碼是如何做的，初始化時(shí)就生成了一個(gè)唯一的值，它是使用 base64 編碼的

func genValue() (string, error) {
  b := make([]byte, 16)
  _, err := rand.Read(b)
  if err != nil {
    return "", err
  }
  return base64.StdEncoding.EncodeToString(b), nil
}

刪除的時(shí)候，調(diào)用 m.Unlock() 方法，m.Unlock() 方法調(diào)用 m.UnlockContext() 方法，在在 m.release() 方法

func (m *Mutex) release(ctx context.Context, pool redis.Pool, value string) (bool, error) {
  // ... 省略其他代碼
  conn, err := pool.Get(ctx)
  conn.Eval(deleteScript, m.name, value)
}

在 m.release() 方法中我們看的也是在執(zhí)行 lua 腳本，腳本的內(nèi)容如下：

var deleteScript = redis.NewScript(1, `
  if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("DEL", KEYS[1])
  else
    return 0
  end
`)

這也是為了保證在刪除鎖的時(shí)候，保證原子性

redlock

通過我們上面講解的已經(jīng)能滿足一般的使用場(chǎng)景，但是在大型項(xiàng)目中，不會(huì)只搭建一個(gè) redis，而是搭建 redis 集群

這樣又會(huì)出現(xiàn)一個(gè)新的問題：redlock

redlock 是什么呢？我們先來看下 redis 集群

一般 redis 集群有一個(gè) master 節(jié)點(diǎn)，多個(gè) slave 節(jié)點(diǎn)

如下圖所示：

當(dāng)我在加鎖時(shí)，如果 master 節(jié)點(diǎn)會(huì)自動(dòng)同步到 slave 節(jié)點(diǎn)，那么就不會(huì)有問題

如果這時(shí) master 節(jié)點(diǎn)出問題了（或者說在同步過程中出問題，還沒有同步完），slave 節(jié)點(diǎn)會(huì)選舉出一個(gè) master 節(jié)點(diǎn)，這個(gè)過程中會(huì)有一段時(shí)間，這時(shí)如果有一個(gè)服務(wù)進(jìn)來寫，發(fā)現(xiàn)是能寫入的，這就出現(xiàn)了問題

如下圖所示：

面對(duì)這種問題如何解決，引入了 redlock 的這個(gè)概念

redlock 的核心思想是：在 redis 集群中，大多數(shù)節(jié)點(diǎn)都能寫入成功，那么就認(rèn)為寫入成功，而不是只向一臺(tái) redis 寫入

當(dāng)?shù)谝粋€(gè)服務(wù)寫入時(shí)，同時(shí)向 5 臺(tái) redis 寫入，這時(shí)如果第二個(gè)服務(wù)寫入，寫同時(shí)向 5 臺(tái) redis 寫入，誰先成功寫入大多數(shù) redis，誰就認(rèn)為寫入成功，鎖就交給誰

這里的大多數(shù)就是比一半多 1 臺(tái)，也就是 n / 2 + 1，所以 redis 應(yīng)該準(zhǔn)備奇數(shù)臺(tái)，同時(shí)也無需關(guān)心這 5 臺(tái) redis 的主從關(guān)系了

如下圖所示：

我們通過 redsync 源碼來學(xué)習(xí) redlock，是如何實(shí)現(xiàn)的：

• 通過 select 實(shí)現(xiàn)超時(shí)控制
• 核心代碼是 actOnPoolsAsync 方法
  • pools：表示向多臺(tái) redis 寫入
  • async：表示異步寫入多臺(tái) redis，同步寫入的話，效率偏低，使用 goroutine（具體可以查看下面 actOnPoolsAsync 方法的分析）
• 判斷是否拿到鎖
  • 如果拿到鎖，更新 m.value 和 m.until
  • 如果沒有拿到鎖，需要釋放已經(jīng)寫入的 redis 的 key

func (m *Mutex) LockContext(ctx context.Context) error {
  if ctx == nil {
    ctx = context.Background()
  }
  value, err := m.genValueFunc()
  if err != nil {
    return err
  }
  // 如果沒有拿到鎖，等待一段時(shí)間在去拿
  for i := 0; i < m.tries; i++ {
    if i != 0 {
      // 使用 select 實(shí)現(xiàn)超時(shí)控制
      select {
      case <-ctx.Done():
        return ErrFailed
      case <-time.After(m.delayFunc(i)):
      }
    }
    // 記錄拿鎖開始時(shí)間
    start := time.Now()
    n, err := func() (int, error) {
      ctx, cancel := context.WithTimeout(ctx, time.Duration(int64(float64(m.expiry)*m.timeoutFactor)))
      defer cancel()
      // 異步寫入多臺(tái) redis
      return m.actOnPoolsAsync(func(pool redis.Pool) (bool, error) {
        return m.acquire(ctx, pool, value)
      })
    }()
    // 記錄拿鎖結(jié)束時(shí)間
    now := time.Now()
    // 計(jì)算還剩多少時(shí)間：過期時(shí)間 - 拿鎖花費(fèi)的時(shí)間 - 時(shí)間偏移
    // 這段代碼是為了防止 `redis` 節(jié)點(diǎn)時(shí)間不同步，導(dǎo)致鎖過期時(shí)間不準(zhǔn)確，所以在過期時(shí)間上加上一個(gè) `driftFactor`，這個(gè)值是 `0.01`，也就是 `1%` 的誤差
    until := now.Add(m.expiry - now.Sub(start) - time.Duration(int64(float64(m.expiry)*m.driftFactor)))
    // 判斷是否競(jìng)爭(zhēng)成功
    if n >= m.quorum && now.Before(until) {
      m.value = value
      m.until = until
      return nil
    }
    // 如果競(jìng)爭(zhēng)失敗，釋放已經(jīng)寫入的 redis 的 key
    func() (int, error) {
      ctx, cancel := context.WithTimeout(ctx, time.Duration(int64(float64(m.expiry)*m.timeoutFactor)))
      defer cancel()
      return m.actOnPoolsAsync(func(pool redis.Pool) (bool, error) {
        return m.release(ctx, pool, value)
      })
    }()
    if i == m.tries-1 && err != nil {
      return err
    }
  }
  return ErrFailed
}

為什么要使用異步寫入多臺(tái) redis 呢？

如果采用同步寫入的多臺(tái)的話，如果寫入的 redis 比較多，就會(huì)很耗時(shí)，可能寫到最后一臺(tái) redis 時(shí)，前面的 redis 已經(jīng)過期了，這樣就會(huì)出現(xiàn)問題

啟用 goroutine 去寫入的話，可以一瞬間都拿到 lock，調(diào)用 setnx 方法去寫入

然后再統(tǒng)計(jì)成功寫入的臺(tái)數(shù)，返回出去

func (m *Mutex) actOnPoolsAsync(actFn func(redis.Pool) (bool, error)) (int, error) {
  type result struct {
    Node   int
    Status bool  // 成功寫入的臺(tái)數(shù)
    Err    error // 未成功寫入的錯(cuò)誤
  }
  // 啟用 goroutine 去調(diào)用 setnx 寫入
  // 用 channel 來接收結(jié)果
  ch := make(chan result)
  for node, pool := range m.pools {
    go func(node int, pool redis.Pool) {
      r := result{Node: node}
      r.Status, r.Err = actFn(pool)
      ch <- r
    }(node, pool)
  }
  n := 0
  var taken []int
  var err error
  for range m.pools {
    r := <-ch
    // 寫入成功，n++；寫入失敗，記錄錯(cuò)誤
    if r.Status {
      n++
    } else if r.Err != nil {
      err = multierror.Append(err, &RedisError{Node: r.Node, Err: r.Err})
    } else {
      taken = append(taken, r.Node)
      err = multierror.Append(err, &ErrNodeTaken{Node: r.Node})
    }
  }
  // 將寫入的臺(tái)數(shù)和錯(cuò)誤返回出去
  if len(taken) >= m.quorum {
    return n, &ErrTaken{Nodes: taken}
  }
  return n, err
}

總結(jié)

分布式鎖的實(shí)現(xiàn)需要考慮的問題：

• 原子性(互斥性)：鎖只能被一個(gè)服務(wù)持有
  • 使用 setnx 命令，將 set 和 get 變成原子性
  • 使用 lua 腳本
• 死鎖：設(shè)置過期時(shí)間，防止服務(wù)掛了變成死鎖
  • 續(xù)期操作需要保證原子性，使用 lua 腳本
• 安全性：鎖只能被持有該鎖的服務(wù)刪除，不能被其他服務(wù)刪除
  • 在 setnx 的時(shí)候，給 key 設(shè)置一個(gè)唯一的值
• redlock：解決多臺(tái) redis 同步問題
  • 一個(gè)服務(wù)同時(shí)向多臺(tái) redis 設(shè)置 lock
  • 哪個(gè)服務(wù)向大多數(shù) redis 寫入成功，控制權(quán)就交給哪個(gè)服務(wù)

以上就是詳解Redis實(shí)現(xiàn)分布式鎖的原理的詳細(xì)內(nèi)容，更多關(guān)于Redis分布式鎖原理的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論