一篇和Redis有關的鎖和事務的文章

部分參考連接

Transaction

StackExchange.Redis Transaction

hashest

正文

Redis 是一種基於內存的單線程數據庫。意味着全部的命令是一個接一個的執行。

---

考慮只有一個Redis實例，也就是Redis自己沒有作分佈式。

---

經過SETNX命令，set if not exist的縮寫。那麼多個服務在調用的時候能夠經過同一個key申請一個lock（也就是調用命令成功返回1），而後根據相應條件作釋放(好比時間到期，or手動釋放)，也就是delete key。

Redis自己有MULTI命令，標記開啓一個事務。開啓以後後面的命令會在調用EXEC命令的時候以一個集合的方式總體執行，也就是原子性（不保證都成功）。

如今有個需求，用redis實現Check and Set，也就是先讀取裏面的值，而後設置(好比作個+=val)；併發的問題是必需要考慮的。

用redis描述大體是這樣的。這裏假設redis沒有incr這個自增命令。

val = GET mykey
val = val + 1
SET mykey $val

直接這樣作，併發問題是確定有的。因此，按照上面的知識，應該有2種方法來避免這個併發問題。

基於SENTX命令。

copy一下文檔的demo

redis> SETNX mykey "Hello"
(integer) 1
redis> SETNX mykey "World"
(integer) 0
redis> GET mykey
"Hello"
redis>

第一次調用setnx，設置mykey的value爲hello，返回1，表示成功。

第二次調用setnx，設置mykey的value爲world，由於第一次調用並無釋放mykey，因此返回0，表示設置失敗。

最後獲取mykey的值，返回的是hello。

最後記得要去釋放mykey。

這實際上是一個悲觀鎖，也就是一個進程獲取到鎖以後要等釋放別的進程才能繼續。

基於MULTI命令。

1. 先看一個簡單的應用

   127.0.0.1:6379> multi
   OK
   127.0.0.1:6379> incr foo
   QUEUED
   127.0.0.1:6379> incr bar
   QUEUED
   127.0.0.1:6379> exec
   1) (integer) 1
   2) (integer) 1

   第一步調用MULTI命令，表示開始多個命令的輸入。返回OK，表示開始接收。

   第二步調用incr foo，給foo對應的值作自增。返回queued，表示已加入隊列。

   第二步調用incr bar，給bar對應的值作資政，返回queued，表示已加入隊列。

   最後調用exec命令，表示執行隊列中的命令。返回每一個命令的結果。

2. 有錯誤了怎麼辦

   首先錯誤分兩種

   • 在enqueue的時候出錯，最多見的就是參數錯誤。好比下面這個例子

   127.0.0.1:6379> multi
   OK
   127.0.0.1:6379> set a 1234
   QUEUED
   127.0.0.1:6379> set a 1 1 1 1 1 1 11
   QUEUED
   127.0.0.1:6379> exec
   1) OK
   2) (error) ERR syntax error
   127.0.0.1:6379>

   第二個set a 1 1 1 1 1 1 11命令是有語法錯誤，因此，在執行exec的時候會返回語法錯誤。第一個是成功的。因此，若是在後面get a是會返回1234，爲成功的設置。

   假設報錯的命令在中間，後面的命令也是會執行的。

   • 還有就是直接命令就不對的。看個例子

   127.0.0.1:6379> multi
   OK
   127.0.0.1:6379> set a 11
   QUEUED
   127.0.0.1:6379> aaa
   (error) ERR unknown command `aaa`, with args beginning with:
   127.0.0.1:6379> exec
   (error) EXECABORT Transaction discarded because of previous errors.

   先set a，進入隊列。

   執行aaa命令，這個命令不存在。直接報錯。

   執行exec，事務由於以前的錯誤，exec停止。

3. 爲何沒有回滾

   經過上面的例子，看到redis對multi的操做是沒有回滾的，或許有點奇怪。根據文檔描述，有兩個緣由。

   • redis的命令執行只有在語法錯誤或者數據類型出錯的時候會失敗，而不是在enqueue的時候。這意味着失敗是由程序設置錯誤致使的。那麼，這種錯誤確定是在開發環境中就應該容易被發現，而不是在生產環境。
   • 爲了快。

