本文涉及內(nèi)容：

• 分布式鎖介紹；
• 用數(shù)據(jù)表做分布式鎖原理介紹 & 數(shù)據(jù)表設(shè)計(jì)；
• 用redis做分布式鎖原理介紹 & 代碼實(shí)操；
• 用redisson做分布式鎖原理介紹 & 代碼實(shí)操；
• 用zookeeper做分布式鎖原理介紹；
• 用curator做分布式鎖代碼實(shí)操；
• 實(shí)現(xiàn)分布式鎖的各方案比較；
• 完整項(xiàng)目的GitHub地址

一、是什么？

1、鎖的應(yīng)用場(chǎng)景：

在單體應(yīng)用中，我們會(huì)使用ReentrantLock或Synchronized來(lái)應(yīng)對(duì)并發(fā)場(chǎng)景。比如最常見的賣票場(chǎng)景，假如總共有100張票，線程A和線程B同時(shí)操作，如下圖：

這時(shí)有一個(gè)共享變量100，線程A和B將100拷貝到自己的工作內(nèi)存中，當(dāng)線程A搶到執(zhí)行權(quán)的時(shí)候，此時(shí)A工作內(nèi)存中的值是100，然后售票，進(jìn)行自減操作，將自己工作內(nèi)存中的值變成了99。當(dāng)A還沒(méi)來(lái)得及將99刷回到主內(nèi)存的時(shí)候，線程B進(jìn)來(lái)了，此時(shí)B拿到的主內(nèi)存的值還是100，然后售票，進(jìn)行自減，也是99。這就出現(xiàn)了同一張票出售了兩次的情況。所以我們會(huì)加鎖加volatile保證原子性保證可見性。

2、分布式鎖是什么？

上面的場(chǎng)景中，我們可以通過(guò)ReentrantLock或者Synchronized搞定，因?yàn)槟愕捻?xiàng)目只運(yùn)行在一臺(tái)服務(wù)器上，只有一個(gè)JVM，所有的共享變量都加載到同一個(gè)主內(nèi)存中。而分布式應(yīng)用中，一個(gè)項(xiàng)目部署在多臺(tái)服務(wù)器上，最基本的架構(gòu)如下圖：

比如現(xiàn)在server1、server2和server3讀取到數(shù)據(jù)庫(kù)的票數(shù)都是100，在每一個(gè)server中，我們可以用JDK的鎖來(lái)保證多個(gè)用戶同時(shí)訪問(wèn)我這臺(tái)server時(shí)不會(huì)出問(wèn)題。但問(wèn)題是，如果client1訪問(wèn)到的是server1，票數(shù)是100，然后購(gòu)票，還沒(méi)來(lái)得及將數(shù)據(jù)庫(kù)票數(shù)改為99，client2也開始訪問(wèn)系統(tǒng)購(gòu)票了，client2如果訪問(wèn)的是server1，自然不會(huì)出問(wèn)題，如果訪問(wèn)的是server2，這時(shí)server2讀取到數(shù)據(jù)庫(kù)的票數(shù)還是100，那么就出問(wèn)題了，又出現(xiàn)了同一張票賣了兩次的情況。在分布式應(yīng)用中，JDK的鎖機(jī)制就無(wú)法滿足需求了，所以就出現(xiàn)了分布式鎖。

3、分布式鎖應(yīng)該滿足的條件：

• 四個(gè)一：同一個(gè)方法在同一時(shí)刻只能被一臺(tái)機(jī)器的一個(gè)線程執(zhí)行
• 三個(gè)具備：具備可重入特性；具備鎖失效機(jī)制，防止死鎖；具備非阻塞鎖特性，即沒(méi)獲取到鎖返回獲取鎖失敗，而不是一直等待
• 兩個(gè)高：高性能地獲取與釋放鎖；高可用的獲取與釋放鎖

4、分布式鎖的實(shí)現(xiàn)方式：

• 基于數(shù)據(jù)庫(kù)：用數(shù)據(jù)庫(kù)的排他鎖實(shí)現(xiàn)
• 基于redis：利用redis的set key value NX EX 30000；也可以用redis的第三方庫(kù)比如Redisson
• 基于zookeeper：利用zookeeper的臨時(shí)順序節(jié)點(diǎn)實(shí)現(xiàn)；也可以用zookeeper的第三方庫(kù)比如Curator

