源?/?? ? ? ??文/?

概述

提到分布式鎖大家都會(huì)想到如下兩種：

• 基于Redisson組件，使用redlock算法實(shí)現(xiàn)
• 基于Apache Curator，利用Zookeeper的臨時(shí)順序節(jié)點(diǎn)模型實(shí)現(xiàn)

今天我們來(lái)說(shuō)說(shuō)第三種，使用?Spring Integration?實(shí)現(xiàn)，也是我個(gè)人比較推薦的一種。

Spring Integration在基于Spring的應(yīng)用程序中實(shí)現(xiàn)輕量級(jí)消息傳遞，并支持通過(guò)聲明適配器與外部系統(tǒng)集成。Spring Integration的主要目標(biāo)是提供一個(gè)簡(jiǎn)單的模型來(lái)構(gòu)建企業(yè)集成解決方案，同時(shí)保持關(guān)注點(diǎn)的分離，這對(duì)于生成可維護(hù)，可測(cè)試的代碼至關(guān)重要。我們熟知的 Spring Cloud Stream的底層就是Spring Integration。

官方地址：https://github.com/spring-projects/spring-integration

Spring Integration提供的全局鎖目前為如下存儲(chǔ)提供了實(shí)現(xiàn)：

• Gemfire
• JDBC
• Redis
• Zookeeper

它們使用相同的API抽象，這意味著，不論使用哪種存儲(chǔ)，你的編碼體驗(yàn)是一樣的。試想一下你目前是基于zookeeper實(shí)現(xiàn)的分布式鎖，哪天你想換成redis的實(shí)現(xiàn)，我們只需要修改相關(guān)依賴和配置就可以了，無(wú)需修改代碼。下面是你使用?Spring Integration?實(shí)現(xiàn)分布式鎖時(shí)需要關(guān)注的方法：

方法名	描述
lock()	Acquires the lock.?加鎖，如果已經(jīng)被其他線程鎖住或者當(dāng)前線程不能獲取鎖則阻塞
lockInterruptibly()	Acquires the lock unless the current thread is interrupted.?加鎖，除非當(dāng)前線程被打斷。
tryLock()	Acquires the lock only if it is free at the time of invocation.?嘗試加鎖，如果已經(jīng)有其他鎖鎖住，獲取當(dāng)前線程不能加鎖，則返回false，加鎖失敗；加鎖成功則返回true
tryLock(long time, TimeUnit unit)	Acquires the lock if it is free within the given waiting time and the current thread has not been interrupted.?嘗試在指定時(shí)間內(nèi)加鎖，如果已經(jīng)有其他鎖鎖住，獲取當(dāng)前線程不能加鎖，則返回false，加鎖失敗；加鎖成功則返回true
unlock()	Releases the lock.?解鎖

實(shí)戰(zhàn)

話不多說(shuō)，我們看看使用?Spring Integration?如何基于redis和zookeeper快速實(shí)現(xiàn)分布式鎖，至于Gemfire 和 Jdbc的實(shí)現(xiàn)大家自行實(shí)踐。

基于Redis實(shí)現(xiàn)

• 引入相關(guān)組件

<dependency>
?<groupId>org.springframework.bootgroupId>
?<artifactId>spring-boot-starter-integrationartifactId>
dependency>

<dependency>
?<groupId>org.springframework.integrationgroupId>
?<artifactId>spring-integration-redisartifactId>
dependency>

<dependency>
?<groupId>org.springframework.bootgroupId>
?<artifactId>spring-boot-starter-data-redisartifactId>
dependency>

• 在application.yml中添加redis的配置

spring:
?redis:
??host:?172.31.0.149
??port:?7111

• 建立配置類(lèi)，注入RedisLockRegistry

@Configuration
public?class?RedisLockConfiguration?{

????@Bean
????public?RedisLockRegistry?redisLockRegistry(RedisConnectionFactory?redisConnectionFactory){
????????return?new?RedisLockRegistry(redisConnectionFactory,?"redis-lock");
????}

}

• 編寫(xiě)測(cè)試代碼