4. WATCH 命令的樂觀鎖

   結合watch命令咱們也能夠實現上面的需求。

   WATCH mykey
   --Begin---
   ##下面兩行是客戶端命令
   val = GET mykey
   val = val + 1
   --End---
   MULTI
   SET mykey $val
   EXEC

   解釋一下，先獲取一下mykey的監控。而後客戶端獲取mykey的值，(是客戶端，不是命令服務端)。而後賦值自增。而後服務端開啓MULTI， 設置新的值。執行。

   假設在MULTI和Exec之間，mykey的值被別的client修改，exec會返回(nil)。

   下面作個演示：

   先在redis-cli上執行如下命令

   127.0.0.1:6379> watch a
   OK
   127.0.0.1:6379> multi
   OK
   127.0.0.1:6379> set a 13
   QUEUED

   如上，已經開啓WATCH，而後設置a =13 進入隊列。

   而後在本地的redis desktop manager上去修改這個值。

   而後再在服務器上執行 exec，

   127.0.0.1:6379> exec
   (nil)

   返回的是nil，表示沒有成功。若是沒有客戶端去更新，執行exec是返回OK。

5. redis-scripting-and-transactions

   在Redis 2.6以後，引入了Redis script來實現事務的功能。一般來講script方式速度會相對快一點(沒有作測試)。不過既然multi已經出來好久了，因此，不太可能會移除這個命令。

在StackExchange.Redis中使用

顯然，也分兩種，基於setnx 或者 MULTI + WATCH。分別對應的是IDatabaseAsync.LockTakeAsync和IDatabaseAsync.CreateTransaction這裏結合了Polly這個庫用於重試，畢竟，悲觀鎖，我多拿幾回總能拿到的；樂觀鎖，執行的命令，我多試幾回，總能成功的。

• LockTakeAsync

  public async Task<T> TakeLockAsync<T>(string key, string token, Func<object, Task<T>> func, object obj)
      where T : class
  {
      var db = GetDb(redisConfigModel.LockDbIndex);//獲取IDatabaseAsync對象
      //定義獲取鎖的策略
      var policy = Policy
          .HandleResult<bool>(w => !w)
          .WaitAndRetryForeverAsync(
              sleepDurationProvider: attemp => TimeSpan.FromSeconds(3), //兩次重複嘗試的間隔
              onRetry: (delegeteRst, ts) =>
              {
                  //能夠記錄日誌啥的
              }
          );
      //競爭獲取鎖。
      await policy.ExecuteAsync(async () => await db.LockTakeAsync(key, token, TimeSpan.MaxValue));  
      try
      {
          return await func(obj);//獲取到鎖以後的具體執行的方法。
      }
      finally
      {
          await db.LockReleaseAsync(key, token); //最後必定要釋放
      }
  }

  LockTakeAsync的時候根據key對應的token值是否已經被獲取來做爲條件。

• CreateTransaction

  StackExchange.Redis 用multiplexer類實現Redis的一些列命令。咱們的代碼不能直接簡單的映射到watch命令，由於，單純調用watch是確定成功的，這樣會致使你們都"成功"(假的)。這裏用的Condition的方式來實現。

  public async Task AddAfterReadAsync(string key, int value, string hashField = "hash_field")
  {
        //處理policy的結果爲false的狀況，一直重試。
      var policy = Policy.HandleResult<bool>(w => !w).RetryForeverAsync();
        //執行
      await policy.ExecuteAsync(async () =>
      {
          var db = GetDb(redisConfigModel.LockDbIndex);
          var trans = db.CreateTransaction();
          var oldValue = Convert.ToInt32(await db.StringGetAsync(key));
          trans.AddCondition(Condition.HashNotExists(key,
              hashField)); //這裏確保hashField不存在。也能夠用Condition.KeyNotExists(key)
          //這裏不能await，由於每一個命令的結果只有在執行了execute後才知道。
          trans.StringSetAsync(key, (oldValue + value).ToString());
          var execSuccess = await trans.ExecuteAsync();
          return execSuccess;
      });
  }

小結

這是一篇和redis有關的鎖，事務的文章。寫了我一整個下午。看完，感受也沒有多少東西。感受開頭連接中關於hashset仍是有點意思的。