二、基于數(shù)據(jù)庫(kù)實(shí)現(xiàn)

1、建表：

CREATE TABLE `tb_distributed_lock` (
 `dl_id` INT NOT NULL auto_increment COMMENT '主鍵，自增',
 `dl_method_name` VARCHAR (64) NOT NULL DEFAULT '' COMMENT '方法名',
 `dl_device_info` VARCHAR (100) NOT NULL DEFAULT '' COMMENT 'ip+線程id',
 `dl_operate_time` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '數(shù)據(jù)被操作的時(shí)間',
 PRIMARY KEY (`dl_id`),
 UNIQUE KEY `uq_method_name` (`dl_method_name`) USING BTREE
) ENGINE = INNODB DEFAULT charset = utf8 COMMENT = '分布式鎖表';

2、思路：

當(dāng)執(zhí)行一個(gè)方法的時(shí)候，我們首先嘗試往表中插入一條數(shù)據(jù)。如果插入成功，則占鎖成功，繼續(xù)往下執(zhí)行，執(zhí)行完刪除該記錄。如果插入失敗，我們?cè)僖?code style="font-size: 14px;word-wrap: break-word;padding: 2px 4px;border-radius: 4px;margin-right: 2px;margin-left: 2px;background-color: rgba(27, 31, 35, 0.05);font-family: "Operator Mono", Consolas, Monaco, Menlo, monospace;word-break: break-all;color: rgb(40, 202, 113);">當(dāng)前方法名、當(dāng)前機(jī)器ip+線程id、數(shù)據(jù)被操作時(shí)間為5分鐘內(nèi)(5分鐘表示鎖失效的時(shí)間)為條件去查詢，如果有記錄，表示該機(jī)器的該線程在5分鐘內(nèi)占有過(guò)鎖了，直接往下執(zhí)行最后刪除記錄；如果沒(méi)有記錄，占有鎖失敗。一個(gè)用戶就是一個(gè)線程，所以我們可以把機(jī)器ip和用戶id組合一起當(dāng)成dl_device_info。

3、占有鎖和釋放鎖：

• 占有鎖：

INSERT INTO tb_distributed_lock (
 dl_method_name,
 dl_device_info
)
VALUES
 ('方法名', 'ip&用戶id');

如果insert失敗，則：

SELECT
 count(*)
FROM
 tb_distributed_lock
WHERE
 dl_method_name = '方法名'
AND dl_device_info = 'ip&用戶id'
AND dl_operate_time < SYSDATE() - 5;

• 釋放鎖：

DELETE
FROM
 tb_distributed_lock
WHERE
 dl_method_name = '方法名'
AND dl_device_info = 'ip&用戶id';

4、小總結(jié)：

以上表結(jié)構(gòu)可能并不是很好，只是提供了這么一個(gè)思路。下面說(shuō)它的優(yōu)缺點(diǎn)：

• 優(yōu)點(diǎn)：成本低，不需要引入其他的技術(shù)
• 缺點(diǎn)：對(duì)數(shù)據(jù)庫(kù)依賴性強(qiáng)，如果數(shù)據(jù)庫(kù)掛了，那就涼涼了，所以數(shù)據(jù)庫(kù)最好也是高可用的

三、基于redis實(shí)現(xiàn)

1、原理：

基于redis的set key value nx ex 30，這條語(yǔ)句的意思就是如果key不存在就設(shè)置，并且過(guò)期時(shí)間為30s，如果key已經(jīng)存在就會(huì)返回false。如果要以毫秒為單位，把ex換成px就好了。我們執(zhí)行方法前，先將方法名當(dāng)成key，執(zhí)行這條語(yǔ)句，如果執(zhí)行成功就是獲取鎖成功，執(zhí)行失敗就是獲取鎖失敗。

2、代碼實(shí)現(xiàn)：

• RedisUtil的部分代碼：

/**
* key不存在時(shí)就設(shè)置，返回true，key已存在就返回false
* @param key
* @param value
* @param timeout
* @return
*/
public static boolean setIfAbsent(String key, String value, Long timeout) {
 return redisTemplate.opsForValue().setIfAbsent(key, value, timeout, TimeUnit.SECONDS);
}
/**
* 獲取key-value
* @param key
* @return
*/
public static String getString(String key) {
 return (String) redisTemplate.opsForValue().get(key);
}
/**
* 刪除key
* @param key
* @return
*/
public static boolean delKey(String key) {
 return redisTemplate.delete(key);
}

