Redis之分布式锁的实现

àì夳堔傛蜴生んèń 2023-02-21 12:43 82阅读 0赞

分布式锁在分布式系统中比较常见，常见的分布式锁解决方案有：

*  zookeeper，创建顺序临时节点来实现
 *  redis实现（`setnx` / `setex`命令）
 *  数据库唯一主键

这篇文章主要介绍下redis分布式锁怎么实现。在了解分布式锁的实现之前，我们需要了解下分布式锁应该注意些什么：

*  互斥性，即任何时刻都是一个客户端持有锁
 *  避免死锁，一个客户端挂掉之后不能释放锁，导致后面的客户端不能加锁
 *  保证加锁和解锁是同一个客户端
 *  只要大部分的Redis节点正常运行，客户端就可以进行加锁和解锁操作
 *  是一个可重入锁

##### 基于redisson实现分布式锁 #####

接下来直接show me the code，毕竟talk is cheap。

*  引入redisson的依赖，目前最新版是3.13.1

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

*  创建redisson的配置bean，因为支持多种配置，所以列出来常用的config

@Configuration
    public class RedissonConfig {
    
        /**
         * 单机配置（默认连接127.0.0.1:6379）
         *
         * @return
         */
        @Bean
        public RedissonClient createConfig() {
            return Redisson.create();
        }
    
        /**
         * 集群配置
         *
         * @return
         */
        @Bean
        public RedissonClient createConfig1() {
            Config config = new Config();
            config.useClusterServers().addNodeAddress("redis://127.0.0.1:7000", "redis://127.0.0.1:7001")
                    .addNodeAddress("redis://127.0.0.1:7002");
            return Redisson.create(config);
        }
    
        /**
         * 哨兵配置
         *
         * @return
         */
        @Bean
        public RedissonClient createConfig2() {
            Config config = new Config();
            config.useSentinelServers().setMasterName("master").addSentinelAddress("redis://127.0.0.1:7000", "redis://127.0.0.1:7001")
                    .addSentinelAddress("redis://127.0.0.1:7002");
            return Redisson.create(config);
        }
    
        /**
         * 主从配置
         *
         * @return
         */
        @Bean
        public RedissonClient createConfig3() {
            Config config = new Config();
            config.useMasterSlaveServers().setMasterAddress("redis://127.0.0.1:7000")
            		.setSlaveAddresses(Sets.newHashSet("redis://127.0.0.1:7001", "redis://127.0.0.1:7002"));
            return Redisson.create(config);
        }
    }

*  我们模拟一个减库存的操作

public class TestRedisson {
    
        private static ExecutorService executorService = new ThreadPoolExecutor(100, 200, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(1000));
    
        private static int inventory = 1000;
    
        public static void main(String[] args) {
            RedissonClient redissonClient = Redisson.create();
            RLock myLock = redissonClient.getLock("myLock");
            for (int i = 0; i < 1000; i++) {
                executorService.execute(() -> {
                    try {
                        boolean res = myLock.tryLock(1, 2, TimeUnit.SECONDS);
                        if (res) {
                            inventory--;
                            System.out.println("inventory is: " + inventory);
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        myLock.unlock();
                    }
                });
            }
        }
    }

输出结果如下：

inventory is: 999
    inventory is: 998
    inventory is: 997
    inventory is: 996
    inventory is: 995
    .
    .
    .
    .
    .
    .
    inventory is: 3
    inventory is: 2
    inventory is: 1
    inventory is: 0

从输出结果来看，我们这个程序应该没有问题。有条件的小伙伴可以在集群的环境下尝试一下。

##### 关于Redisson其他的分布式锁 #####

除了`myLock.tryLock(1, 2, TimeUnit.SECONDS)`的用法，还支持很多的用法，比如连锁、读写锁和公平锁等等。

*  可重入锁

RLock lock = redisson.getLock("anyLock");
    // 最常见的使用方法
    lock.lock();
    
