本文涉及內容：

分布式鎖介紹；
用資料表做分布式鎖原理介紹 & 資料表設計；
用redis做分布式鎖原理介紹 & 代碼實操；
用redisson做分布式鎖原理介紹 & 代碼實操；
用zookeeper做分布式鎖原理介紹；
用curator做分布式鎖代碼實操；
實作分布式鎖的各方案比較；
完整專案的GitHub地址

一、是什么？

1、鎖的應用場景：
在單體應用中，我們會使用ReentrantLock或Synchronized來應對并發場景，
比如最常見的賣票場景，假如總共有100張票，執行緒A和執行緒B同時操作，如下圖：

JMM記憶體模型

這時有一個共享變數100，執行緒A和B將100拷貝到自己的作業記憶體中，當執行緒A搶到執行權的時候，此時A作業記憶體中的值是100，然后售票，進行自減操作，將自己作業記憶體中的值變成了99，當A還沒來得及將99刷回到主記憶體的時候，執行緒B進來了，此時B拿到的主記憶體的值還是100，然后售票，進行自減，也是99，這就出現了同一張票出售了兩次的情況，所以我們會加鎖加volatile保證原子性保證可見性，

2、分布式鎖是什么？
上面的場景中，我們可以通過ReentrantLock或者Synchronized搞定，因為你的專案只運行在一臺服務器上，只有一個JVM，所有的共享變數都加載到同一個主記憶體中，而分布式應用中，一個專案部署在多臺服務器上，最基本的架構如下圖：

最簡單的分布式架構

比如現在server1、server2和server3讀取到資料庫的票數都是100，在每一個server中，我們可以用JDK的鎖來保證多個用戶同時訪問我這臺server時不會出問題，但問題是，如果client1訪問到的是server1，票數是100，然后購票，還沒來得及將資料庫票數改為99，client2也開始訪問系統購票了，client2如果訪問的是server1，自然不會出問題，如果訪問的是server2，這時server2讀取到資料庫的票數還是100，那么就出問題了，又出現了同一張票賣了兩次的情況，在分布式應用中，JDK的鎖機制就無法滿足需求了，所以就出現了分布式鎖，

3、分布式鎖應該滿足的條件：

四個一：同一個方法在同一時刻只能被一臺機器的一個執行緒執行
三個具備：具備可重入特性；具備鎖失效機制，防止死鎖；具備非阻塞鎖特性，即沒獲取到鎖回傳獲取鎖失敗，而不是一直等待
兩個高：高性能地獲取與釋放鎖；高可用的獲取與釋放鎖

4、分布式鎖的實作方式：

基于資料庫：用資料庫的排他鎖實作
基于redis：利用redis的set key value NX EX 30000；也可以用redis的第三方庫比如Redisson
基于zookeeper：利用zookeeper的臨時順序節點實作；也可以用zookeeper的第三方庫比如Curator

二、基于資料庫實作

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 '資料被操作的時間',
	PRIMARY KEY (`dl_id`),
	UNIQUE KEY `uq_method_name` (`dl_method_name`) USING BTREE
) ENGINE = INNODB DEFAULT charset = utf8 COMMENT = '分布式鎖表';

2、思路：
當執行一個方法的時候，我們首先嘗試往表中插入一條資料，如果插入成功，則占鎖成功，繼續往下執行，執行完洗掉該記錄，如果插入失敗，我們再以當前方法名、當前機器ip+執行緒id、資料被操作時間為5分鐘內(5分鐘表示鎖失效的時間)為條件去查詢，如果有記錄，表示該機器的該執行緒在5分鐘內占有過鎖了，直接往下執行最后洗掉記錄；如果沒有記錄，占有鎖失敗，
一個用戶就是一個執行緒，所以我們可以把機器ip和用戶id組合一起當成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、小總結：
以上表結構可能并不是很好，只是提供了這么一個思路，下面說它的優缺點：

優點：成本低，不需要引入其他的技術
缺點：對資料庫依賴性強，如果資料庫掛了，那就涼涼了，所以資料庫最好也是高可用的

三、基于redis實作

1、原理：
基于redis的set key value nx ex 30，這條陳述句的意思就是如果key不存在就設定，并且過期時間為30s，如果key已經存在就會回傳false，如果要以毫秒為單位，把ex換成px就好了，我們執行方法前，先將方法名當成key，執行這條陳述句，如果執行成功就是獲取鎖成功，執行失敗就是獲取鎖失敗，