• 業(yè)務(wù)方法中使用：

public String hello() {
 // 方法名當(dāng)作key
 String key = "hello";
 String value = "hellolock";
 if (RedisUtil.setIfAbsent(key, value, 60 * 2L)) {
  System.out.println("成功獲取到鎖，開始執(zhí)行業(yè)務(wù)邏輯……");
  // 假如執(zhí)行業(yè)務(wù)邏輯需要1分鐘
  try {TimeUnit.MINUTES.sleep(1L); } catch (Exception e) { e.printStackTrace();};
  // 釋放鎖先校驗(yàn)value，避免釋放錯(cuò)
  if (value.equals(RedisUtil.getString(key))) {
   RedisUtil.delKey(key);
   System.out.println("執(zhí)行完業(yè)務(wù)邏輯，釋放鎖成功");
  }
  return "success";
 } else {
  System.out.println("鎖被別的線程占有，獲取鎖失敗");
  return "acquire lock failed";
 }
}

3、小總結(jié)：

• 優(yōu)點(diǎn)：簡(jiǎn)單易用，一條redis命令就搞定?？梢栽O(shè)置過(guò)期時(shí)間，避免釋放鎖失敗造成其他線程長(zhǎng)時(shí)間無(wú)法獲取鎖的問(wèn)題。

• 缺點(diǎn)：這種做法只適合redis是單機(jī)的時(shí)候，如果redis有集群，這樣做就會(huì)出問(wèn)題。假如一個(gè)線程在master上獲取鎖成功了，在master還沒(méi)來(lái)得及將數(shù)據(jù)同步到slave上的時(shí)候，master掛了，slave升級(jí)為master。第二個(gè)線程進(jìn)來(lái)嘗試獲取鎖，因?yàn)樾碌膍aster上并沒(méi)有這個(gè)key，所以，也能成功獲取到鎖。

• 解決辦法：針對(duì)上面的缺點(diǎn)，我們可以采用redis的RedLock算法。假如集群中有n個(gè)redis，我們先從這n個(gè)redis中嘗試獲取鎖（鎖的過(guò)期時(shí)間為x），并記錄獲取鎖的消耗的總時(shí)間t，獲取鎖成功數(shù)量為s，當(dāng)且僅當(dāng)t < x 并且 s >= (n/2 + 1)時(shí)，認(rèn)為獲取鎖成功。

四、基于Redisson實(shí)現(xiàn)

1、是什么？

官網(wǎng)地址：https://github.com/redisson/redisson/wiki/Table-of-Content Redisson是一個(gè)功能十分強(qiáng)大的redis客戶端，封裝了很多分布式操作，比如分布式對(duì)象、分布式集合、分布式鎖等。它的分布式鎖也很多，什么公平鎖、可重入鎖、redlock等一應(yīng)俱全，下面來(lái)看看如何在springboot項(xiàng)目中使用它。

2、使用redisson做分布式鎖：

• 添加依賴：

<!-- redisson-springboot-starter -->
<dependency>
 <groupId>org.redisson</groupId>
 <artifactId>redisson-spring-boot-starter</artifactId>
 <version>3.12.3</version>
</dependency>
<!-- io.netty/netty-all  -->
<dependency>
 <groupId>io.netty</groupId>
 <artifactId>netty-all</artifactId>
</dependency>

• application.yml：

spring:
  application:
    name: distributed-lock
  redis:
    # redis單機(jī)版的寫法
    host: 192.168.2.43
    port: 6379
    # 集群的寫法
    #cluster:
      #nodes:
      #- 192.168.0.106,192.168.0.107
    #哨兵的寫法
    #sentinel:
      #master: 192.168.0.106
      #nodes:
      #- 192.168.0.107,192.168.0.108

• 用法：直接注入RedissonClient，然后用它獲取鎖，得到鎖之后就可以進(jìn)行占鎖和釋放鎖了。有阻塞式鎖，也有非阻塞式鎖，具體用法如下：

@Autowired
private RedissonClient redisson;

/**
 * 未設(shè)置過(guò)期時(shí)間，沒(méi)獲取到就會(huì)一直阻塞著
 * @return
 */
