緩存中間件-Redis(二)

在上一篇中我們簡單總結和介紹了Redis的幾個方面

1.使用Redis背景

2.Redis通信多路復用的基本原理

3.Redis基本數據結構

4.Redis持久化方式

這一篇我們使用簡單的業務場景來介紹Redis的分布式鎖和集群

1.分布式鎖

首先我們應該知道什么是分布式鎖，用來做什么的，解決了什么問題，我們應該怎么做？

簡單來說，分布式鎖就是鎖住進程的鎖，通常我們分布式鎖的使用場景在多實例的集群系統中，為了防止多個進程對資源的同時操作，而引發業務邏輯錯誤而誕生的，和我們的單個實例中的lock作用一樣，一個是針對多實例單進程，一個是針對單實例多線程，具體在web中的表現在下面用實例的方式說明。

2.單進程單實例多線程

1.例如現在我們部署了一個服務，他的功能是提供用戶來搶紅包的接口，我們在發紅包時，總共設置了10個紅包，我們將紅包剩余數存入數據庫，所以用一張表來存儲紅包的個數信息，每當有一個人來搶，我們就需要將紅包的個數減掉1，主要的核心業務邏輯如下：

• 1.用戶發起搶紅包請求，服務器首先查詢，紅包剩余個數，如果為0，則直接返回，告訴用戶紅包搶完了

• 2.如果紅包個數不為0，則提示用戶搶到了紅包，然后扣減紅包總數。

偽代碼如下

 //查詢剩余紅包個數
 int RedEnvelopeCount = getRedEnvelopes();
 if (RedEnvelopeCount.Count == 0)
 {
     //返回用戶紅包搶完了 do something
     return;
 }
 //通知用戶搶到了 do something
 //扣減紅包個數
 subtracRedEnvelopeCount();

2.此時如果同一時間多個並發請求進來，我們就會發生多個線程同時執行這段核心邏輯，導致業務錯誤

• 1.發生超搶,搶到的人大於實際個數，誰搶到的不算？

• 2.發生少搶，幾個人同時搶到同一個，算誰的？

我相信這2種情況，要是在真實的搶紅包中，發生到誰頭上，都不爽吧？按東哥的話說我都氣的想打人

所以我們要着手解決，解決的思路就是，核心業務的代碼同一時刻只能被一個線程執行，所以為了保證原子性，我們就需要使用到鎖來鎖住核心業務代碼，在這里我們使用C#的Lock來實現

//靜態全局唯一
private readonly static object _lock = new object();
using (_lock)
{
    int RedEnvelopeCount = getRedEnvelopes();
    if (RedEnvelopeCount.Count == 0)
    {
        //返回用戶紅包搶完了
        return;
    }
    //通知用戶搶到了
    //扣減紅包個數
    subtracRedEnvelopeCount();
}

3.單進程多實例多線程

上面我們列舉並介紹了在單個實例中遇到並發產生的業務問題，我們通過鎖的方式來暫時解決了問題，那么接下來我們繼續思考，如果紅包再多一點，搶的並發量起來了，單台服務器不能承載這么多並發，為了保證高可用，我們需要將紅包服務，進行集群部署，利用nginx來負載均衡一下。

1.在單個實例中我們為了保證業務執行的原子性，鎖住了核心業務代碼，保證同一時刻只有一個線程執行，但是現在多個實例來處理請求，假設現在2個用戶提交搶紅包的請求，在同一時刻被服務器負載均衡到2台不同的實例，那么此時就會遇到2個進程同時操作核心業務邏輯，超搶和少搶的事情又會發生，瞬間頭又大了，那么針對這種業務情況，我們應該使用分布式鎖來解決不同進程之間爭奪資源的情況。

2.接下來就是我們最后一個問題，如何實現分布式鎖，其實對於集群系統實現分布式鎖的方式有很多種，基於Zookeeper,Redis，包括數據庫，等都可以實現，相較於Zookeeper和數據庫 ,Redis單線程對於實現分布式鎖擁有天生絕對的優勢，而且它操作內存的方式就性能而言是非常優秀的，鑒於咱們這篇博客主要是跟Redis相關的，所以主要介紹使用Redis的方式，當然基於Redis中也存在不同的做法。