2、代碼實作：

RedisUtil的部分代碼：

/**
* key不存在時就設定，回傳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);
}

業務方法中使用：

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

3、小總結：

優點：簡單易用，一條redis命令就搞定，可以設定過期時間，避免釋放鎖失敗造成其他執行緒長時間無法獲取鎖的問題，
缺點：這種做法只適合redis是單機的時候，如果redis有集群，這樣做就會出問題，假如一個執行緒在master上獲取鎖成功了，在master還沒來得及將資料同步到slave上的時候，master掛了，slave升級為master，第二個執行緒進來嘗試獲取鎖，因為新的master上并沒有這個key，所以，也能成功獲取到鎖，
解決辦法：針對上面的缺點，我們可以采用redis的RedLock演算法，假如集群中有n個redis，我們先從這n個redis中嘗試獲取鎖（鎖的過期時間為x），并記錄獲取鎖的消耗的總時間t，獲取鎖成功數量為s，當且僅當t < x 并且 s >= (n/2 + 1)時，認為獲取鎖成功，

四、基于Redisson實作

1、是什么？
官網地址：https://github.com/redisson/redisson/wiki/Table-of-Content
Redisson是一個功能十分強大的redis客戶端，封裝了很多分布式操作，比如分布式物件、分布式集合、分布式鎖等，它的分布式鎖也很多，什么公平鎖、可重入鎖、redlock等一應俱全，下面來看看如何在springboot專案中使用它，

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單機版的寫法
    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，然后用它獲取鎖，得到鎖之后就可以進行占鎖和釋放鎖了，有阻塞式鎖，也有非阻塞式鎖，具體用法如下：

@Autowired
private RedissonClient redisson;

/**
 * 未設定過期時間，沒獲取到就會一直阻塞著
 * @return
 */
@GetMapping("/testLock")
public String testLock() {
	log.info("進入testLock方法，開始獲取鎖");
	String key = "testLock";
	RLock lock = redisson.getLock(key);
	lock.lock();
	log.info("獲取鎖成功，開始執行業務邏輯……");
	try {TimeUnit.SECONDS.sleep(10L); } catch (Exception e) { e.printStackTrace();};
	log.info("執行完業務邏輯，釋放鎖");
	lock.unlock();
	return "success";
}
	
/**
 * 嘗試獲取鎖，沒獲取到就直接失敗，不會阻塞
 * @return
 */
@GetMapping("/testTryLock")
public String testTryLock() {
	log.info("進入testTryLock方法，開始獲取鎖");
	String key = "testTryLock";
	RLock lock = redisson.getLock(key);
	boolean res = lock.tryLock();
	if (!res) {
		log.error("嘗試獲取鎖失敗");
		return "fail";
	} else {
		log.info("獲取鎖成功，開始執行業務邏輯……");
		try {TimeUnit.SECONDS.sleep(30L); } catch (Exception e) { e.printStackTrace();};
		log.info("執行完業務邏輯，釋放鎖");
		lock.unlock();
		return "success";
	}
}
	
/**
 * 鎖設定了過期時間，即使最后面的unlock失敗，20秒后也會自動釋放鎖
 * @return
 */
@GetMapping("/testLockTimeout")
public String testLockTimeout() {
	log.info("進入testLockTimeout方法，開始獲取鎖");
	String key = "testLockTimeout";
	RLock lock = redisson.getLock(key);
	// 20秒后自動釋放鎖
	lock.lock(20, TimeUnit.SECONDS);
	log.info("獲取鎖成功，開始執行業務邏輯……");
	try {TimeUnit.SECONDS.sleep(10L); } catch (Exception e) { e.printStackTrace();};
	lock.unlock();
	return "success";
}
	
/**
 * 嘗試獲取鎖，15秒還沒獲取到就獲取鎖失敗；獲取到了會持有20秒，20秒后自動釋放鎖
 * @return
 */
@GetMapping("/testTryLockTimeout")
public String testTryLockTimeout() {
	log.info("進入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("獲取鎖成功，開始執行業務邏輯……");
		try {TimeUnit.SECONDS.sleep(10L); } catch (Exception e) { e.printStackTrace();};
		log.info("執行完業務邏輯，釋放鎖");
		lock.unlock();
		return "success";
	}
}

3、小總結：
以上就是使用redisson做分布式鎖的簡單demo，用起來十分的方便，上面是與springboot專案集成，直接用它提供的springboot的starter就好了，用它來做分布式鎖的更多用法請移步至官網：redisson分布式鎖，