@GetMapping("/testLock")
public String testLock() {
 log.info("進(jìn)入testLock方法，開始獲取鎖");
 String key = "testLock";
 RLock lock = redisson.getLock(key);
 lock.lock();
 log.info("獲取鎖成功，開始執(zhí)行業(yè)務(wù)邏輯……");
 try {TimeUnit.SECONDS.sleep(10L); } catch (Exception e) { e.printStackTrace();};
 log.info("執(zhí)行完業(yè)務(wù)邏輯，釋放鎖");
 lock.unlock();
 return "success";
}
 
/**
 * 嘗試獲取鎖，沒(méi)獲取到就直接失敗，不會(huì)阻塞
 * @return
 */
@GetMapping("/testTryLock")
public String testTryLock() {
 log.info("進(jìn)入testTryLock方法，開始獲取鎖");
 String key = "testTryLock";
 RLock lock = redisson.getLock(key);
 boolean res = lock.tryLock();
 if (!res) {
  log.error("嘗試獲取鎖失敗");
  return "fail";
 } else {
  log.info("獲取鎖成功，開始執(zhí)行業(yè)務(wù)邏輯……");
  try {TimeUnit.SECONDS.sleep(30L); } catch (Exception e) { e.printStackTrace();};
  log.info("執(zhí)行完業(yè)務(wù)邏輯，釋放鎖");
  lock.unlock();
  return "success";
 }
}
 
/**
 * 鎖設(shè)置了過(guò)期時(shí)間，即使最后面的unlock失敗，20秒后也會(huì)自動(dòng)釋放鎖
 * @return
 */
@GetMapping("/testLockTimeout")
public String testLockTimeout() {
 log.info("進(jìn)入testLockTimeout方法，開始獲取鎖");
 String key = "testLockTimeout";
 RLock lock = redisson.getLock(key);
 // 20秒后自動(dòng)釋放鎖
 lock.lock(20, TimeUnit.SECONDS);
 log.info("獲取鎖成功，開始執(zhí)行業(yè)務(wù)邏輯……");
 try {TimeUnit.SECONDS.sleep(10L); } catch (Exception e) { e.printStackTrace();};
 lock.unlock();
 return "success";
}
 
/**
 * 嘗試獲取鎖，15秒還沒(méi)獲取到就獲取鎖失??；獲取到了會(huì)持有20秒，20秒后自動(dòng)釋放鎖
 * @return
 */
@GetMapping("/testTryLockTimeout")
public String testTryLockTimeout() {
 log.info("進(jìn)入testTryLockTimeout方法，開始獲取鎖");
 String key = "testTryLockTimeout";
 RLock lock = redisson.getLock(key);
 boolean res = false;
 try {
  res = lock.tryLock(15, 20, TimeUnit.SECONDS);
 } catch (InterruptedException e1) {
  e1.printStackTrace();
 }
 if (!res) {
  log.error("嘗試獲取鎖失敗");
  return "fail";
 } else {
  log.info("獲取鎖成功，開始執(zhí)行業(yè)務(wù)邏輯……");
  try {TimeUnit.SECONDS.sleep(10L); } catch (Exception e) { e.printStackTrace();};
  log.info("執(zhí)行完業(yè)務(wù)邏輯，釋放鎖");
  lock.unlock();
  return "success";
 }
}

3、小總結(jié)：

以上就是使用redisson做分布式鎖的簡(jiǎn)單demo，用起來(lái)十分的方便。上面是與springboot項(xiàng)目集成，直接用它提供的springboot的starter就好了。用它來(lái)做分布式鎖的更多用法請(qǐng)移步至官網(wǎng)：redisson分布式鎖。

五、基于zookeeper實(shí)現(xiàn)

1、zookeeper知識(shí)點(diǎn)回顧：

zookeeper有四種類型的節(jié)點(diǎn)：

• 持久節(jié)點(diǎn)：默認(rèn)的節(jié)點(diǎn)類型，客戶端與zookeeper斷開連接后，節(jié)點(diǎn)依然存在

• 持久順序節(jié)點(diǎn)：首先是持久節(jié)點(diǎn)，順序的意思是，zookeeper會(huì)根據(jù)節(jié)點(diǎn)創(chuàng)建的順序編號(hào)

