緩存鎖
我們常常將緩存作為分布式鎖的解決方案，但是卻不能單純的判斷某個 key 是否存在 來作為鎖的獲得依據，因為無論是 exists 和 get 命名都不是線程安全的，都無法保證只有一個線程可以獲得鎖，存在線程爭搶，可能會有多個線程同時拿到鎖的情況（經典的 Redis “讀后寫”的問題）。
incr 緩存鎖
@Component
public class LockClient {
private StringRedisTemplate stringRedisTemplate;
private ValueOperations<String, String> valueOperations;
@Autowired
public void setStringRedisTemplate(StringRedisTemplate stringRedisTemplate) {
this.stringRedisTemplate = stringRedisTemplate;
this.valueOperations = stringRedisTemplate.opsForValue();
}
public void lockIncr() {
Long lockIncr = valueOperations.increment("lockIncr", 1);
// 說明拿到了鎖
if (lockIncr == 1) {
// 業務操作
}
}
}
- incr：遞增指定鍵對應的數值，如果不存在 key 對應的值，那么會先將 key 的值設置為 0，然后執行 incr 操作，返回遞增的值。
- 這種鎖的實現原理主要是利用 incr 命令的原子性，同一時間只會有一個線程操作這個命令。
- 這種鎖的實現方式，不在乎結果數據。保證只有唯一線程能夠執行到業務代碼。
setnx 緩存鎖
上面的鎖實現方式，我們對資源做了隔離，保證只有唯一線程可以拿到資源並執行操作。但是如果資源並不是唯一線程執行的呢？存在多個線程爭搶的情況下呢？
public void lockSetnx() {
String lock = "lockSetnx";
long millis = System.currentTimeMillis();
long timeout = millis + 3000L + 1;
try {
while (true) {
boolean setnx = valueOperations.setIfAbsent(lock, timeout + "");
if (setnx == true) {
break;
}
String oldTimeout = valueOperations.get(lock);
// 這一步是為了解決客戶端異常宕機，鎖沒有被正常釋放的時候。
// 當 p1、p2 同時執行到這里，發現鎖的時間過期了。p1、p2 同時執行 getSet 命令。
// 假設 p1 先執行成功了，那么 p1 得到的值就是原來鎖的過期時間（可以符合下面的判斷式），表示爭搶鎖成功。
// 假設 p2 后執行成功了，那么 p2 得到的值就是 p1 set 進去的值（不會符合下面的表達式），表示爭搶鎖失敗。
String oldValue = valueOperations.getAndSet(lock, timeout + "");
if (millis > Long.valueOf(oldTimeout) && millis > Long.valueOf(oldValue)) {
break;
}
// 休眠 100 毫秒，再去爭搶鎖
Thread.sleep(100);
}
// 執行業務代碼
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
if (millis < timeout) {
stringRedisTemplate.delete(lock);
}
}
}
- setnx：只有第一個線程會執行成功，返回 true，其余線程執行失敗，返回 false。
- getSet：返回 key 中的舊值，並把新的值 set 進去。
- 細細看來，好像似乎 setnx 命令就能夠實現分布式鎖了，為什么還要 getSet 命名呢？getSet 命令是為了解決客戶端異常宕機，鎖沒有被正常釋放的情況下，結合過期時間來保證線程安全。可以看看官網的介紹，有詳細解釋這個問題。
zookeeper 鎖
zookeeper，天生的分布式協調工具，生來就是為了解決各種分布式的難題，比如分布式鎖、分布式計數器、分布式隊列等等。
zookeeper 分布式鎖，如果自己實現的話，大抵的實現方式如下：
公平鎖：
- 在 zookeeper 的指定節點（locks）下創建臨時順序節點 node_n ;
- 獲取 locks 下面的所有子節點 children。
- 對子節點按節點自增序號從小到大排序。
- 判斷本節點是不是第一個子節點，如果是，則獲取到鎖。如果不是，則監聽比該節點小的那個節點的刪除事件。
- 若監聽事件生效，則回到第二步重新判斷，直到獲取到鎖。
不公平鎖
- 在 zookeeper 的某個節點（lock）上創建臨時節點 znode。
- 創建成功，就表示獲取到了這個鎖；其他客戶端來創建鎖會失敗，只能注冊對這個鎖的監聽。
- 其他客戶端監聽到這個鎖被釋放（znode節點被刪除），就會嘗試加鎖（創建節點），繼續執行第二步。
幸運的是，zookeeper recipes 客戶端為我們提供了多種分布式鎖實現：
- InterProcessMutex（可重入排他鎖）
- InterProcessSemaphoreMutex（不可重入排他鎖）
- InterProcessReadWriteLock（分布式讀寫鎖）
- InterProcessSemaphore（共享信號量 —— 設置最大並行數量）
zookeeper recipes 鎖的簡單使用：
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.14</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-framework</artifactId>
<version>4.0.1</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>4.0.1</version>
</dependency>
public InterProcessMutex interProcessMutex(String lockPath) {
CuratorFramework client = CuratorFrameworkFactory.newClient(zookeeper, new ExponentialBackoffRetry(1000, 3));
// 啟用命名空間，做微服務間隔離
client.usingNamespace(namespace);
client.start();
return new InterProcessMutex(client, lockPath);
}
public void lockUse() {
InterProcessMutex interProcessMutex = interProcessMutex("/lockpath");
try {
// 獲取鎖
if (interProcessMutex.acquire(100, TimeUnit.MILLISECONDS)) {
// 執行業務代碼
}
} catch (Exception e) {
e.printStackTrace();
} finally {
// 釋放鎖
try {
interProcessMutex.release();
} catch (Exception e) {
e.printStackTrace();
}
}
}
- 推薦一篇 zookeeper 介紹很全面的文章：https://www.cnblogs.com/shamo89/p/9800925.html
比較
- 緩存分布式鎖，必須采用輪詢的方式去嘗試加鎖，對性能浪費很大；zookeeper 分布式鎖，可以通過監聽的方式等待通知或超時，當有鎖釋放，通知使用者即可。
- 如果緩存獲取鎖的那個客戶端宕機了，鎖不會被釋放，只能通過其它方式解決（上面的 getSet 判斷）；而 zookeeper 的話，因為創建的是臨時 znode，只要客戶端掛了，znode 就沒了，此時就自動釋放鎖。