前言
本文力爭以最簡單的語言，以博主自己對分布式鎖的理解，按照自己的語言來描述分布式鎖的概念、作用、原理、實現。如有錯誤，還請各位大佬海涵，懇請指正。分布式鎖分兩篇來講解，本篇講解客戶端，下一篇講解redis服務端。
概念
如果把分布式鎖的概念搬到這里，博主也會覺得枯燥。博主這里以舉例的形式來描繪它。
試想一種場景，在一個偏遠小鎮上的火車站，只有一個售票窗口。
火車站來了10名旅客，前往售票窗口購買火車票，旅客只能排隊購票，排到第一的旅客，可以與售票員溝通，買票。
好啦，以上就是一個分布式鎖的場景，我們來分析一下每一個細節。
每位旅客可以理解為一個系統或者線程。他們在競爭售票員的工作時間。
是不是覺得分布式鎖也不是什么高大上的概念。有同學會問，鎖到底在哪里呢？還是買票場景，我們看看鎖長什么樣子。
我們深入想一下，這10位旅客本來是並行的（沒有買票前，他們有的在吃飯，有的在玩手機，等等等），而到了買票的時候，就必須排隊（串行），而不是一起買票。
沒錯，就是在特定的場景下，將並行的場景，變成串行，就是分布式鎖的奧義所在。
作用
分布式鎖的作用不但非常大，而且非常多。
在軟件設計中，比如電商秒殺活動。商家預備了1000件貨物，也就只有這1000件貨，有1500人參與秒殺，可以理解為1500個線程來排隊購買商品。那就必須將這1500個線程排個隊（比如按照時間），設置一把鎖，一個購買過程結束，再開始下一個。
為什么redis可以實現分布式鎖呢？
我們以購票舉例，購票窗口前的這個鎖，是每位旅客都可以看到的。
這里我們可以得出一個結論，一把鎖首先要具有的屬性是：想要獲得鎖的人都可以看到。
這把鎖既不能屬於服務器A，也不能屬於服務器B，因為他們都不知道另一方的存在，那就必須選擇一個公信的第三方來作為鎖。當當當~ redis閃亮登場。當然zookeeper也可以實現，這里先挖一個坑，以后再填zookeeper吧。
原理
加鎖的基本思路
redis中有一條指令非常有意思，它叫做setnx
當redis中不存在key值為“lock”的時候，可以設置成功；當存在key值時，設置失敗。
這句指令，好比是，詢問一下，到我買票了嗎？返回結果是1的時候，到您買票了；返回結果是0的時候，還沒到您，稍后再詢問。
我們的鎖過程可以這樣來操作：
- setnx lock 鎖值
- 處理業務邏輯
- 釋放鎖 del lock
優化一
為什么要優化？
試想，如果setnx lock 1 加鎖成功，這個時候系統因為其他原因，掛掉了，就永遠無法執行del lock了。
要避免這種情況，怎么辦呢？給鎖一個過期時間。
這樣無論系統是否宕機，都會在10秒后釋放鎖。看似很美好，雖然setnx lock 1 與 expire lock 10之間的時間間隙非常小，但仍然有風險，加入系統執行完 setnx lock 1 后，宕機了，並沒有執行 過期指令 expire lock 10，再次產生了一把無法解開的鎖，“死鎖”。
這時候引入了一個概念，叫做原子操作。即這兩條指令需要在一個原子操作內執行完成。
set key value [expiration EX seconds|PX milliseconds] [NX|XX]
優化二
why?上一個優化已經把上鎖過程做成了原子操作，還需要什么優化呢？
當然有，試想一下，之前代碼set lock 1 ex 10 nx，設置過期時間是10秒，那么這個10秒是否可靠呢？顯然不可靠。
我們加鎖的過程是 加鎖---執行業務代碼---釋放鎖
加入業務代碼的執行時間超過10秒呢？是不是業務代碼還沒有執行完，鎖就已經釋放了。放在購票場景中，第一位旅客還沒有完成購票，第二位旅客就開始購票。顯然不合理。怎么辦呢？
這里我們需要估計業務代碼的執行時間，加入預估出來的時間是10秒，可以在業務代碼中開辟一個“續命”的操作。
- 加鎖 set lock 1 ex 10 nx
- 每過3秒，把該鎖的時間重新設置為 10秒
- 執行業務代碼
- 釋放鎖 del lock
這里的續命時間間隔 = 過期時間 10S / 3
這樣設置比較合理，可以防止一次續命失敗。
優化三
納尼?還有問題嗎？
有，而且可以算是一個bug，我們一直在用 set lock 1 ex 10 nx 來加鎖，用del lock 來釋放鎖。
我們需要明確知道，釋放的鎖，是自己加上的。
可以set lock uuid ex 10 nx 來解決該問題。
拓展-可重入鎖
一個線程獲取到鎖以后，再次獲取鎖，就是可重入鎖。
但博主現在遇到的問題，一般不需要可重入鎖即可解決。java中ReentrantLock就是可重入鎖。
可重入鎖，對代碼的復雜度增加了很多，玩不好，容易扯襠。謹慎使用。
實現
已經講了很多優化相關的內容，這里博主就直接寫優化后的代碼了。
博主使用java來實現。而redis官方（https://redis.io/clients#java）推薦的有三個框架。分別是Jedis、lettuce、Redisson。
由於博主在本篇中主要討論單個redis的情況，而redisson主要用來處理分布式redis，下一篇博文使用redisson，敬請期待。
springboot2.x 默認采用了 lettuce，所以博主就使用lettuce來實現分布式鎖。
引入依賴
<!-- data-redis中集成了lettuce -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!-- redis鏈接池 -->
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
</dependency>
<!-- alibaba json -->
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.72</version>
</dependency>
配置文件
既然要測試分布式鎖，那么就至少應該跑兩份代碼，所以配置文件也應該是兩份，這里博主偷個懶，提供一份配置文件，另一份配置文件修改下server的端口即可。
server:
port: 80
spring:
redis:
# redis的ip地址
host: redis的ip地址
# redis的端口號
port: 6379
# redis的密碼
password: 你的密碼
lettuce:
pool:
# 最大鏈接數
max-active: 30
# 鏈接池中最大空閑鏈接數
max-idle: 15
# 最大阻塞等待鏈接時長 默認不限制 -1
max-wait: 2000
# 最小空閑鏈接數
min-idle: 10
# 鏈接超時時長
shutdown-timeout: 10000
lettuce配置類
這個類博主就不細講了，springboot整合lettuce，序列化博主更偏愛FastJson
import com.alibaba.fastjson.support.spring.GenericFastJsonRedisSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
/**
* @author xujp
* redis 配置類 將RedisTemplate交給spring托管
*/
@Configuration
public class RedisConfig {
@Bean
public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory){
RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<>();
redisTemplate.setConnectionFactory(redisConnectionFactory);
StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
GenericFastJsonRedisSerializer genericFastJsonRedisSerializer = new GenericFastJsonRedisSerializer();
redisTemplate.setKeySerializer(stringRedisSerializer);
redisTemplate.setValueSerializer(genericFastJsonRedisSerializer);
redisTemplate.setHashKeySerializer(stringRedisSerializer);
redisTemplate.setHashValueSerializer(genericFastJsonRedisSerializer);
redisTemplate.afterPropertiesSet();
return redisTemplate;
}
}
分布式鎖
重頭戲來了，手寫分布式鎖的核心代碼示例。
import com.redis.demo1.thread.WatchDog;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
/**
* @author xujp
*/
@RestController
@RequestMapping("/test")
public class TestController {
@Autowired
private RedisTemplate redisTemplate;
@GetMapping
public void lock(){
String uuid = UUID.randomUUID().toString();
//System.out.println(uuid);
WatchDog watchDog;
try {
// 自旋
while (true) {
// 嘗試獲取鎖
Boolean hasLock = redisTemplate.opsForValue().setIfAbsent("lock", uuid, 3l, TimeUnit.SECONDS);
if(hasLock) {
// 看門狗“續命“
watchDog = new WatchDog(redisTemplate, uuid);
watchDog.start();
// 業務邏輯start
int num = (int) redisTemplate.opsForValue().get("num");
//Thread.sleep(4000); // 假設業務需要4s處理時間
redisTemplate.opsForValue().set("num", num - 1);
System.out.println(num);
// 業務邏輯處理 end
break;
}else{
// 睡眠100ms再自旋
Thread.sleep(100);
}
}
}catch (Exception e){
System.out.println(e);
}finally {
// 關閉鎖
String l = (String) redisTemplate.opsForValue().get("lock");
if (l.equalsIgnoreCase(uuid)) {
redisTemplate.delete("lock");
}
}
}
}
分布式鎖“續命”代碼示例
import org.springframework.data.redis.core.RedisTemplate;
import java.util.concurrent.TimeUnit;
/**
* @author xujp
*/
public class WatchDog extends Thread {
private RedisTemplate redisTemplate;
private String uuid;
public WatchDog(RedisTemplate redisTemplate, String uuid){
this.redisTemplate = redisTemplate;
this.uuid = uuid;
}
public void run(){
// 續命邏輯
while (true){
try {
// 獲取鎖的value
Object redisUUID = redisTemplate.opsForValue().get("lock");
// 判斷當前父線程是否已經釋放鎖，如果父線程已釋放，則跳出線程
if(redisUUID==null || !redisUUID.toString().equals(uuid)){
break;
}
// 續命
redisTemplate.expire("lock", 3l, TimeUnit.SECONDS);
// 沒隔1s續命一次
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
break;
}
}
}
}
測試
首先我們將代碼分別以80和81端口run起來。
有精力的同學，還可以再搭建一個nginx將請求分流到80和81。這里博主簡單粗暴地使用jmeter請求。
博主使用jmeter來測試，博主默認大家都會使用（不會使用的童鞋需要學習嘍）。
jmeter准備工作
在jmeter中設置50個線程
在該線程下設置兩個接口，分別請求80和81
redis准備工作
在redis中設置一對鍵值 num
至此，就可以在jmeter中開啟請求了
測試結果
我們先來看redis中num的值
我們再分別查看80和81的日志
總結
本文講述了利用redis實現分布式鎖的原理，分布式鎖本質上是將並發請求按順序處理，那么這把鎖就成為了所有請求的瓶頸，如何打破鎖的瓶頸呢？敬請關注博主，后續填坑（博主挖坑必填）。
本文留下的兩個坑：
1，zookeeper分布式鎖？
2，分布式鎖實現了並發排隊，鎖成為了性能瓶頸，如何提高性能？