分布式锁的五种实现方案

淩亂°似流年 2022-11-21 11:29 237阅读 0赞

## 分布式核心问题系列目录 ##

[分布式核心问题 - SSO单点登录][- SSO]

[分布式核心问题 - 分布式锁][-]

[分布式核心问题 - 分布式接口幂等性设计][- 1]

[分布式核心问题 - MyCat实现读写分离][- MyCat]

**--------------------------------------------------------------------------------------------------------------------------**

## 1.为什么需要锁 ##

当我们要操作的一个资源存在中间态（资源在初始状态和结束状态之间还存在其他的状态），那我们在多线程下对此资源的访问就要小心了，因为状态的变迁需要时间，如果在此资源处于中间态的时，其他线程来获取此资源，那么获取到的结果则是这个资源的初始状态的值，这显然是不合理的。举个例子：

int a = 1;
    a++;

在java中++不是原子操作，因为此操作分为了三步，第一步先将a变量所在内存的值加载到寄存器，第二步将寄存器的值自增1，第三步将寄存器中的值写回内存，那么a变量就可以被称为是存在中间态的资源。假如我们启了两个线程，A线程执行到了a++的第二步将寄存器的值自增1但还没有写回内存，这时候B线程读到a变量的值还是1，这种现象在业务代码中是不可以发生的，所以针对**多线程访问具有中间态资源导致幻读的问题，我们需要用加锁的方式去解决**。

## 2.锁要实现的思路是什么 ##

锁要实现的思路为，对具有中间态资源的访问进行排它处理，也就是在a++这个非原子操作执行结束之前，禁止让其他线程获取变量a的值，在宏观上的体现为，所有任务必须排队有序的执行，这里的有序不单指顺序，还可以为竞争得到锁的顺序。多线程获取锁的步骤：

（1）通过竞争获取锁（排队）

（2）线程在对资源的操作中禁止其他线程对该资源的访问（占坑）

（3）其他线程阻塞或者异步尝试获取锁（等待）

（4）线程处理完任务释放锁，其他线程拿到锁（释放）

## 3.单体应用锁的局限性 ##

当项目部署方式为单体应用部署时，如果需要对某一个共享变量进行多线程同步访问，可以通过synchronize或者ReentranceLock实现。但如果我们的项目采用集群部署，前端请求采用Nginx转发的话，那么项目中业务方法上加的synchronize或者ReentranceLock就会失效，因为我们加的锁是JDK提供的锁，这种锁只能在一个JVM下起作用。这就是**单体应用锁的局限性，只能在一个JVM内加锁，无法跨JVM在整个应用层面去加锁。**

## 4.什么是分布式锁 ##

随着业务发展的需要，原单体单机部署的系统被演化成分布式集群系统后，由于分布式系统多线程、多进程且部署在不同的主机上的特点，将使原单机部署情况下的并发控制锁策略失效，单纯的Java API并不能提供分布式锁的能力。为了解决这个问题就需要一种跨JVM的互斥机制来控制共享资源的访问，这就是分布式锁。

## 5.分布式锁应该具备哪些条件 ##

我们要找到一个所有的JVM都能访问的第三方公共主键，并且其可以发出两个原子性操作的信号量，那么我们就可以把这两个信号量当做加锁和解锁的标识。分布式锁应该还具有以下条件：

（1）**原子性：**在分布式系统环境下，一个方法在同一时间只能被一个机器的一个线程执行；   
       （2）**高可用：**高可用的获取锁与释放锁，具备锁失效机制，防止死锁；   
       （3）**高性能：**高性能的获取锁与释放锁；   
       （4）**可重入性：**具备可重入特性；   
       （5）**阻塞性：**没有获得锁之前，其他任务既可以阻塞等待获取锁，也可以在没有获取到锁时直接返回获取锁失败。

**注：**上述列举的是作为分布式锁**应该**满足的五种条件，非必须满足。如高可用和高性能本身就是互斥的，在实现高可用的同时或多或少会对性能产生一定的影响，再比如要不要实现锁的可重入性和阻塞性是根据实际业务来确定的，所以分布式锁的设计还需灵活配置。

