分布式锁Redisson

ゞ浴缸里的玫瑰 2022-09-13 06:26 256阅读 0赞

## 1、背景 ##

分布式锁需要达到的目标有：

*  互斥 （ 不同应用之间、线程与线程之间互斥。）
 *  性能 （锁的粒度和范围都要尽量小，减少不必要的竞争。）
 *  锁的特性 （可重入锁；超时设置；锁判断等）

## 2、实战 ##

引入依赖

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

项目中集成：

@Bean(destroyMethod = "shutdown")
        RedissonClient redissonClient() throws IOException { 
            Config config = new Config();
            config.useSingleServer()
                    .setAddress("redis://" + host + ":" + port)
                    .setPassword(password);
            return Redisson.create(config);
        }

测试代码：

//业务场景key
            RLock lock=redissonClient.getLock("redisson-test-lock");
            try { 
            	//锁租期设置1分钟，方便查看
                lock.lock(1, TimeUnit.MINUTES);
                System.out.println(Thread.currentThread().getName()+"get lock ... start sleep");
                //业务逻辑处理
                Thread.sleep(50000);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }finally { 
                lock.unlock();
            }

测试：  
启动两个服务器，多个线程分别同时访问这两个服务器，可以看到，同一时间只有一个线程能够拿到锁。在redis中的存储key如下：  
![在这里插入图片描述][20d23b90ccfa49ef8efd7a3b07820504.png]

### 锁存储 ###

由上图可见，Redisson的分布式锁是以hash的形式存储在Redis中的。key为用户自定义的用于区分不同业务场景的值，value为uuid+threadId，其中threadId可以为可重入做准备。

### 加锁流程 ###

若线程A拿到了锁，执行lua脚本，并将线程信息保存到redis中。否则，循环尝试获取锁。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAeXlzeA_size_20_color_FFFFFF_t_70_g_se_x_16]

### 为什么要用lua脚本？ ###

之前提到过，加锁时，要同时设置锁的失效时间，防止锁持有时间过长或者无法正常释放。因此这两个步骤要保证原子性，使用lua脚本就是个很好的选择。

### 锁到期了如何续约？ ###

若线程A持有的锁，马上到过期时间即要释放锁了，但线程A的任务还没做完。这时如果释放锁，就要再次参与锁竞争，其实并不是特别合理。最好能实现续约，Redisson提供“看门狗”的机制来实现续约。

加锁成功后，Redisson会给每个线程配置一个看门狗，其生命周期同该线程。负责检查是否需要续约。

## 3、源码 ##

常用类：  
RLock——Lock的分布式实现  
除了Lock接口的方法外，RLock还扩展了一些其他方法，如下：

public interface RLock extends Lock, RExpirable, RLockAsync { 
    
        void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException;
        
        boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException;
    
    	int getHoldCount();
    	......
    }

RedissonLock  
Redisson的核心逻辑处理类，其继承关系如下图所示。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAeXlzeA_size_20_color_FFFFFF_t_70_g_se_x_16 1]

(1) RedissonClient的getLock()

public RLock getLock(String name) { 
            return new RedissonLock(connectionManager.getCommandExecutor(), name);
        }
    
    //返回一个RedissonLock对象：
    //其中id是executor的id，即存储的uuid。
    //name为用户自定义的场景key，对应redis中的key。
    public RedissonLock(CommandAsyncExecutor commandExecutor, String name) { 
            super(commandExecutor, name);
            this.commandExecutor = commandExecutor;
            this.id = commandExecutor.getConnectionManager().getId();
            this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();
            this.entryName = id + ":" + name;
            this.pubSub = commandExecutor.getConnectionManager().getSubscribeService().getLockPubSub();
        }
    protected String getLockName(long threadId) { 
            return id + ":" + threadId;
        }

（2）lock.lock()

public void lock(long leaseTime, TimeUnit unit) { 
            try { 
                lockInterruptibly(leaseTime, unit);
            } catch (InterruptedException e) { 
                Thread.currentThread().interrupt();
            }
        }
        
    public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException { 
    		//获取线程id
            long threadId = Thread.currentThread().getId();
            //尝试获取锁，1-ttl为null，即该线程未获取锁。2-ttl有值，即该线程已获得锁，还剩ttl的有效时间。
            Long ttl = tryAcquire(leaseTime, unit, threadId);
            // 已获得锁，返回。
            if (ttl == null) { 
                return;
            }
    		//未获得锁，订阅锁的消息，当锁释放时，能收到消息。
            RFuture<RedissonLockEntry> future = subscribe(threadId);
            commandExecutor.syncSubscription(future);
    
    		//未获得锁，循环尝试获取锁。
            try { 
                while (true) { 
                    ttl = tryAcquire(leaseTime, unit, threadId);
                    //当锁已经获得后，返回。
                    if (ttl == null) { 
                        break;
                    }
    
                    //阻塞等待锁，通过信号量阻塞，等待锁释放消息
                    if (ttl >= 0) { 
                        getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                    } else { 
                        getEntry(threadId).getLatch().acquire();
                    }
                }
            } finally { 
            	//最终都要取消订阅
                unsubscribe(future, threadId);
            }
        }

（3）lock.unlock()

public void unlock() { 
            try { 
                get(unlockAsync(Thread.currentThread().getId()));
            } catch (RedisException e) { 
                if (e.getCause() instanceof IllegalMonitorStateException) { 
                    throw (IllegalMonitorStateException) e.getCause();
                } else { 
                    throw e;
                }
            }
        }
        //异步解锁-根据线程id
    public RFuture<Void> unlockAsync(long threadId) { 
            RPromise<Void> result = new RedissonPromise<Void>();
            RFuture<Boolean> future = unlockInnerAsync(threadId);
    
            future.onComplete((opStatus, e) -> { 
                if (e != null) { 
                    cancelExpirationRenewal(threadId);
                    result.tryFailure(e);
                    return;
                }
    
                if (opStatus == null) { 
                    IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
                            + id + " thread-id: " + threadId);
                    result.tryFailure(cause);
                    return;
                }
                if (opStatus) { 
                    cancelExpirationRenewal(null);
                }
                result.trySuccess(null);
            });
    
            return result;
        }
        //在异步结果返回之前阻塞当前线程
     public <V> V get(RFuture<V> future) { 
            if (!future.isDone()) { 
                CountDownLatch l = new CountDownLatch(1);
                future.onComplete((res, e) -> { 
                    l.countDown();
                });
    
                boolean interrupted = false;
                while (!future.isDone()) { 
                    try { 
                        l.await();
                    } catch (InterruptedException e) { 
                        interrupted = true;
                        break;
                    }
                }
    
                if (interrupted) { 
                    Thread.currentThread().interrupt();
                }
            }

## 4、小结 ##

Redisson不仅实现了分布式锁，也做了额外的补充。RedissonLock是可重入的，并且考虑了失败重试，可以设置锁的最大等待时间， 在实现上也做了一些优化，减少了无效的锁申请，提升了资源的利用率。

[20d23b90ccfa49ef8efd7a3b07820504.png]: /images/20220828/234bfff8e55f4f069c138ee04683778c.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAeXlzeA_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/49c39dcf889a48d3b2e919db0f30d8d8.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAeXlzeA_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220828/33f990d194d24cddb10db327d3f7acdc.png