五、基于zookeeper實作

1、zookeeper知識點回顧：

zookeeper有四種型別的節點：

持久節點：默認的節點型別，客戶端與zookeeper斷開連接后，節點依然存在
持久順序節點：首先是持久節點，順序的意思是，zookeeper會根據節點創建的順序編號
臨時節點：客戶端與zookeeper斷開連接后節點不復存在
臨時順序節點：客戶端與zookeeper斷開連接后節點不復存在，zookeeper會根據節點創建的順序編號

2、基于zookeeper實作分布式鎖的原理：
我們正是利用了zookeeper的臨時順序節點來實作分布式鎖，首先我們創建一個名為lock(節點名稱隨意)的持久節點，執行緒1獲取鎖時，就在lock下面創建一個名為lock1的臨時順序節點，然后查找lock下所有的節點，判斷自己的lock1是不是第一個，如果是，獲取鎖成功，繼續執行業務邏輯，執行完后洗掉lock1節點；如果不是第一個，獲取鎖失敗，就watch排在自己前面一位的節點，當排在自己前一位的節點被干掉時，再檢查自己是不是排第一了，如果是，獲取鎖成功，圖解程序如下：

zookeeper分布式鎖原理

執行緒1創建了一個lock1，發現lock1的第一個節點，占鎖成功；在執行緒1還沒釋放鎖的時候，執行緒2來了，創建了一個lock2，發現lock2不是第一個，便監控lock1，執行緒3此時進行就監控lock2，直到自己是第一個節點時才占鎖成功，假如某個執行緒釋放鎖的時候zookeeper崩了也沒關系，因為是臨時節點，斷開連接節點就沒了，其他執行緒還是可以正常獲取鎖，這就是要用臨時節點的原因，

說清楚了原理，用代碼實作也就不難了，可以引入zookeeper的客戶端zkClient，自己寫代碼實作(偷個懶，自己就不寫了，有興趣的可以參考我zookeeper的文章，肯定可以自己寫出來的)，不過有非常優秀的開源解決方案比如curator，下面就看看curator怎么用，

六、基于curator實作

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是沒有集成的，也就是說寫的時候不會有提示

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

配置類：讀取application.yml中的屬性，創建CuratorFramework實體

@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));
	}
}

測驗類：測驗整合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成功啟動了的情況下，執行這個單元測驗，最后回到linux中，用zkCli.sh連接，查看是否成功創建節點，

2、使用Curator做分布式鎖：
Curator封裝了很多鎖，比如可重入共享鎖、不可重入共享鎖、可重入讀寫鎖、聯鎖等，具體可以參考官網：curator分布式鎖的用法，

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

@Component
@Slf4j
public class ZookeeperUtil {

	private static CuratorFramework curatorFramework;
	
	private static InterProcessLock lock;

	/** 持久節點 */
	private final static String ROOT_PATH = "/lock/";
	
	/** 可重入共享鎖 */
	private static InterProcessMutex interProcessMutex;
	/** 不可重入共享鎖 */
	private static InterProcessSemaphoreMutex interProcessSemaphoreMutex;
	/** 可重入讀寫鎖 */
	private static InterProcessReadWriteLock interProcessReadWriteLock;
	/** 多共享鎖(將多把鎖當成一把來用) */
	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);
	}

	/**
	 * 獲取聯鎖(多把鎖當成一把來用)
	 * @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);
		}
	}
	
	/**
	 * 釋放聯鎖
	 */
	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：寫一個介面，用Curator加鎖，然后用瀏覽器進行訪問

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

打開一個瀏覽器視窗訪問，后臺列印出獲取鎖成功的日志，在1分鐘之內，開啟另一個視窗再次訪問，列印出獲取鎖失敗的日志，說明分布式鎖生效了，

七、實作分布式鎖的各方案比較

基于資料庫實作最簡單，不需要引入第三方應用，但是因為每次加鎖和解鎖都要進行IO操作，性能不是很好，
基于redis實作比較均衡，性能很好，也不是很難，比較可靠，
基于zookeeper實作難度較大，因為需要維護一個zookeeper集群，如果專案原本沒有用到zookeeper，還是用redis比較好，

本文專案地址：分布式鎖

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/397598.html

標籤：其他

上一篇：資料庫-ElasticSearch學習筆記

下一篇：Spark SQL functions.scala 原始碼決議（七）String functions （基于 Spark 3.3.0）