• 臨時(shí)節(jié)點(diǎn)：客戶端與zookeeper斷開連接后節(jié)點(diǎn)不復(fù)存在

• 臨時(shí)順序節(jié)點(diǎn)：客戶端與zookeeper斷開連接后節(jié)點(diǎn)不復(fù)存在，zookeeper會(huì)根據(jù)節(jié)點(diǎn)創(chuàng)建的順序編號(hào)

2、基于zookeeper實(shí)現(xiàn)分布式鎖的原理：

我們正是利用了zookeeper的臨時(shí)順序節(jié)點(diǎn)來(lái)實(shí)現(xiàn)分布式鎖。首先我們創(chuàng)建一個(gè)名為lock(節(jié)點(diǎn)名稱隨意)的持久節(jié)點(diǎn)。線程1獲取鎖時(shí)，就在lock下面創(chuàng)建一個(gè)名為lock1的臨時(shí)順序節(jié)點(diǎn)，然后查找lock下所有的節(jié)點(diǎn)，判斷自己的lock1是不是第一個(gè)，如果是，獲取鎖成功，繼續(xù)執(zhí)行業(yè)務(wù)邏輯，執(zhí)行完后刪除lock1節(jié)點(diǎn)；如果不是第一個(gè)，獲取鎖失敗，就watch排在自己前面一位的節(jié)點(diǎn)，當(dāng)排在自己前一位的節(jié)點(diǎn)被干掉時(shí)，再檢查自己是不是排第一了，如果是，獲取鎖成功。圖解過(guò)程如下：

線程1創(chuàng)建了一個(gè)lock1，發(fā)現(xiàn)lock1的第一個(gè)節(jié)點(diǎn)，占鎖成功；在線程1還沒(méi)釋放鎖的時(shí)候，線程2來(lái)了，創(chuàng)建了一個(gè)lock2，發(fā)現(xiàn)lock2不是第一個(gè)，便監(jiān)控lock1，線程3此時(shí)進(jìn)行就監(jiān)控lock2。直到自己是第一個(gè)節(jié)點(diǎn)時(shí)才占鎖成功。假如某個(gè)線程釋放鎖的時(shí)候zookeeper崩了也沒(méi)關(guān)系，因?yàn)槭桥R時(shí)節(jié)點(diǎn)，斷開連接節(jié)點(diǎn)就沒(méi)了，其他線程還是可以正常獲取鎖，這就是要用臨時(shí)節(jié)點(diǎn)的原因。

說(shuō)清楚了原理，用代碼實(shí)現(xiàn)也就不難了，可以引入zookeeper的客戶端zkClient，自己寫代碼實(shí)現(xiàn)(偷個(gè)懶，自己就不寫了，有興趣的可以參考我zookeeper的文章，肯定可以自己寫出來(lái)的)。不過(guò)有非常優(yōu)秀的開源解決方案比如curator，下面就看看curator怎么用。

六、基于curator實(shí)現(xiàn)

1、springboot整合curator：

• pom.xml：

<!-- curator start-->
<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.2.0</version>
</dependency>
<dependency>
 <groupId>org.apache.curator</groupId>
 <artifactId>curator-recipes</artifactId>
 <version>4.2.0</version>
</dependency>
<dependency>
 <groupId>org.seleniumhq.selenium</groupId>
 <artifactId>selenium-java</artifactId>
</dependency>
<!-- curator end-->

• application.yml：注意，curator下面這些屬性spring是沒(méi)有集成的，也就是說(shuō)寫的時(shí)候不會(huì)有提示

curator:
  retryCount: 5 # 連接失敗的重試次數(shù)
  retryTimeInterval: 5000 # 每隔5秒重試一次
  url: 192.168.2.43:2181 # zookeeper連接地址
  sessionTimeout: 60000 # session超時(shí)時(shí)間1分鐘
  connectionTimeout: 5000 # 連接超時(shí)時(shí)間5秒鐘

• 配置類：讀取application.yml中的屬性，創(chuàng)建CuratorFramework實(shí)例

@Configuration
public class CutatorConfig {

 @Value("${curator.retryCount}")
 private Integer retryCount;

 @Value("${curator.retryTimeInterval}")
 private Integer retryTimeInterval;

 @Value("${curator.url}")
 private String url;

 @Value("${curator.sessionTimeout}")
 private Integer sessionTimeout;

