如何优雅的实现分布式锁

落日映苍穹つ 2023-03-12 09:27 78阅读 0赞

![image.png][]

## 概述 ##

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

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

<table> 
 <thead> 
  <tr> 
   <th>方法名</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>lock()</code></td> 
   <td><code>Acquires the lock.</code> 加锁，如果已经被其他线程锁住或者当前线程不能获取锁则阻塞</td> 
  </tr> 
  <tr> 
   <td><code>lockInterruptibly()</code></td> 
   <td><code>Acquires the lock unless the current thread is interrupted.</code> 加锁，除非当前线程被打断。</td> 
  </tr> 
  <tr> 
   <td><code>tryLock()</code></td> 
   <td><code>Acquires the lock only if it is free at the time of invocation.</code> 尝试加锁，如果已经有其他锁锁住，获取当前线程不能加锁，则返回false，加锁失败；加锁成功则返回true</td> 
  </tr> 
  <tr> 
   <td><code>tryLock(long time, TimeUnit unit)</code></td> 
   <td><code>Acquires the lock if it is free within the given waiting time and the current thread has not been interrupted.</code> 尝试在指定时间内加锁，如果已经有其他锁锁住，获取当前线程不能加锁，则返回false，加锁失败；加锁成功则返回true</td> 
  </tr> 
  <tr> 
   <td><code>unlock()</code></td> 
   <td><code>Releases the lock.</code> 解锁</td> 
  </tr> 
 </tbody> 
</table>

## 实战 ##

话不多说，我们看看使用 `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`

@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` 端点，一个正常秩序业务逻辑，另外一个实例访问出现如下错误  
    ![image.png][image.png 1]  
    说明第二个实例没有拿到锁，证明了分布式锁的存在。

> 注意，如果使用新版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 command
    org.springframework.data.redis.RedisSystemException: Error in execution; nested exception is io.lettuce.core.RedisCommandExecutionException: ERR unknown command 'UNLINK'

### 基于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` 端点，一个正常秩序业务逻辑，另外一个实例访问出现如下错误  
    ![image.png][image.png 2]  
    说明第二个实例没有拿到锁，证明了分布式锁的存在。

[image.png]: /images/20230312/58f415391c254a7facd5a40119216431.png
[image.png 1]: /images/20230312/77abb079fac74579be1392624b505442.png
[image.png 2]: /images/20230312/663bf719643d41e0aff1eea28e3af949.png