3.實現鎖的核心業務就是加鎖``釋放鎖這2個功能，我們根據Redis內部單線程執行的原理自己來初步實現一個

 //設置key，加鎖 防止死鎖超時，設置60毫秒自動過期
 bool result = database.StringSet("distributed-lock", "lock", expiry: TimeSpan.FromSeconds(60));
 //失敗說明鎖已經被獲取，直接返回
 if (!result)
 {
     // "人有點多，再點點...";
     return;
 }
 //獲取剩余個數
 int RedEnvelopeCount = getRedEnvelopes();
 if (RedEnvelopeCount.Count == 0)
 {
     //返回用戶紅包搶完了
     return;
 }
 //通知用戶搶到了
 //扣減紅包個數
 subtracRedEnvelopeCount();
 //釋放鎖
database.KeyDelete("distributed-lock");

上面實現了最簡單和最基礎的分布式鎖，但是還存在缺陷，仔細思考不難發現

• 1.執行業務的時間大於持有鎖的最大時間，導致鎖釋放，業務還沒執行完，亂了那也玩不了

• 2.獲取鎖后執行到業務代碼時拋異常了，鎖沒及時釋放，而是等待自動釋放，所以需要加異常處理

我們根據這2個缺陷來進一步完善,解決上面的問題，首先需要先保證執行業務的情況下，誰加的鎖由誰來釋放,然后添加異常處理代碼用於處理業務異常能及時釋放鎖

//獲取Guid作為當前請求唯一身份信息
string value = Guid.NewGuid().ToString("N");

//設置key，加鎖，防止死鎖超時，設置60毫秒自動過期
bool result = database.StringSet("distributed-lock", value, expiry: TimeSpan.FromSeconds(60));

//失敗說明鎖已經被獲取，直接返回
if (!result) return;  // "人有點多，再點點...";

try
{
    int RedEnvelopeCount = getRedEnvelopes(); //獲取剩余個數
    if (RedEnvelopeCount.Count == 0)return; //返回用戶紅包搶完了

    //通知用戶搶到了
    //扣減紅包個數
    subtracRedEnvelopeCount();
}
finally
{
    //保證由當前請求釋放，不會別的進來釋放導致亂套
    if (value.Equals(database.StringGet("distributed-lock")))
    {
        //釋放鎖
        bool isDelete = database.KeyDelete("distributed-lock");
    }
}

4.最后我們使用StackExchange.Redis提供的方法來實現封裝一個分布式鎖，首先擴展IServiceCollection 將StackExchange.Redis 操作實例注入

//擴展連接客戶端
public static class RedisServiceCollectionExtensions
{
    public static IServiceCollection AddRedisCache(this IServiceCollection services, string connectionString)
    {
        ConfigurationOptions configuration = ConfigurationOptions.Parse(connectionString);
        ConnectionMultiplexer connectionMultiplexer = ConnectionMultiplexer.Connect(connectionString);
        services.AddSingleton(connectionMultiplexer);
        return services;
    }
}
//注入容器
services.AddRedisCache("127.0.0.1:6379,password=xxx,connectTimeout=2000");

5.具體利用LockTake通過輪訓詢問的方式獲取鎖，通過LockRelease釋放鎖，在實現的源碼中，加鎖的原理就是設置一個string類型的key value以及過期時間

public class DistributedLock
{
    private readonly ConnectionMultiplexer _connectionMultiplexer;
    private IDatabase db = null;
    public DistributedLock(ConnectionMultiplexer connectionMultiplexer)
    {
       _connectionMultiplexer = connectionMultiplexer;
       db = connectionMultiplexer.GetDatabase(0);
    }

    private void FetchLock()
    {
      while (true)
       {
          //當前鎖的持有時間超過60毫秒就會被釋放，這個時間應該設置比被鎖住業務執行時間長
          bool flag = db.LockTake("distributed-lock", Thread.CurrentThread.ManagedThreadId, TimeSpan.FromSeconds(60));
          if (flag)  // 1、如果加鎖成功直接退出，否則繼續獲取鎖
          {
             break;
          }
           Thread.Sleep(10); // 防止死循環。通過等待10毫秒時間，釋放資源
       }
   }

  public void UnLock()
  {
     bool flag = db.LockRelease("distributed-lock", Thread.CurrentThread.ManagedThreadId);
     _connectionMultiplexer.Close();
  }

}