    // 加锁以后10秒钟自动解锁
    // 无需调用unlock方法手动解锁
    lock.lock(10, TimeUnit.SECONDS);
    
    // 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
    boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
    if (res) {
       try {
         ...
       } finally {
           lock.unlock();
       }
    }

*  公平锁（Fair Lock）

RLock fairLock = redisson.getFairLock("anyLock");
    // 最常见的使用方法
    fairLock.lock();
    
    // 10秒钟以后自动解锁
    // 无需调用unlock方法手动解锁
    fairLock.lock(10, TimeUnit.SECONDS);
    
    // 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
    boolean res = fairLock.tryLock(100, 10, TimeUnit.SECONDS);
    ...
    fairLock.unlock();

*  联锁（MultiLock）

RLock lock1 = redissonInstance1.getLock("lock1");
    RLock lock2 = redissonInstance2.getLock("lock2");
    RLock lock3 = redissonInstance3.getLock("lock3");
    
    RedissonMultiLock lock = new RedissonMultiLock(lock1, lock2, lock3);
    // 同时加锁：lock1 lock2 lock3
    // 所有的锁都上锁成功才算成功。
    lock.lock();
    ...
    lock.unlock();
    
    RedissonMultiLock lock = new RedissonMultiLock(lock1, lock2, lock3);
    // 给lock1，lock2，lock3加锁，如果没有手动解开的话，10秒钟后将会自动解开
    lock.lock(10, TimeUnit.SECONDS);
    
    // 为加锁等待100秒时间，并在加锁成功10秒钟后自动解开
    boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
    ...
    lock.unlock();

*  读写锁（ReadWriteLock）

RReadWriteLock rwlock = redisson.getReadWriteLock("anyRWLock");
    // 最常见的使用方法
    rwlock.readLock().lock();
    // 或
    rwlock.writeLock().lock();
    
    // 10秒钟以后自动解锁
    // 无需调用unlock方法手动解锁
    rwlock.readLock().lock(10, TimeUnit.SECONDS);
    // 或
    rwlock.writeLock().lock(10, TimeUnit.SECONDS);
    
    // 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
    boolean res = rwlock.readLock().tryLock(100, 10, TimeUnit.SECONDS);
    // 或
    boolean res = rwlock.writeLock().tryLock(100, 10, TimeUnit.SECONDS);
    ...
    lock.unlock();

##### 源码分析 #####

在了解了Redisson的使用方法和常见分布式锁之后，我们来简单了解一下他的实现。

*  先看一下构造方法，用来获取锁的实例

public RedissonLock(CommandAsyncExecutor commandExecutor, String name) {
            super(commandExecutor, name);
            // 命令执行器
            this.commandExecutor = commandExecutor;
            // UUID
            this.id = commandExecutor.getConnectionManager().getId();
            // 看门狗，锁超时时间，如果超过这个时间，强制释放锁
            this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();
            this.entryName = id + ":" + name;
            this.pubSub = commandExecutor.getConnectionManager().getSubscribeService().getLockPubSub();
        }

*  加锁的实现

@Override
    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        long time = unit.toMillis(waitTime);
        long current = System.currentTimeMillis();
        long threadId = Thread.currentThread().getId();
        // 获取锁
        Long ttl = tryAcquire(leaseTime, unit, threadId);
        // 如果已经获取到锁，就返回
        if (ttl == null) {
            return true;
        }
        
        // 获取锁超时
        time -= System.currentTimeMillis() - current;
        if (time <= 0) {
            acquireFailed(threadId);
            return false;
        }
        
