使用Spring Integration实现分布式锁-电子发烧友网

概述

提到分布式锁大家都会想到如下两种：

基于Redisson组件，使用redlock算法实现
基于Apache Curator，利用Zookeeper的临时顺序节点模型实现

今天我们来说说第三种，使用Spring Integration实现。

Spring Integration在基于Spring的应用程序中实现轻量级消息传递，并支持通过声明适配器与外部系统集成。

Spring Integration的主要目标是提供一个简单的模型来构建企业集成解决方案，同时保持关注点的分离，这对于生成可维护，可测试的代码至关重要。我们熟知的 Spring Cloud Stream的底层就是Spring Integration。

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

Spring Integration提供的全局锁目前为如下存储提供了实现：

Gemfire
JDBC
Redis
Zookeeper

它们使用相同的API抽象，这意味着，不论使用哪种存储，你的编码体验是一样的。试想一下你目前是基于zookeeper实现的分布式锁，哪天你想换成redis的实现，我们只需要修改相关依赖和配置就可以了，无需修改代码。下面是你使用Spring Integration实现分布式锁时需要关注的方法：

方法名	描述
`lock()`	`Acquires the lock.`加锁，如果已经被其他线程锁住或者当前线程不能获取锁则阻塞
`lockInterruptibly()`	`Acquires the lock unless the current thread is interrupted.`加锁，除非当前线程被打断。
`tryLock()`	`Acquires the lock only if it is free at the time of invocation.`尝试加锁，如果已经有其他锁锁住，获取当前线程不能加锁，则返回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.`尝试在指定时间内加锁，如果已经有其他锁锁住，获取当前线程不能加锁，则返回false，加锁失败；加锁成功则返回true
`unlock()`	`Releases the lock.`解锁

实战

话不多说，我们看看使用Spring Integration如何基于redis和zookeeper快速实现分布式锁，至于Gemfire 和 Jdbc的实现大家自行实践。

基于Redis实现

引入相关组件


org.springframework.boot
spring-boot-starter-integration



org.springframework.integration
spring-integration-redis



org.springframework.boot
spring-boot-starter-data-redis

在application.yml中添加redis的配置

spring:
redis:
host:172.31.0.149
port:7111

建立配置类，注入RedisLockRegistry

@Configuration
publicclassRedisLockConfiguration{

@Bean
publicRedisLockRegistryredisLockRegistry(RedisConnectionFactoryredisConnectionFactory){
returnnewRedisLockRegistry(redisConnectionFactory,"redis-lock");
}

}

编写测试代码

@RestController
@RequestMapping("lock")
@Log4j2
publicclassDistributedLockController{
@Autowired
privateRedisLockRegistryredisLockRegistry;

@GetMapping("/redis")
publicvoidtest1(){
Locklock=redisLockRegistry.obtain("redis");
try{
//尝试在指定时间内加锁，如果已经有其他锁锁住，获取当前线程不能加锁，则返回false，加锁失败；加锁成功则返回true
if(lock.tryLock(3,TimeUnit.SECONDS)){
log.info("lockisready");
TimeUnit.SECONDS.sleep(5);
}
}catch(InterruptedExceptione){
log.error("obtainlockerror",e);
}finally{
lock.unlock();
}
}
}

测试
启动多个实例，分别访问/lock/redis端点，一个正常秩序业务逻辑，另外一个实例访问出现如下错误说明第二个实例没有拿到锁，证明了分布式锁的存在。

注意，如果使用新版Springboot进行集成时需要使用Redis4版本，否则会出现下面的异常告警，主要是unlock()释放锁时使用了UNLINK命令，这个需要Redis4版本才能支持。

2020-05-141124,781WARNRedisLockRegistry:339-TheUNLINKcommandhasfailed(notsupportedontheRedisserver?);fallingbacktotheregularDELETEcommand
org.springframework.data.redis.RedisSystemException:Errorinexecution;nestedexceptionisio.lettuce.core.RedisCommandExecutionException:ERRunknowncommand'UNLINK'

基于Zookeeper实现

引入组件


org.springframework.boot
spring-boot-starter-integration



org.springframework.integration
spring-integration-zookeeper

在application.yml中添加zookeeper的配置

zookeeper:
host:172.31.0.43:2181

建立配置类，注入ZookeeperLockRegistry

@Configuration
publicclassZookeeperLockConfiguration{
@Value("${zookeeper.host}")
privateStringzkUrl;


@Bean
publicCuratorFrameworkFactoryBeancuratorFrameworkFactoryBean(){
returnnewCuratorFrameworkFactoryBean(zkUrl);
}

@Bean
publicZookeeperLockRegistryzookeeperLockRegistry(CuratorFrameworkcuratorFramework){
returnnewZookeeperLockRegistry(curatorFramework,"/zookeeper-lock");
}
}

编写测试代码

@RestController
@RequestMapping("lock")
@Log4j2
publicclassDistributedLockController{

@Autowired
privateZookeeperLockRegistryzookeeperLockRegistry;

@GetMapping("/zookeeper")
publicvoidtest2(){
Locklock=zookeeperLockRegistry.obtain("zookeeper");
try{
//尝试在指定时间内加锁，如果已经有其他锁锁住，获取当前线程不能加锁，则返回false，加锁失败；加锁成功则返回true
if(lock.tryLock(3,TimeUnit.SECONDS)){
log.info("lockisready");
TimeUnit.SECONDS.sleep(5);
}
}catch(InterruptedExceptione){
log.error("obtainlockerror",e);
}finally{
lock.unlock();
}
}
}