如何优雅的实现分布式锁

张坚

关注

发布于: 2020 年 05 月 15 日

概述

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

基于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 实现分布式锁时需要关注的方法：

实战

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

基于 Redis 实现

引入相关组件

<dependency>	<groupId>org.springframework.boot</groupId>	<artifactId>spring-boot-starter-integration</artifactId></dependency>
<dependency>	<groupId>org.springframework.integration</groupId>	<artifactId>spring-integration-redis</artifactId></dependency>
<dependency>	<groupId>org.springframework.boot</groupId>	<artifactId>spring-boot-starter-data-redis</artifactId></dependency>

复制代码

在 application.yml 中添加 redis 的配置

spring:	redis:		host: 172.31.0.149		port: 7111

复制代码

建立配置类，注入RedisLockRegistry

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

复制代码

编写测试代码

@RestController@RequestMapping("lock")@Log4j2public 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 11:30:24,781 WARN  RedisLockRegistry:339 - The UNLINK command has failed (not supported on the Redis server?); falling back to the regular DELETE commandorg.springframework.data.redis.RedisSystemException: Error in execution; nested exception is io.lettuce.core.RedisCommandExecutionException: ERR unknown command 'UNLINK'

复制代码

基于 Zookeeper 实现

引入组件

<dependency>	<groupId>org.springframework.boot</groupId>	<artifactId>spring-boot-starter-integration</artifactId></dependency>
 <dependency>	<groupId>org.springframework.integration</groupId>	<artifactId>spring-integration-zookeeper</artifactId></dependency>

复制代码

在 application.yml 中添加 zookeeper 的配置

zookeeper:      host: 172.31.0.43:2181

复制代码

建立配置类，注入ZookeeperLockRegistry

@Configurationpublic 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")@Log4j2public 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();        }    }}