        // 等待锁释放，并订阅锁
        current = System.currentTimeMillis();
        RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
        // 阻塞等待subscribe的future的结果对象，如果subscribe方法调用超过了time，
        // 说明已经超过了客户端设置的最大wait time，则直接返回false，取消订阅，不再继续申请锁了。
        if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
            if (!subscribeFuture.cancel(false)) {
                subscribeFuture.onComplete((res, e) -> {
                    if (e == null) {
                        unsubscribe(subscribeFuture, threadId);
                    }
                });
            }
            acquireFailed(threadId);
            return false;
        }
    
        try {
        	// 获取锁超时
            time -= System.currentTimeMillis() - current;
            if (time <= 0) {
                acquireFailed(threadId);
                return false;
            }
        	 // 循环获取锁。会有两种情况：①在等待时间内获取锁成功 返回true。②等待时间结束了还没有获取到锁那么返回false
            while (true) {
                long currentTime = System.currentTimeMillis();
                ttl = tryAcquire(leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    return true;
                }
    
                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
                    acquireFailed(threadId);
                    return false;
                }
    
                // waiting for message
                currentTime = System.currentTimeMillis();
                if (ttl >= 0 && ttl < time) {
                    subscribeFuture.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    subscribeFuture.getNow().getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
                }
    
                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
                    acquireFailed(threadId);
                    return false;
                }
            }
        } finally {
            unsubscribe(subscribeFuture, threadId);
        }
    //        return get(tryLockAsync(waitTime, leaseTime, unit));
    }

底层加锁原理

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    	//过期时间
        internalLockLeaseTime = unit.toMillis(leaseTime);
    
        return evalWriteAsync(getName(), LongCodec.INSTANCE, command,
        		//如果锁不存在，则通过hset设置它的值，并设置过期时间
                "if (redis.call('exists', KEYS[1]) == 0) then " +
                        "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        "return nil; " +
                        "end; " +
                        //如果锁已存在，并且锁的是当前线程，则通过hincrby给数值递增1
                        "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                        "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        "return nil; " +
                        "end; " +
                        //如果锁已存在，但并非本线程，则返回过期时间ttl
                        "return redis.call('pttl', KEYS[1]);",
                Collections.singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
    }

*  解锁的实现

@Override
    public RFuture<Void> unlockAsync(long threadId) {
        RPromise<Void> result = new RedissonPromise<Void>();
        // 解锁的方法
        RFuture<Boolean> future = unlockInnerAsync(threadId);
    
        future.onComplete((opStatus, e) -> {
            cancelExpirationRenewal(threadId);
    
            if (e != null) {
                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;
            }
    
            result.trySuccess(null);
        });
    
        return result;
    }
    
    protected RFuture<Boolean> unlockInnerAsync(long threadId) {
        return evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
        		//如果锁已经不存在， 发布锁释放的消息
                "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                        "return nil;" +
                        "end; " +
                        //通过hincrby递减1的方式，释放一次锁
                		//若剩余次数大于0 ，则刷新过期时间
                        "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
                        "if (counter > 0) then " +
                        "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                        "return 0; " +
                        "else " +
                        //否则证明锁已经释放，删除key并发布锁释放的消息
                        "redis.call('del', KEYS[1]); " +
                        "redis.call('publish', KEYS[2], ARGV[1]); " +
                        "return 1; " +
                        "end; " +
                        "return nil;",
                Arrays.asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
    }

##### 参考链接 #####

*  [https://github.com/redisson/redisson][https_github.com_redisson_redisson]
 *  [https://github.com/redisson/redisson/wiki/8.-%E5%88%86%E5%B8%83%E5%BC%8F%E9%94%81%E5%92%8C%E5%90%8C%E6%AD%A5%E5%99%A8][https_github.com_redisson_redisson_wiki_8.-_E5_88_86_E5_B8_83_E5_BC_8F_E9_94_81_E5_92_8C_E5_90_8C_E6_AD_A5_E5_99_A8]

[https_github.com_redisson_redisson]: https://github.com/redisson/redisson
[https_github.com_redisson_redisson_wiki_8.-_E5_88_86_E5_B8_83_E5_BC_8F_E9_94_81_E5_92_8C_E5_90_8C_E6_AD_A5_E5_99_A8]: https://github.com/redisson/redisson/wiki/8.-%E5%88%86%E5%B8%83%E5%BC%8F%E9%94%81%E5%92%8C%E5%90%8C%E6%AD%A5%E5%99%A8