 @Value("${curator.connectionTimeout}")
 private Integer connectionTimeout;

 @Bean
 public CuratorFramework curatorFramework() {
  return CuratorFrameworkFactory.newClient(url, sessionTimeout, connectionTimeout,
    new RetryNTimes(retryCount, retryTimeInterval));
 }
}

• 測(cè)試類：測(cè)試整合curator框架是否成功

@SpringBootTest(classes = {DistributedLockApplication.class})
@RunWith(SpringRunner.class)
public class DistributedLockApplicationTests {
 
 @Autowired
 private CuratorFramework curatorFramework;

 @Test
 public void contextLoads() {
  curatorFramework.start();
  try {
   curatorFramework.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/zhusl", "test".getBytes());
  } catch (Exception e) {
   e.printStackTrace();
  }
 }
}

在確保zookeeper成功啟動(dòng)了的情況下，執(zhí)行這個(gè)單元測(cè)試，最后回到linux中，用zkCli.sh連接，查看是否成功創(chuàng)建節(jié)點(diǎn)。

2、使用Curator做分布式鎖：

Curator封裝了很多鎖，比如可重入共享鎖、不可重入共享鎖、可重入讀寫鎖、聯(lián)鎖等。具體可以參考官網(wǎng)：curator分布式鎖的用法。

• ZookeeperUtil.java：工具類，封裝獲取鎖，釋放鎖等方法。這里主要簡(jiǎn)單地封裝了上面說(shuō)的四種鎖，僅供參考。

@Component
@Slf4j
public class ZookeeperUtil {

 private static CuratorFramework curatorFramework;
 
 private static InterProcessLock lock;

 /** 持久節(jié)點(diǎn) */
 private final static String ROOT_PATH = "/lock/";
 
 /** 可重入共享鎖 */
 private static InterProcessMutex interProcessMutex;
 /** 不可重入共享鎖 */
 private static InterProcessSemaphoreMutex interProcessSemaphoreMutex;
 /** 可重入讀寫鎖 */
 private static InterProcessReadWriteLock interProcessReadWriteLock;
 /** 多共享鎖(將多把鎖當(dāng)成一把來(lái)用) */
 private static InterProcessMultiLock interProcessMultiLock;

 @Autowired
 private void setCuratorFramework(CuratorFramework curatorFramework) {
  ZookeeperUtil.curatorFramework = curatorFramework;
  ZookeeperUtil.curatorFramework.start();
 }

 /**
  * 獲取可重入排他鎖
  * 
  * @param lockName
  * @return
  */
 public static boolean interProcessMutex(String lockName) {
  interProcessMutex = new InterProcessMutex(curatorFramework, ROOT_PATH + lockName);
  lock = interProcessMutex;
  return acquireLock(lockName, lock);
 }

 /**
  * 獲取不可重入排他鎖
  * 
  * @param lockName
  * @return
  */
 public static boolean interProcessSemaphoreMutex(String lockName) {
  interProcessSemaphoreMutex = new InterProcessSemaphoreMutex(curatorFramework, ROOT_PATH + lockName);
  lock = interProcessSemaphoreMutex;
  return acquireLock(lockName, lock);
 }

 /**
  * 獲取可重入讀鎖
  * 
  * @param lockName
  * @return
  */
 public static boolean interProcessReadLock(String lockName) {
  interProcessReadWriteLock = new InterProcessReadWriteLock(curatorFramework, ROOT_PATH + lockName);
  lock = interProcessReadWriteLock.readLock();
  return acquireLock(lockName, lock);
 }

 /**
  * 獲取可重入寫鎖
  * 
  * @param lockName
  * @return
  */
 public static boolean interProcessWriteLock(String lockName) {
  interProcessReadWriteLock = new InterProcessReadWriteLock(curatorFramework, ROOT_PATH + lockName);
  lock = interProcessReadWriteLock.writeLock();
  return acquireLock(lockName, lock);
 }

