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

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

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

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

如下圖所示

這種情況有沒有問題呢？

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

因為在分布式場景下，redis 是多個服務(wù)共享的，如果多個服務(wù)同時判斷 key 不存在，那么就會同時設(shè)置 key，就會導(dǎo)致多個服務(wù)同時執(zhí)行，這不是我們想要的結(jié)果

為什么這樣做會有問題？

因為 get 和 set 操作不是原子操作，你先要做操作 get，然后在操作 set，這個過程中

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

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

這個怎么做呢？

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

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

如下圖所示：

當(dāng)我們了解了原理之后，看下人家是不是這樣實現(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)建了一個 mutex，然后調(diào)用 mutex.Lock() 方法

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

比較重要的有：

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

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 是核心方法，里面會做很多事情，這一步我們關(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 方法

通過過期時間防止死鎖

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

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

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

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

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

這個過期時間是怎么來的呢？

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

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

我們很快就會想到，在過期前刷新下過期時間不就行了？

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

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
`)

它為什么要這樣做呢？

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

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

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

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

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

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

我們在使用 m.Extend() 續(xù)期時，需要用協(xié)程去做

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

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

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

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

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

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

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

刪除的時候，調(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
`)

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

redlock

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

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

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

一般 redis 集群有一個 master 節(jié)點，多個 slave 節(jié)點

如下圖所示：

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

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

如下圖所示：

面對這種問題如何解決，引入了 redlock 的這個概念

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

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

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

如下圖所示：

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

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

為什么要使用異步寫入多臺 redis 呢？

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

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

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

func (m *Mutex) actOnPoolsAsync(actFn func(redis.Pool) (bool, error)) (int, error) {
  type result struct {
    Node   int
    Status bool  // 成功寫入的臺數(shù)
    Err    error // 未成功寫入的錯誤
  }
  // 啟用 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++；寫入失敗，記錄錯誤
    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})
    }
  }
  // 將寫入的臺數(shù)和錯誤返回出去
  if len(taken) >= m.quorum {
    return n, &ErrTaken{Nodes: taken}
  }
  return n, err
}

總結(jié)

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

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

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

最新評論