分散式應用進行邏輯處理時經常會遇到併發問題，我們首先肯定會想到鎖。關於鎖大家都很熟悉。在併發程式設計中，我們通過鎖，來避免由於競爭而造成的資料不一致問題。通常我們使用 synchronized 、Lock 來加鎖。但是 Java中的鎖，只能保證在同一個 JVM程序內中執行。如果在分散式叢集環境下呢？

一、分散式鎖

分散式鎖的本質與 Java中的鎖一樣，就是在 Redis裡面佔了一個“坑”（一個固定 key 的值），當別的程序也要來佔坑時（給key 設定值時）發現坑已經有人了（key 已經有值）了，就只好放棄或者稍後再試。一般使用 setnx(set if not exists)指令，當 key不存在時返回1，否則返回0；呼叫 del指令釋放key 的值。

1 > setnx lock true                #加鎖
2 OK
3 ... do something critical ...    #業務處理
4 > del lock                       #釋放鎖
5 (integer> 1

存在一個問題：如果邏輯執行中間出現異常，可能會導致 del指令沒有被呼叫，這樣就是陷入死鎖。於是當拿到鎖後，應該給鎖加上一個過期時間，這樣即使中間出現了異常，也可以保證在規定的時間內自動釋放鎖。

1 > setnx lock true
2 OK
3 > expire lock 5
4
 
 ... do something critical ...
5 >del lock
6 (integer) 1

但是上述的邏輯也存在問題，如果在 setnx 與 expire 之間伺服器程序突然掛掉了，也會導致死鎖。這個問題的根源是因為 setnx 與 expire 不是原子的。如果這兩個命令一塊執行就沒問題了。redis2.8版本中，提供瞭如下命令（重點）：加鎖保證了原子性

1 > set lock true ex 5 nx    #設定key=lock vlue=true 過期時間5s
2 OK
3 ... do something critical ...
4 > del
 
 lock

但是在 Redis 叢集環境下，這種方式是有缺陷的，它不是絕對的安全。在哨兵模式中，當主節點掛掉時，從節點會取而代之，但客戶端上卻沒有明顯感知。比如，原先第一個客戶端在主節點中申請成功了一把鎖，但是這把鎖還沒來得及同步到從節點，主節點就掛掉了，然後從節點變成了主節點，這個新的主節點內部沒有這個鎖，所以當另一個客戶端過來請求加鎖，立即就批准了。這樣就會導致系統中同樣一把鎖被兩個客戶端同時持有，不安全性由此產生。不過這種不安全也僅在主從發生 failover 的情況下才會產生，而且持續時間極短，業務系統多數情況下可以容忍。

為了解決這個問題，Antirez 發明了 Redlock演算法，它的流程比較複雜，不過已經有了很多開源的 library 做了良好的封裝，使用者可以拿來即用，比如 redlock-py。

 1 addrs = [{
 2     "host" : "localhost",
 3     "port" : 6379,
 4     "db": 0
 5 },{
 6     "host" : "localhost",
 7     "port" : 6479,
 8     "db": 0
 9 },{
10     "host" : "localhost",
11     "port" : 6579,
12     "db": 0
13 }]
14 
15 dlm = redlock.Redlock(addrs)
16 success = dlm.lock("user-lck",5000)
17 if success:
18     print "lock success"
19     dlm.unlock('user-lck')
20 else:
21     print 'ock failed'

為了使用 Redlock，需要提供多個 Redis 例項，這些例項之間相互獨立，沒有主從關係。同很多分散式演算法一樣，Redis 也使用“大多數機制”。加鎖時，它會向過半節點發送set(key,value,nx=True,ex=xxx)指令，只要過半節點set 成功，就認為加鎖成功。釋放鎖時，需要向所有節點發送del 指令。不過 Redlock 演算法還需要考慮出錯重試，時鐘漂移等很多細節問題，同時因為Redlock 需要向多個節點進行讀寫，意味著其相比單例項 Redis 的效能會下降一些。

Redlock 使用場景：如果你很在乎高可用性，希望即使掛了一臺 Redis 也完全不受影響，就應該考慮 Redlock。不過代價也是有的，需要更多的 Redis 例項，效能也下降了，程式碼上還需要引入額外的 library，運維上也需要特殊對待，這些都是需要考慮的成本。

二、超時問題

Redis 的分散式鎖不能解決超時問題，如果在加鎖和釋放鎖之間的邏輯執行時間超出了鎖的過期時間，就會出現問題：執行緒1刪除的鎖不是自己的（自己的已經過期，自動刪除），而是剛獲取到分散式鎖的執行緒2的 key值。【簡單點就是釋放了別人剛拿到的鎖】，有一種稍微安全一點的方案是將 set 指令的 value 引數設定為一個隨機數，釋放鎖時，先匹配隨機數是否一致，然後再刪除key，就可以避免當前執行緒佔用的鎖，不會被其他執行緒所刪除。除非這個鎖是因為過期了而被伺服器自動釋放了。但是匹配 value 和刪除 key 不是一個原子操作，Redis 也沒有提供類似原子性的操作。這就需要使用 Lua 指令碼來處理了，因為 Lua 指令碼可以保證連續多個指令的原子性執行。

1 if redis.call("get",KEYS[1]) == ARGV[1] then
2    return redis.call("del",KEYS[1])
3 else
4    return 0
5 end

但是這也不是一個完美的方案，它只是相對安全一點，因為如果真的超時了，當前執行緒的邏輯沒有執行完，其他執行緒也會乘虛而入。

三、可重入性

可重入性是指執行緒在持有鎖的情況下再次請求加鎖，如果一個鎖支援同一個執行緒多次加鎖，那麼這個鎖就是可重入鎖。比如Java語言中的 ReentrantLock 就是可重入鎖。Redis分散式鎖如果支援可重入，需要對客戶端的set 方法進行包裝，使用執行緒的 Threadlocal 變數儲存當前持有鎖的計數。還需要考慮記憶體鎖計數的過期時間。

 1 /**
 2  * 可重入鎖
 3  */
 4 public class RedisWithReentrantLock {
 5 
 6     //每一個執行緒都可以獨立地改變自己的副本，而不會影響其它執行緒所對應的副本
 7     private ThreadLocal<Map<String,Integer>> lockers = new ThreadLocal<>();
 8 
 9     private Jedis jedis;
10     //構造器
11     public RedisWithReentrantLock(Jedis jedis) {
12         this.jedis = jedis;
13     }
14 
15     //判斷當前key是否已存在
16     private boolean _lock(String key){
17         return jedis.set(key,"","nx","ex",5L) != null;
18     }
19 
20     //釋放鎖
21     private void _unlock(String key){
22         jedis.del(key);
23     }
24 
25     //從當前執行緒中獲取鎖資訊
26     private Map<String,Integer> currentLockers(){
27         //從當前執行緒中獲取存放的值
28         Map<String,Integer> refs = lockers.get();
29         if(refs != null){
30             return refs;
31         }
32         lockers.set(new HashMap<>());
33         return lockers.get();
34     }
35 
36     //加鎖
37     public boolean lock(String key){
38         //給當前執行緒設定值
39         Map<String, Integer> refs = currentLockers();
40         Integer refCnt = refs.get(key);
41         if(refCnt != null){
42             refs.put(key,refCnt+1);
43             return true;
44         }
45         boolean ok = this._lock(key);
46         if(!ok){
47             return false;
48         }
49         refs.put(key,1);
50         return true;
51     }
52     
53     //釋放鎖
54     public boolean unlock(String key){
55         Map<String, Integer> refs = currentLockers();
56         Integer refCnt = refs.get(key);
57         if(refCnt == null){
58             return false;
59         }
60         refCnt-=1;
61         if(refCnt > 0){
62             refs.put(key,refCnt);
63         }else{
64             refs.remove(key);
65             this._unlock(key);
66         }
67         return true;
68     }
69 
70     public static void main(String[] args) {
71         Jedis jedis = new Jedis();
72         RedisWithReentrantLock redis = new RedisWithReentrantLock(jedis);
73         redis.lockers("codehole");
74         redis.lockers("codehole");
75         redis.unlock("codehole");
76         redis.unlock("codehole");
77     }
78 }

四、Redisson

在併發較大的情況下，直接在 Redis 中扣減庫存一定會導致商品出現超買現象，可以引入分散式鎖來避免超賣。當然分散式鎖自身必須滿足一下三點要求：
【1】在任何情況下分散式鎖都不能淪為系統瓶頸；
【2】不能產生死鎖；
【3】支援鎖重入；
相比 Jedis，Redisson 確實算得上一款嶄新的 Redis 客戶端 API，它支援豐富的資料型別，並且是執行緒安全的，底層還使用了 Netty4 進行網路通訊。那麼我們能夠在程式中用 Redisson 代替 Jedis 來與 Redis 進行互動？其實，Redisson 僅僅是為了擴充套件 Jedis 的部分功能，兩者是並存的，比如Redisson 並不支援 String 型別的資料結構。

1 <dependency>
2     <groupId>org.redisson</groupId>
3     <artifactId>redisson</artifactId>
4     <version>2.2.11</version>
5 </dependency>

在程式中使用 Redisson 客戶端實現基於 Redis 的分散式鎖，如下：

 1 private static Config config;
 2 private static ClusterServersConfig clusterServersConfig;
 3 private static String address = "127.0.0.1:6379";
 4 public static @BeforeClass void init(){
 5     config = new Config();
 6     //使用叢集模式
 7     clusterServersConfig = config.useClusterServers();
 8     clusterServersConfig.addNodeAddress(address);
 9     clusterServersConfig.setMasterConnectionPoolSize(100);
10     clusterServersConfig.setSlaveConnectionPoolSize(100);
11     clusterServersConfig.setTimeout(1000);
12 }
13 
14 public @Test void testLock(){
15     Redisson Redisson = null;
16     try{
17         redisson = Redisson.create(config);
18         RLock lock = redisson.getLock("testLock");
19         //獲取鎖
20         lock.lock(20,TimeUnit.MILLISECONDS);
21         //釋放鎖
22         lock.unlock();
23         //嘗試獲取鎖
24         boolean result = lock.tryLock(10,20,TimeUnit.MILLISECONDS);
25         //判斷是否成功獲取到鎖
26         if(result){
27             lock.forceUnlock();
28         }
29     } catch (Exception e){
30         e.printStackTrace();
31     } finally {
32         if(null != redisson){
33             redisson.shutdown();
34         }
35     }
36 }

上述程式中，使用了兩種獲取分散式鎖的方法。lock(long leaseTime,TimeUnit unit) 方法中的第1個引數用於設定分散式鎖的租約時間，而第2個引數則為時間單位。使用這種方式意味著在某一個獲取到鎖的執行緒未釋放鎖之前，其他執行緒只能夠在佇列中阻塞等待，這和 InnoDB 引擎提供的行鎖機制如出一轍，併發越高等待的執行緒越多。因此，在併發較大時，建議使用 tryLock(long waitTime,long leaseTime,TimeUnit unit) 方法獲取分散式鎖。

在 tryLock() 方法中開發人員可以通過引數 “waitTime” 來設定獲取分散式鎖的等待時間，超出規定的時間閾值後，執行緒將不再繼續等待拿鎖；那麼為了提升庫存扣減的成功率，可以在獲取鎖失敗後嘗試多次。相比 lock() 方法的拿鎖方式，後者在併發較大的情況下不會使分散式鎖淪為系統瓶頸，但是商品庫存的扣減成功率會受到一定影響。

例如將商品庫存的扣減操作轉移到 Redis 中主要是為了避免資料庫淪為系統瓶頸。既然效能問題得到了解決，那麼變化後的實時庫存應該如何同步到資料庫？當系統獲取到分散式鎖併成功扣減 Redis 中的實時庫存後，可以將訊息寫入到訊息佇列中，由消費者負責實際庫存的扣減。由於採用了排隊機制，併發寫入資料庫時的流量可控，因此資料庫的負載壓力就會始終保持在一個恆定的範圍內，不會因為流量的影響而導致資料庫效能下降。