6.基於Redis實現的分布式鎖，在Redis主從或者哨兵乃至集群中，會存在鎖丟失的情況，具體就是當請求給Master節點寫入鎖后，Master節點宕機，此時主從還沒來得及同步，但此時已經選擇了一個Slaver為主節點，而最開始的設置的鎖不見了，針對這種情況比較極端，可以利用Zookeeper來實現分布式鎖作為解決方案，因為在數據一致性上Zookeeper是優於Redis的，當然Redis的性能也是較Zookeeper更為突出一些，用CAP理論來定義的話，Zookeeper趨向於CP更關注一致性,Redis趨向於AP更關注可用性.

4.Redis容災高可用

就服務實例而言，單個服務實例保證容災和高可用基本很難，因為一個實例隨時都有宕機的風險，所以引入了集群的概念，通過多個服務實例，來共同保證系統提供穩定的服務，其實在Redis中也一樣，也有保證容災和高可用的方案，在一定程度上，實現的方案在大的策略層面思路都差不多，只是具體實現的方式不同罷了，至於最終選擇保證數據信息層面的高可用，還是性能資源層面的高可用，需要用戶去根據自己業務來分析，選擇使用哪一種方案。

在Redis中，有3種方式來實現高可用的方案，分別是

• 1.主從模式 （Master Slaver）

• 2.哨兵模式 (Sentinel)

• 3.數據分片 (Cluster)

1.主從模式和哨兵

主從模式是Redis提供高可用的第一種方案，理解起來也比較簡單，一個主節點多個從節點，主節點負責寫入，然后會把數據同步到從節點，從節點負責讀取，但是會存在不可避免的缺陷

• 1.如果宕機，它不具備自動容錯和恢復功能，需要手動切換從節點為主節點。

• 2.宕機導致數據未及時同步，降低可用性

• 3.所有節點數據一樣，內存浪費，並且在線擴容復雜，伸縮性差，海量數據存儲無法解決。

• 4.高並發主節點寫入數據壓力極大。

哨兵模式是Redis提供的第二種方案，可以說是第一種主從方案的完善版本，因為它在主從的基礎上，增加了監控功能，主節點宕機后提供哨兵自動選舉master並將其他的slave指向新的master，哨兵是一個獨立運行的進程，它的實現原理是哨兵進程向所有的redis機器發送心跳消息，從而監控運行狀況,同樣由於是基於主從而誕生，那么也存在主從中的一些缺陷。

2.集群分片模式

1.集群和分片基本概念

前面2種方案歸根結底，就是使用主從復制和讀寫分離的方式保證高可用，但是內存占用，主節點壓力過大，相較於Redis提供的第三種集群和數據分片的方案，存在的缺陷也非常明顯,下面我們介紹集群和數據分片。

1.集群和數據分片簡單理解就是將主從復制的核心策略，做了集群來橫向擴展，包含多個主從節點，采用去中心化思想，數據按照 slot 存儲分布在多個節點，並且節點間數據共享，可動態調整數據分布，

2.在集群中每一個節點都是互相通信采用 Gossip 病毒協議，在一個節點寫入的數據，其他任何節點都會查詢到。

2.集群內部數據存取原理

1.Slot 槽，存儲數據的容器,總共16384個Slot ,采用平均分配，並且只有主節點才能被分配

2.Hash算法用於對數據進行計算，得到一個Hash值，然后對16384取模選擇存到哪個區間的Slot

3.Windows下集群搭建

1.我們部署集群使用redis-trib.rb，首先下載環境 提取碼1234

2.安裝需要用到Ruby語言運行環境，然后進入安裝目錄Ruby22-x64中，通過gem命令安裝文件

gem install –-local youFloder\redis-3.2.2.gem

3.准備Redis服務實例，此處准備6個實例，並配置Redis配置文件，然后啟動每一個服務實例

redis-server redis-6380.conf 
redis-server redis-6381.conf 
....
....

port 6380 #節點端口
bind 127.0.0.1   #節點主機ip
appendonly yes 開啟aof
appendfilename "appendonly.6380.aof"   
cluster-enabled yes                                    
cluster-config-file nodes.6380.conf
cluster-node-timeout 15000
cluster-slave-validity-factor 10
cluster-migration-barrier 1
cluster-require-full-coverage yes

3.進入到redis-trib.rb目錄打開命令窗口，進行搭建集群

ruby redis-trib.rb create -–replicas 1 127.0.0.1:6380 127.0.0.1:6381 127.0.0.1:6382 127.0.0.1:6383 127.0.0.1:6384 127.0.0.1:6385

4.檢查集群狀態,出現圖片中的提示代表搭建成功

ruby redis-trib.rb check 127.0.0.1:6380