使用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
public class RedisLockConfiguration {

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

}

• 编写测试代码

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

    @GetMapping("/redis")
    public void test1() {
        Lock lock = redisLockRegistry.obtain("redis");
        try{
            //尝试在指定时间内加锁，如果已经有其他锁锁住，获取当前线程不能加锁，则返回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();
        }
    }
}

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

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

2020-05-14 1124,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实现

• 引入组件

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

• 编写测试代码

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

    @Autowired
    private ZookeeperLockRegistry zookeeperLockRegistry;

    @GetMapping("/zookeeper")
    public void test2() {
        Lock lock = zookeeperLockRegistry.obtain("zookeeper");
        try{
            //尝试在指定时间内加锁，如果已经有其他锁锁住，获取当前线程不能加锁，则返回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();
        }
    }
}

测试
启动多个实例，分别访问/lock/zookeeper 端点，一个正常秩序业务逻辑，另外一个实例访问出现如下错误

 

说明第二个实例没有拿到锁，证明了分布式锁的存在。

 

 

 

原文标题：这样实现分布式锁，才叫优雅！

文章出处：【微信公众号：数据分析与开发】欢迎添加关注！文章转载请注明出处。

　　审核编辑：汤梓红