@RestController
@RequestMapping("lock")
@Log4j2
public?class?DistributedLockController?{
????@Autowired
????private?RedisLockRegistry?redisLockRegistry;

????@GetMapping("/redis")
????public?void?test1()?{
????????Lock?lock?=?redisLockRegistry.obtain("redis");
????????try{
????????????//嘗試在指定時(shí)間內(nèi)加鎖，如果已經(jīng)有其他鎖鎖住，獲取當(dāng)前線程不能加鎖，則返回false，加鎖失敗；加鎖成功則返回true
????????????if(lock.tryLock(3,?TimeUnit.SECONDS)){
????????????????log.info("lock?is?ready");
????????????????TimeUnit.SECONDS.sleep(5);
????????????}
????????}?catch?(InterruptedException?e)?{
????????????log.error("obtain?lock?error",e);
????????}?finally?{
????????????lock.unlock();
????????}
????}
}

• 測(cè)試
  啟動(dòng)多個(gè)實(shí)例，分別訪問(wèn)/lock/redis?端點(diǎn)，一個(gè)正常秩序業(yè)務(wù)邏輯，另外一個(gè)實(shí)例訪問(wèn)出現(xiàn)如下錯(cuò)誤說(shuō)明第二個(gè)實(shí)例沒(méi)有拿到鎖，證明了分布式鎖的存在。

注意，如果使用新版Springboot進(jìn)行集成時(shí)需要使用Redis4版本，否則會(huì)出現(xiàn)下面的異常告警，主要是?unlock()?釋放鎖時(shí)使用了UNLINK命令，這個(gè)需要Redis4版本才能支持。

2020-05-14?11:30:24,781?WARN??RedisLockRegistry:339?-?The?UNLINK?command?has?failed?(not?supported?on?the?Redis?server?);?falling?back?to?the?regular?DELETE?command
org.springframework.data.redis.RedisSystemException:?Error?in?execution;?nested?exception?is?io.lettuce.core.RedisCommandExecutionException:?ERR?unknown?command?'UNLINK'

基于Zookeeper實(shí)現(xiàn)

• 引入組件

<dependency>
?<groupId>org.springframework.bootgroupId>
?<artifactId>spring-boot-starter-integrationartifactId>
dependency>

?<dependency>
?<groupId>org.springframework.integrationgroupId>
?<artifactId>spring-integration-zookeeperartifactId>
dependency>

• 在application.yml中添加zookeeper的配置

zookeeper:??
????host:?172.31.0.43:2181

• 建立配置類(lèi)，注入ZookeeperLockRegistry

@Configuration
public?class?ZookeeperLockConfiguration?{
????@Value("${zookeeper.host}")
????private?String?zkUrl;


????@Bean
????public?CuratorFrameworkFactoryBean?curatorFrameworkFactoryBean(){
????????return?new?CuratorFrameworkFactoryBean(zkUrl);
????}

????@Bean
????public?ZookeeperLockRegistry?zookeeperLockRegistry(CuratorFramework?curatorFramework){
????????return?new?ZookeeperLockRegistry(curatorFramework,"/zookeeper-lock");
????}
}

• 編寫(xiě)測(cè)試代碼

@RestController
@RequestMapping("lock")
@Log4j2
public?class?DistributedLockController?{

????@Autowired
????private?ZookeeperLockRegistry?zookeeperLockRegistry;

????@GetMapping("/zookeeper")
????public?void?test2()?{
????????Lock?lock?=?zookeeperLockRegistry.obtain("zookeeper");
????????try{
????????????//嘗試在指定時(shí)間內(nèi)加鎖，如果已經(jīng)有其他鎖鎖住，獲取當(dāng)前線程不能加鎖，則返回false，加鎖失敗；加鎖成功則返回true
????????????if(lock.tryLock(3,?TimeUnit.SECONDS)){
????????????????log.info("lock?is?ready");
????????????????TimeUnit.SECONDS.sleep(5);
????????????}
????????}?catch?(InterruptedException?e)?{
????????????log.error("obtain?lock?error",e);
????????}?finally?{
????????????lock.unlock();
????????}
????}
}

• 測(cè)試
  啟動(dòng)多個(gè)實(shí)例，分別訪問(wèn)/lock/zookeeper?端點(diǎn)，一個(gè)正常秩序業(yè)務(wù)邏輯，另外一個(gè)實(shí)例訪問(wèn)出現(xiàn)如下錯(cuò)誤說(shuō)明第二個(gè)實(shí)例沒(méi)有拿到鎖，證明了分布式鎖的存在。

END

頂級(jí)程序員：topcoding

做最好的程序員社區(qū)：Java后端開(kāi)發(fā)、Python、大數(shù)據(jù)、AI

一鍵三連「分享」、「點(diǎn)贊」和「在看」