 /**
  * 獲取聯(lián)鎖(多把鎖當(dāng)成一把來(lái)用)
  * @param lockNames
  * @return
  */
 public static boolean interProcessMultiLock(List<String> lockNames) {
  if (lockNames == null || lockNames.isEmpty()) {
   log.error("no lockNames found");
   return false;
  }
  interProcessMultiLock = new InterProcessMultiLock(curatorFramework, lockNames);
  try {
   if (!interProcessMultiLock.acquire(10, TimeUnit.SECONDS)) {
    log.info("Thread:" + Thread.currentThread().getId() + " acquire distributed lock fail");
    return false;
   } else {
    log.info("Thread:" + Thread.currentThread().getId() + " acquire distributed lock success");
    return true;
   }
  } catch (Exception e) {
   log.info("Thread:" + Thread.currentThread().getId() + " release lock occured an exception = " + e);
   return false;
  }
 }

 /**
  * 釋放鎖
  * 
  * @param lockName
  */
 public static void releaseLock(String lockName) {
  try {
   if (lock != null && lock.isAcquiredInThisProcess()) {
    lock.release();
    curatorFramework.delete().inBackground().forPath(ROOT_PATH + lockName);
    log.info("Thread:" + Thread.currentThread().getId() + " release lock success");
   }
  } catch (Exception e) {
   log.info("Thread:" + Thread.currentThread().getId() + " release lock occured an exception = " + e);
  }
 }
 
 /**
  * 釋放聯(lián)鎖
  */
 public static void releaseMultiLock(List<String> lockNames) {
  try {
   if (lockNames == null || lockNames.isEmpty()) {
    log.error("no no lockNames found to release");
    return;
   }
   if (interProcessMultiLock != null && interProcessMultiLock.isAcquiredInThisProcess()) {
    interProcessMultiLock.release();
    for (String lockName : lockNames) {
     curatorFramework.delete().inBackground().forPath(ROOT_PATH + lockName);
    }
    log.info("Thread:" + Thread.currentThread().getId() + " release lock success");
   }
  } catch (Exception e) {
   log.info("Thread:" + Thread.currentThread().getId() + " release lock occured an exception = " + e);
  }
 }
 

 /**
  * 獲取鎖
  * 
  * @param lockName
  * @param interProcessLock
  * @return
  */
 private static boolean acquireLock(String lockName, InterProcessLock interProcessLock) {
  int flag = 0;
  try {
   while (!interProcessLock.acquire(2, TimeUnit.SECONDS)) {
    flag++;
    if (flag > 1) {
     break;
    }
   }
  } catch (Exception e) {
   log.error("acquire lock occured an exception = " + e);
   return false;
  }
  if (flag > 1) {
   log.info("Thread:" + Thread.currentThread().getId() + " acquire distributed lock fail");
   return false;
  } else {
   log.info("Thread:" + Thread.currentThread().getId() + " acquire distributed lock success");
   return true;
  }
 }
}

• ZookeeperLockController.java：寫一個(gè)接口，用Curator加鎖，然后用瀏覽器進(jìn)行訪問(wèn)

@RestController
@RequestMapping("/zookeeper-lock")
public class ZookeeperLockController {
 
 @GetMapping("/testLock")
 public String testLock() {
  // 獲取鎖
  boolean lockResult = ZookeeperUtil.interProcessMutex("testLock");
  if (lockResult) {
   try {
    // 模擬執(zhí)行業(yè)務(wù)邏輯
    TimeUnit.MINUTES.sleep(1L);
   } catch (InterruptedException e) {
    e.printStackTrace();
   }
   // 釋放鎖
   ZookeeperUtil.releaseLock("testLock");
   return "success";
  } else {
   return "fail";
  }
 }
}

打開一個(gè)瀏覽器窗口訪問(wèn)，后臺(tái)打印出獲取鎖成功的日志，在1分鐘之內(nèi)，開啟另一個(gè)窗口再次訪問(wèn)，打印出獲取鎖失敗的日志，說(shuō)明分布式鎖生效了。

七、實(shí)現(xiàn)分布式鎖的各方案比較

• 基于數(shù)據(jù)庫(kù)實(shí)現(xiàn)最簡(jiǎn)單，不需要引入第三方應(yīng)用。但是因?yàn)槊看渭渔i和解鎖都要進(jìn)行IO操作，性能不是很好。
• 基于redis實(shí)現(xiàn)比較均衡，性能很好，也不是很難，比較可靠。
• 基于zookeeper實(shí)現(xiàn)難度較大，因?yàn)樾枰S護(hù)一個(gè)zookeeper集群，如果項(xiàng)目原本沒(méi)有用到zookeeper，還是用redis比較好。

本文項(xiàng)目地址：分布式鎖