我們將分布式鎖基于緩存擴展了一版，也就是說本starter即有分布式緩存功能，又有分布式鎖功能。而注解版的分布式鎖能夠解決大多數場景的并核問題，小粒度的Lock鎖方式補全其他場景。
1、為什么要使用分布式鎖？
在分布式，微服務環境中，我們的服務被拆分為很多個，并且每一個服務可能存在多個實例，部署在不同的服務器上。
此時JVM中的synchronized和lock鎖，將只能對自己所在服務的JVM加鎖，而跨機器，跨JMV的場景，仍然需要鎖的場景就需要使用到分布式鎖了。
2、為什么要使用redis實現分布式鎖？
因為Redis的性能很好，并且Redis是單線程的，天生線程安全。
并且Redis的key過期效果與Zookeeper的臨時節點的效果相似，都能實現鎖超時自動釋放的功能。
而且Redis還可以使用lua腳本來保證redis多條命令實現整體的原子性，Redisson就是使用lua腳本的原子性來實現分布式鎖的。
3、我們如何基于Redisson封裝分布式鎖？
1）、基于RedissonClient實現手動加鎖
2）、基于AOP+Redisson封裝注解版的分布式鎖
3）、將分布式鎖功能封裝成一個starter, 引入jar包即可實現分布式鎖
4、代碼實現
4.1、整合封裝Redisson
我們前面封裝了基于Redis擴展了SpringCache，封裝了
redis-cache-spring-boot-starter。
我們的分布式鎖基于這個模塊實現，下面引入依賴。
引入依賴
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.Apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>itdl-parent</artifactId>
<groupId>com.itdl</groupId>
<version>1.0</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>redis-lock-spring-boot-starter</artifactId>
<description>Redis實現分布式鎖的自定義starter封裝模塊</description>
<properties>
<maven.compiler.source>${JAVA.version}</maven.compiler.source>
<maven.compiler.target>${java.version}</maven.compiler.target>
</properties>
<dependencies>
<!--redis cache-->
<dependency>
<groupId>com.itdl</groupId>
<artifactId>redis-cache-spring-boot-starter</artifactId>
</dependency>
<!--redisson-->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
</dependency>
<!--aop-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-aop</artifactId>
</dependency>
</dependencies>
</project>
編寫RedisLockConfig配置RedissonClient
/**
* Redis實現分布式鎖的配置（使用Redisson）
*/
@Configuration // 標識為一個配置項，注入Spring容器
@AutoConfigureBefore({CustomRedisConfig.class, CacheNullValuesHandle.class})
@ConditionalOnProperty(value = "redis.enable", havingValue = "true") // 開啟redis.enable=true時生效
@Slf4j
public class RedisLockConfig {
private volatile boolean isCluster = false;
private volatile String redisHostsStr = "";
@Bean
@ConditionalOnMissingBean
public RedissonClient redissonClient(CustomRedisProperties redisProperties){
// 構建配置
Config config = buildConfig(redisProperties);
RedissonClient redissonClient = Redisson.create(config);
log.info("==============創建redisClient{}版成功：{}==================", isCluster ? "集群": "單機", redisHostsStr);
return redissonClient;
}
private Config buildConfig(CustomRedisProperties redisProperties) {
final Config config = new Config();
// 根據逗號切割host列表
Set<String> hosts = org.springframework.util.StringUtils.commaDelimitedListToSet(redisProperties.getHost());
if (CollectionUtils.isEmpty(hosts)){
throw new RuntimeException("redis host address cannot be empty");
}
// 只有一個host, 表示是單機host
if (hosts.size() == 1){
String hostPort = hosts.stream().findFirst().get();
redisHostsStr = "redis://" + hostPort.trim();
config.useSingleServer()
.setAddress(redisHostsStr)
.setDatabase(redisProperties.getDatabase())
.setPassword(StringUtils.isBlank(redisProperties.getPassword()) ? null : redisProperties.getPassword())
;
isCluster = false;
}else {
// 集群處理
String[] redisHosts = new String[hosts.size()];
int i = 0;
for (String host : hosts) {
String[] split = host.split(":");
if (split.length != 2){
throw new RuntimeException("host or port err");
}
redisHosts[i] = "redis://" + host.trim();
i++;
}
redisHostsStr = String.join(",", redisHosts);
// 配置集群
config.useClusterServers()
.addNodeAddress(redisHosts)
.setPassword(StringUtils.isBlank(redisProperties.getPassword()) ? null : redisProperties.getPassword())
// 解決Not all slots covered! Only 10922 slots are avAIlable
.setCheckSlotsCoverage(false);
isCluster = true;
}
return config;
}
}
我們配置時需要優先配置好redis-cache-spring-boot-starter，使用@AutoConfigureBefore({CustomRedisConfig.class, CacheNullValuesHandle.class})
直接使用，不再重復造輪子，然后我們根據自定義屬性配置文件CustomRedisProperties來創建RedissonClient的Bean。
編寫META-INF/spring.factories進行自動配置
org.springframework.boot.autoconfigure.EnableAutoConfiguration=
com.itdl.lock.config.RedisLockConfig
在測試模塊緩存service添加分布式鎖
@Cacheable(cacheNames = "demo2#3", key = "#id")
public TestEntity getById2(Long id){
// 創建分布式鎖
RLock lock = redissonClient.getLock("demo2_lock");
// 加鎖
lock.lock(10, TimeUnit.SECONDS);
if (id > 1000){
log.info("id={}沒有查詢到數據，返回空值", id);
return null;
}
TestEntity testEntity = new TestEntity(new Random().nextLong(), UUID.randomUUID().toString(), new Random().nextInt(20) + 10);
log.info("模擬查詢數據庫：{}", testEntity);
// 釋放鎖
lock.unlock();
return testEntity;
}
我們這里的@Cacheable沒有加sync=true, 此時并發請求會存在線程安全問題，但是我們在方法體局部添加了分布式鎖，因此我們的程序會按照順序執行。
如果我們的參數被定死了，最終請求會被先存儲到緩存，所以后續的查詢就會走緩存，這能很好的測試分布式鎖的效果。
編寫測試程序
@SpringBootTest
public class TestRedisLockRunner6 {
@Autowired
private MyTestService myTestService;
// 創建一個固定線程池
private ExecutorService executorService = Executors.newFixedThreadPool(16);
/**
* 多線程訪問請求，測試切面的線程安全性
*/
@Test
public void testMultiMyTestService() throws InterruptedException {
for (int i = 0; i < 100; i++) {
executorService.submit(() -> {
// 每次查詢同一個參數
TestEntity t1 = myTestService.getById2(1L);
});
}
// 主線程休息10秒種
Thread.sleep(10000);
}
}
我們可以看到，結果并沒有符合我們預期，但是又部分符合我們預期，為什么呢？
因為我們的@Cacheable是存在線程安全問題的，因為它先查詢緩存這個操作存在并發問題，查詢時就同時有N個請求進入@Cacheable, 并且都查詢沒有緩存。
然后同時執行方法體，但方法體加了分布式鎖，所以排隊進行處理，因此序號有序。
但打印數量不足總數，是因為這一批次沒有全部到達@Cacheable，而是執行完畢之后才將緩存回填，所以后續的請求就是走緩存了。
解決方案：我們加上sync=true之后就能實現，只查詢一次數據庫，就可以回填緩存了。如果我們去掉@Cacheable注解，則會每一次都查詢數據庫，但是時按照順序執行的。
加上sync=true測試
效果達到了我們的預期，繼續看一下去掉@Cacheable注解的情況。
去掉@Cacheable注解測試
我們的分布式鎖功能是沒有問題的，但是每次我們都需要執行getLock(), lock.lock(), lock.unlock()，是不是很麻煩，能不能一個注解搞定？
當然是可以的。
4.2、封裝注解版分布式鎖
編寫@RedisLock注解
/**
* 自定義Redis分布式鎖
*/
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RedisLock {
/**分布式鎖的名稱，支持el表達式*/
String lockName() default "";
/**鎖類型 默認為可重入鎖*/
LockType lockType() default REENTRANT_LOCK;
/**獲取鎖等待時間，默認60秒*/
long waitTime() default 30000L;
/** 鎖自動釋放時間，默認60秒*/
long leaseTime() default 60000L;
/**
* 被加鎖方法執行完是否立即釋放鎖
*/
boolean immediatelyUnLock() default true;
/** 時間單位, 默認毫秒*/
TimeUnit timeUnit() default TimeUnit.MILLISECONDS;
}
編寫分布式鎖切面RedisLockAop
/**
* Redis分布式鎖的切面邏輯實現
*/
@ConditionalOnProperty(value = "redis.enable", havingValue = "true") // 開啟redis.enable=true時生效
@AutoConfigureBefore(RedisLockConfig.class)
@Aspect
@Configuration
@Slf4j
public class RedisLockAop {
@Resource
private RedissonClient redissonClient;
/**
* 切點
*/
@Pointcut("@annotation(com.itdl.lock.anno.RedisLock)")
public void pointcut(){
}
/**
* 環繞通知 注解針對的是方法，這里切點也獲取方法進行處理就可以了
*/
@Around("pointcut()")
public Object around(ProceedingJoinPoint joinPoint) throws Throwable {
// 獲取方法
Method method = ((MethodSignature) joinPoint.getSignature()).getMethod();
// 獲取方法上的分布式鎖注解
RedisLock redisLock = method.getDeclaredAnnotation(RedisLock.class);
// 獲取注解的參數
// 鎖名稱
String lockName = redisLock.lockName();
// 鎖類型
LockType lockType = redisLock.lockType();
// 獲取RedissonClient的Lock
RLock lock = getRLock(lockName, lockType, redisLock);
//獲取到鎖后, 開始執行方法，執行完畢后釋放鎖
try {
log.debug("=========>>>獲取鎖成功, 即將執行業務邏輯：{}", lockName);
Object proceed = joinPoint.proceed();
// 釋放鎖
if (redisLock.immediatelyUnLock()) {
//是否立即釋放鎖
lock.unlock();
}
log.debug("=========>>>獲取鎖成功且執行業務邏輯成功：{}", lockName);
return proceed;
} catch (Exception e) {
log.error("=========>>>獲取鎖成功但執行業務邏輯失敗：{}", lockName);
e.printStackTrace();
throw new RedisLockException(LockErrCode.EXEC_BUSINESS_ERR);
}finally {
// 查詢當前線程是否保持此鎖定 被鎖定則解鎖
lock.unlock();
log.debug("=========>>>釋放鎖成功：{}", lockName);
}
}
/**
* 根據鎖名稱和類型創建鎖
* @param lockName 鎖名稱
* @param lockType 鎖類型
* @return 鎖
*/
private RLock getRLock(String lockName, LockType lockType, RedisLock redisLock) throws InterruptedException {
RLock lock;
switch (lockType){
case FAIR_LOCK:
lock = redissonClient.getFairLock(lockName);
break;
case READ_LOCK:
lock = redissonClient.getReadWriteLock(lockName).readLock();
break;
case WRITE_LOCK:
lock = redissonClient.getReadWriteLock(lockName).writeLock();
break;
default:
// 默認加可重入鎖，也就是普通的分布式鎖
lock = redissonClient.getLock(lockName);
break;
}
// 首先嘗試獲取鎖，如果在規定時間內沒有獲取到鎖，則調用lock等待鎖，直到獲取鎖為止
if (lock.tryLock()) {
lock.tryLock(redisLock.waitTime(), redisLock.leaseTime(), redisLock.timeUnit());
}else {
// 如果leaseTime>0，規定時間內獲取鎖，超時則自動釋放鎖
long leaseTime = redisLock.leaseTime();
if (leaseTime > 0) {
lock.lock(redisLock.leaseTime(), redisLock.timeUnit());
} else {
// 自動釋放鎖時間設置為0或者負數，則加鎖不設置超時時間
lock.lock();
}
}
return lock;
}
}
話不多說，封裝的邏輯已經在注釋中寫的很清晰了。
將切面也放入自動配置spring.factories中
org.springframework.boot.autoconfigure.EnableAutoConfiguration=
com.itdl.lock.config.RedisLockConfig,
com.itdl.lock.anno.RedisLockAop
測試注解版分布式鎖
@RedisLock(lockName = "demo4_lock")
public TestEntity getById4(Long id) throws InterruptedException {
index++;
log.info("current index is : {}", index);
Thread.sleep(new Random().nextInt(10) * 100);
TestEntity testEntity = new TestEntity(new Random().nextLong(), UUID.randomUUID().toString(), new Random().nextInt(20) + 10);
log.info("模擬查詢數據庫：{}", testEntity);
return testEntity;
}
可以看到，我們就是一個注解分布式鎖的效果，而分布式鎖與緩存注解通常不會一起使用，因為一般會在存在事務問題的地方我們會使用鎖，在多個JMV操作同一條數據做寫操作時需要加分布式鎖。
編寫測試程序
@SpringBootTest
public class TestRedisLockRunner6 {
@Autowired
private MyTestService myTestService;
// 創建一個固定線程池
private ExecutorService executorService = Executors.newFixedThreadPool(16);
/**
* 多線程訪問請求，測試切面的線程安全性
*/
@Test
public void testMultiMyTestService() throws InterruptedException {
for (int i = 0; i < 100; i++) {
executorService.submit(() -> {
try {
TestEntity t1 = myTestService.getById4(1L);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
// 主線程休息10秒種
Thread.sleep(60000);
}
}
測試結果
5、小結
我們將分布式鎖基于緩存擴展了一版，也就是說本starter即有分布式緩存功能，又有分布式鎖功能。
而注解版的分布式鎖能夠解決大多數場景的并核問題，小粒度的Lock鎖方式補全其他場景。
將兩者封裝成為一個starter，我們就可以很方便的使用分布式鎖功能，引入相關包即可，開箱即用。