## 6.实现分布式锁的五种方案 ##

###        6.1 基于数据库实现分布式锁 ###

**              实现原理：**把数据库select检索出的数据用for update加锁，其他事务不能修改这条数据，也不能再给这条数据加锁，解锁使用comit提交当前事务。

**              加锁信号量：**select ...... for update

**              解锁信号量：**commit

**              实现流程：**

**sql：**

DROP TABLE IF EXISTS `distribute_lock`;
    CREATE TABLE `distribute_lock`  (
      `id` int(11) NOT NULL,
      `business_code` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,
      `business_name` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,
      PRIMARY KEY (`id`) USING BTREE
    ) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Compact;
    
    INSERT INTO `distribute_lock` VALUES (1, 'demo', 'demo演示');
    
    SET FOREIGN_KEY_CHECKS = 1;

**dao：**

DistributeLock selectDistributeLock(@Param("businessCode") String businessCode);

**mapper：**

<select id="selectDistributeLock" resultType="com.example.distributelock.DistributeLock">
        select 
            * 
        from 
            distribute_lock
        where 
            business_code = #{businessCode,jdbcType=VARCHAR}
        for update
    </select>

**controller:**

@RequestMapping("singleLock")
    @Transactional(rollbackFor = Exception.class)
    public void  singleLock() throws Exception {
        DistributeLock distributeLock = distributeLockMapper.selectDistributeLock("demo");
        if (distributeLock==null) {
            throw new Exception("分布式锁找不到");
        }
        try {
            //模拟业务代码执行
            Thread.sleep(20000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

上述代码都很简单，唯一要注意的就是controller层需要打上事务注解，如果不加的话查询sql会自动提交事务导致分布式锁失效。

###        6.2 基于redis实现分布式锁 ###

**              实现原理：**加锁时利用NX的原子性，多线程并发时，只有一个线程可以设置成功，从而获得锁。释放锁的时利用redis的delete命令配合LUA脚本实现，校验之前设置的随机数，相同才能释放，证明此时redis里面的值是当前线程设置的，避免释放了其他线程的锁

**              加锁信号量：**SET resource\_name my\_random\_value NX PX 30000

resource\_name：redis里面的key，资源的名称，根据不同的业务设置不同的锁

my\_random\_value：随机值，每个线程的随机值不同，用于释放锁时的校验

NX：key不存在的时候设置成功，key存在的时候设置不成功，此操作为原子操作

PX：自动失效时间，可以使锁过期失效，防止释放锁时出现异常，其他线程一直获取不到锁。

**              解锁信号量：**LUA脚本实现

**              实现流程：**

**引入redis依赖：**

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

**redis分布式锁工具类：**

public class RedisLock implements AutoCloseable {
    
        private RedisTemplate redisTemplate;
        private String key;
        private String value;
        private int expireTime;
    
        public RedisLock(RedisTemplate redisTemplate,String key,int expireTime){
            this.redisTemplate = redisTemplate;
            this.key = key;
            this.expireTime=expireTime;
            this.value = UUID.randomUUID().toString();
        }
        /**
         * 获取分布式锁
         */
        public boolean getLock(){
            RedisCallback<Boolean> redisCallback = connection -> {
                //设置NX
                RedisStringCommands.SetOption setOption = RedisStringCommands.SetOption.ifAbsent();
                //设置过期时间
                Expiration expiration = Expiration.seconds(expireTime);
                //序列化key
                byte[] redisKey = redisTemplate.getKeySerializer().serialize(key);
                //序列化value
                byte[] redisValue = redisTemplate.getValueSerializer().serialize(value);
                //执行setnx操作
                Boolean result = connection.set(redisKey, redisValue, expiration, setOption);
                return result;
            };
            //获取分布式锁
            Boolean lock = (Boolean)redisTemplate.execute(redisCallback);
            return lock;
        }
    
        /**
         * 解锁
         */
        public boolean unLock() {
            String script = "if redis.call(\"get\",KEYS[1]) == ARGV[1] then\n" +
                    "    return redis.call(\"del\",KEYS[1])\n" +
                    "else\n" +
                    "    return 0\n" +
                    "end";
            RedisScript<Boolean> redisScript = RedisScript.of(script,Boolean.class);
            List<String> keys = Arrays.asList(key);
            Boolean result = (Boolean)redisTemplate.execute(redisScript, keys, value);
            return result;
        }
        /**
         * 自动关闭
         */
        @Override
        public void close(){
            unLock();
        }
    }

**controller：**

@RequestMapping("redisLock")
    public String redisLock(){
        try (RedisLock redisLock = new RedisLock(redisTemplate,"redisKey",30)){
            if (redisLock.getLock()) {
                //延时模拟业务代码执行
                Thread.sleep(15000);
            }
        }catch (Exception e) {
            e.printStackTrace();
        }
        return "方法执行完成";
    }

工具类中实现了AutoCloseable接口和controller中try中的写法是为了实现JDK提供的自动关闭动能（不用写finally解锁）。

###        6.3 基于Zookeeper实现分布式锁 ###

**             ** 实现Zookeeper分布式锁之前，小伙伴们需要对Zookeeper的持久节点、瞬时节点以及观察器有一定的了解，笔者Zookeeper学的略渣，就不在这瞎总结了，这里就默认大家已经了解。

**              实现原理：**利用Zookeeper创建瞬时节点有序的特性，在多线程并发创建瞬时节点，会得到有序的节点序列，我们规定序号最小的线程获得锁。其他线程利用观察器监听自己序号前一个序号的存在状态，前一个线程执行完成，删除自己序号的节点，下一个序号的线程得到通知，执行自己的任务。

**              加锁信号量：**Zookeeper创建瞬时节点有序性

**              解锁信号量：**Zookeeper删除节点动作

**              实现流程：**

**引入Zookeeper依赖：**

<dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.14</version>
    </dependency>

**Zookeeper分布式锁工具类：**

public class ZkLock implements AutoCloseable, Watcher {
    
        private ZooKeeper zooKeeper;
        private String znode;
    
        public ZkLock() throws IOException {
            this.zooKeeper = new ZooKeeper("localhost:2181", 10000,this);
        }
    
        public boolean getLock(String businessCode) {
            try {
                //创建业务根节点
                Stat stat = zooKeeper.exists("/" + businessCode, false);
                if (stat==null){
                    zooKeeper.create("/" + businessCode,businessCode.getBytes(),
                            ZooDefs.Ids.OPEN_ACL_UNSAFE,
                            CreateMode.PERSISTENT);
                }
                //创建瞬时有序节点  /order/order_00000001
                znode = zooKeeper.create("/" + businessCode + "/" + businessCode + "_", businessCode.getBytes(),
                        ZooDefs.Ids.OPEN_ACL_UNSAFE,
                        CreateMode.EPHEMERAL_SEQUENTIAL);
                //获取业务节点下 所有的子节点
                List<String> childrenNodes = zooKeeper.getChildren("/" + businessCode, false);
                //子节点排序
                Collections.sort(childrenNodes);
                //获取序号最小的（第一个）子节点
                String firstNode = childrenNodes.get(0);
                //如果创建的节点是第一个子节点，则获得锁
                if (znode.endsWith(firstNode)){
                    return true;
                }
                //不是第一个子节点，则监听前一个节点
                String lastNode = firstNode;
                for (String node:childrenNodes){
                    if (znode.endsWith(node)){
                        zooKeeper.exists("/"+businessCode+"/"+lastNode,true);
                        break;
                    }else {
                        lastNode = node;
                    }
                }
                //实现阻塞功能
                synchronized (this){
                    wait();
                }
                return true;
            } catch (Exception e) {
                e.printStackTrace();
            }
            return false;
        }
        @Override
        public void close() throws Exception {
            zooKeeper.delete(znode,-1);
            zooKeeper.close();
        }
        @Override
        public void process(WatchedEvent event) {
            if (event.getType() == Event.EventType.NodeDeleted){
                synchronized (this){
                    notify();
                }
            }
        }
    }

**controller：**

@RequestMapping("zkLock")
    public String zookeeperLock(){
        try (ZkLock zkLock = new ZkLock()) {
            if (zkLock.getLock("order")){
                //延时模拟业务代码执行
                Thread.sleep(10000);
            }
        }catch (Exception e) {
            e.printStackTrace();
        }
        return "方法执行完成！";
    }

###        6.4 基于Curator实现分布式锁 ###

**             ** Curator是Zookeeper客户端的升级版，已经实现了分布式锁的功能，直接使用即可。

**              实现流程：**

**引入Curator依赖：**

<dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>4.2.0</version>
    </dependency>

**controller：**

@Autowired
    private CuratorFramework client;
    
    @RequestMapping("curatorLock")
    public String curatorLock(){
         InterProcessMutex lock = new InterProcessMutex(client, "/order");
            try{
                if (lock.acquire(30, TimeUnit.SECONDS)){
                    //延时模拟业务代码执行
                    Thread.sleep(10000);
                }
            }catch (Exception e) {
                e.printStackTrace();
            }finally {
                try {
                    lock.release();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        return "方法执行完成！";
    }

###        6.5 基于Redisson实现分布式锁      ###

**             ** Redisson是Redis客户端的升级版，已经实现了分布式锁的功能，直接使用即可。

**              实现流程：**

**引入Redisson依赖：**

<dependency>
        <groupId>org.redisson</groupId>
        <artifactId>redisson</artifactId>
        <version>3.11.2</version>
    </dependency>

**controller：**

@Autowired
    private RedissonClient redisson;
    
    @RequestMapping("redissonLock")
    public String redissonLock() {
        RLock rLock = redisson.getLock("order");
        try {
            rLock.lock(30, TimeUnit.SECONDS);
            //延时模拟业务执行
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            rLock.unlock();
        }
        return "方法执行完成！！";
    }

**在springboot配置文件中配置redis端口：**

spring.redis.host=127.0.0.1

## 7.实现分布式锁五种方案对比 ##

<table>  实现分布式锁五种方案对比  
 <tbody> 
  <tr> 
   <td style="width:121px;">方式</td> 
   <td style="width:275px;">优点</td> 
   <td style="width:405px;">缺点</td> 
  </tr> 
  <tr> 
   <td style="width:121px;">数据库</td> 
   <td style="width:275px;">实现简单，易于理解，利用<code>for update</code>实现阻塞性，数据库宕机自己会把锁释放掉</td> 
   <td style="width:405px;">需要数据库中初始化所有需要加锁的资源；无法实现超时自动释放锁；无法实现可重入性；会对数据库产生性能损耗</td> 
  </tr> 
  <tr> 
   <td style="width:121px;">Redis</td> 
   <td style="width:275px;">易于理解</td> 
   <td style="width:405px;">不支持阻塞，需要自己实现；在redis集群下会出现锁丢失的问题</td> 
  </tr> 
  <tr> 
   <td style="width:121px;">Zookeeper</td> 
   <td style="width:275px;">自己实现的支持阻塞功能</td> 
   <td style="width:405px;">需对Zookeeper有一定理解，程序复杂</td> 
  </tr> 
  <tr> 
   <td style="width:121px;">Curator</td> 
   <td style="width:275px;">官方分布式锁方案，支持阻塞；安全性高，ZK可以持久化且实时监听持有锁的客户端的状态</td> 
   <td style="width:405px;">Zookeeper的强一致性需要同步所有集群节点，有性能损耗；</td> 
  </tr> 
  <tr> 
   <td style="width:121px;">Redisson</td> 
   <td style="width:275px;">官方分布式锁方案，支持阻塞；采用官方RedLock集群化方案，支持Redis集群</td> 
   <td style="width:405px;">&nbsp;</td> 
  </tr> 
 </tbody> 
</table>

[- SSO]: https://blog.csdn.net/qq_36756682/article/details/109203841
[-]: https://blog.csdn.net/qq_36756682/article/details/109411847
[- 1]: https://blog.csdn.net/qq_36756682/article/details/109557781
[- MyCat]: https://blog.csdn.net/qq_36756682/article/details/109669020