利用Redisson实现分布式锁及其底层原理解析

Posted 2023-02-15 yjl33

Redis介绍

redis是一个key-value存储系统。和Memcached类似，它支持存储的value类型相对更多，包括string(字符串)、list(链表)、set(集合)、zset(sorted set --有序集合)和hash（哈希类型）。这些数据类型都支持push/pop、add/remove及取交集并集和差集及更丰富的操作，而且这些操作都是原子性的。在此基础上，redis支持各种不同方式的排序。与memcached一样，为了保证效率，数据都是缓存在内存中。区别的是redis会周期性的把更新的数据写入磁盘或者把修改操作写入追加的记录文件，并且在此基础上实现了master-slave(主从)同步。

Redis的优缺点

优点

1. 速度快，因为数据存在内存中，类似于HashMap，HashMap的优势就是查找和操作的时间复杂度都是O(1)
2. 支持丰富数据类型，支持string，list，set，sorted set，hash
3. 支持事务，操作都是原子性，所谓的原子性就是对数据的更改要么全部执行，要么全部不执行
4. 丰富的特性：可用于缓存，消息，按key设置过期时间，过期后将会自动删除

缺点

在不使用框架的情况下使用起来较为麻烦

分布式Redis的搭建

搭建集群的第一件事情我们需要一些运行在集群模式的Redis实例. 这意味这集群并不是由一些普通的Redis实例组成的，集群模式需要通过配置启用，开启集群模式后的Redis实例便可以使用集群特有的命令和特性了.
下面是一个最少选项的集群的配置文件:

port 7000

cluster-enabled yes

cluster-config-file nodes.conf

cluster-node-timeout 5000

appendonly yes

文件中的 cluster-enabled 选项用于开实例的集群模式， 而 cluster-conf-file 选项则设定了保存节点配置文件的路径， 默认值为 nodes.conf.节点配置文件无须人为修改， 它由 Redis 集群在启动时创建， 并在有需要时自动进行更新。

要让集群正常运作至少需要三个主节点，不过在刚开始试用集群功能时， 强烈建议使用六个节点： 其中三个为主节点， 而其余三个则是各个主节点的从节点。

为了方便测试，直接在同一台计算机内建立六个文件夹7000到7005，分别表示六个Redis实例，在文件夹 7000 至 7005 中， 各创建一个 redis.conf 文件， 文件的内容使用上面的示例配置文件， 但记得将配置中的端口号从 7000 改为与文件夹名字相同的号码。

从 Redis Github 页面 的 unstable 分支中取出最新的 Redis 源码， 编译出可执行文件 redis-server ， 并将文件复制到 cluster-test 文件夹， 然后使用类似以下命令， 在每个标签页中打开一个实例：

/redis-server ./redis.conf

现在我们已经有了六个正在运行中的 Redis 实例， 接下来我们需要使用这些实例来创建集群， 并为每个节点编写配置文件。

通过使用 Redis 集群命令行工具 redis-trib ， 编写节点配置文件的工作可以非常容易地完成： redis-trib 位于 Redis 源码的 src 文件夹中， 它是一个 Ruby 程序， 这个程序通过向实例发送特殊命令来完成创建新集群， 检查集群， 或者对集群进行重新分片（reshared）等工作。

./redis-trib.rb create --replicas 1 127.0.0.1:7000 127.0.0.1:7001
127.0.0.1:7002 127.0.0.1:7003 127.0.0.1:7004 127.0.0.1:7005

这个命令在这里用于创建一个新的集群, 选项–replicas 1 表示我们希望为集群中的每个主节点创建一个从节点。

之后跟着的其他参数则是这个集群实例的地址列表,3个master3个slave redis-trib 会打印出一份预想中的配置给你看， 如果你觉得没问题的话， 就可以输入 yes ， redis-trib 就会将这份配置应用到集群当中,让各个节点开始互相通讯,最后可以得到如下信息：

[OK] All 16384 slots covered

这表示集群中的 16384 个槽都有至少一个主节点在处理， 集群运作正常。

分布式Redis的原理

由上文可知，Redis是以哈希槽的形式对集群进行划分的，整个集群的哈希槽一共有16384个，在有3个Redis实例的情况下，节点A包含从0到5500的哈希槽，节点B包含从5501到11000 的哈希槽，节点C包含从11001到16384的哈希槽。当有新的节点添加进来的时候，会从当前的各个节点中选取一定的槽分配给新添加的节点，当有节点从集群中被删除时，则会将当前节点的槽分配给集群中其他正在运行的节点。每当有新的key添加到Redis中时，会根据算法算出相应的哈希槽来找到对应的集群节点。

Redisson介绍

什么是Redisson

Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。其中包括(BitSet, Set, Multimap, SortedSet, Map, List, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, AtomicLong, CountDownLatch, Publish / Subscribe, Bloom filter, Remote service, Spring cache, Executor service, Live Object service, Scheduler service) Redisson提供了使用Redis的最简单和最便捷的方法。Redisson的宗旨是促进使用者对Redis的关注分离（Separation of Concern），从而让使用者能够将精力更集中地放在处理业务逻辑上。

Redisson的使用

Redisson配置（以spring XML为例）

Maven配置

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>2.2.12</version>
</dependency>
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-annotations</artifactId>
    <version>2.6.0</version>
</dependency>

Spring XML配置

<redisson:client
    id="redisson"
    name="redisson1,redisson2"
    threads="0"
    netty-threads="0"
    codec-ref="myCodec" 
    transport-mode="NIO"
    redisson-reference-enabled="true"
    codec-provider-ref="myCodecProvider"
    resolver-provider-ref="myResolverProvider"
    executor-ref="myExecutor"
    event-loop-group-ref="myEventLoopGroup"
>
    <!--
    这里的name属性和qualifier子元素不能同时使用。

    id和name的属性都可以被用来作为qualifier的备选值。
    -->
    <!--<qualifier value="redisson3"/>-->
    <redisson:cluster-servers
        idle-connection-timeout="10000"
        ping-timeout="1000"
        connect-timeout="10000"
        timeout="3000"
        retry-attempts="3"
        retry-interval="1500"
        reconnection-timeout="3000"
        failed-attempts="3"
        password="do_not_use_if_it_is_not_set"
        subscriptions-per-connection="5"
        client-name="none"
        load-balancer-ref="myLoadBalancer"
        subscription-connection-minimum-idle-size="1"
        subscription-connection-pool-size="50"
        slave-connection-minimum-idle-size="10"
        slave-connection-pool-size="64"
        master-connection-minimum-idle-size="10"
        master-connection-pool-size="64"
        read-mode="SLAVE"
        subscription-mode="SLAVE"
        scan-interval="1000"
    >
        <redisson:node-address value="redis://127.0.0.1:6379" />
        <redisson:node-address value="redis://127.0.0.1:6380" />
        <redisson:node-address value="redis://127.0.0.1:6381" />
    </redisson:cluster-servers>
</redisson:client>

<!-- 最基本配置 -->
<redisson:client>
    <redisson:cluster-servers>
        <redisson:node-address value="redis://127.0.0.1:6379" />
        <redisson:node-address value="redis://127.0.0.1:6380" />
        <redisson:node-address value="redis://127.0.0.1:6381" />
        ...
    </redisson:cluster-servers>
</redisson:client>

• 更多配置方法

Redisson实现分布式锁

锁的种类

可重入锁

RLock lock = redisson.getLock("anyLock");
// 最常见的使用方法
lock.lock();
...
lock.unlock()

// 加锁以后10秒钟自动解锁
// 无需调用unlock方法手动解锁
lock.lock(10, TimeUnit.SECONDS);

// 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
...
lock.unlock();

公平锁

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();

• Redisson同时还为分布式可重入公平锁提供了异步执行的相关方法：

RLock fairLock = redisson.getFairLock("anyLock");
fairLock.lockAsync();
fairLock.lockAsync(10, TimeUnit.SECONDS);
Future<Boolean> res = fairLock.tryLockAsync(100, 10, TimeUnit.SECONDS);

• 联锁

基于Redis的Redisson分布式联锁RedissonMultiLock对象可以将多个RLock对象关联为一个联锁，每个RLock对象实例可以来自于不同的Redisson实例。

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();

红锁

基于Redis的Redisson红锁RedissonRedLock对象实现了Redlock介绍的加锁算法。该对象也可以用来将多个RLock对象关联为一个红锁，每个RLock对象实例可以来自于不同的Redisson实例。

RLock lock1 = redissonInstance1.getLock("lock1");
RLock lock2 = redissonInstance2.getLock("lock2");
RLock lock3 = redissonInstance3.getLock("lock3");

RedissonRedLock lock = new RedissonRedLock(lock1, lock2, lock3);
// 同时加锁：lock1 lock2 lock3
// 红锁在大部分节点上加锁成功就算成功。
lock.lock();
...
lock.unlock();

另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了。

RedissonRedLock lock = new RedissonRedLock(lock1, lock2, lock3);
// 给lock1，lock2，lock3加锁，如果没有手动解开的话，10秒钟后将会自动解开
lock.lock(10, TimeUnit.SECONDS);

// 为加锁等待100秒时间，并在加锁成功10秒钟后自动解开
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
...
lock.unlock();

读写锁

RReadWriteLock rwlock = redisson.getLock("anyRWLock");
// 最常见的使用方法
rwlock.readLock().lock();
// 或
rwlock.writeLock().lock();

信号量

RSemaphore semaphore = redisson.getSemaphore("semaphore");
semaphore.acquire();
//或
semaphore.acquireAsync();
semaphore.acquire(23);
semaphore.tryAcquire();
//或
semaphore.tryAcquireAsync();
semaphore.tryAcquire(23, TimeUnit.SECONDS);
//或
semaphore.tryAcquireAsync(23, TimeUnit.SECONDS);
semaphore.release(10);
semaphore.release();
//或
semaphore.releaseAsync();

• 可过期性信号量

基于Redis的Redisson可过期性信号量（PermitExpirableSemaphore）是在RSemaphore对象的基础上，为每个信号增加了一个过期时间。每个信号可以通过独立的ID来辨识，释放时只能通过提交这个ID才能释放。

RPermitExpirableSemaphore semaphore = redisson.getPermitExpirableSemaphore("mySemaphore");
String permitId = semaphore.acquire();
// 获取一个信号，有效期只有2秒钟。
String permitId = semaphore.acquire(2, TimeUnit.SECONDS);
// ...
semaphore.release(permitId);

闭锁

RCountDownLatch latch = redisson.getCountDownLatch("anyCountDownLatch");
latch.trySetCount(1);
latch.await();

// 在其他线程或其他JVM里
RCountDownLatch latch = redisson.getCountDownLatch("anyCountDownLatch");
latch.countDown();

锁的原理

在Redisson中，使用key来作为是否上锁的标志，当通过getLock(String key)方法获得相应的锁之后，这个key即作为一个锁存储到Redis集群中，在接下来如果有其他的线程尝试获取名为key的锁时，便会向集群中进行查询，如果能够查到这个锁并发现相应的value的值不为0，则表示已经有其他线程申请了这个锁同时还没有释放，则当前线程进入阻塞，否则由当前线程获取这个锁并将value值加一，如果是可重入锁的话，则当前线程每获得一个自身线程的锁，就将value的值加一，而每释放一个锁则将value值减一，直到减至0，完全释放这个锁。因为底层是基于分布式的Redis集群，所以Redisson实现了分布式的锁机制。

加锁

在Redisson中，加锁需要以下三个参数：

KEYS[1] ：需要加锁的key，这里需要是字符串类型。

ARGV[1] ：锁的超时时间，防止死锁

ARGV[2] ：锁的唯一标识，id（UUID.randomUUID()） + “:” + threadId

Future tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId) 
        internalLockLeaseTime = unit.toMillis(leaseTime);
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_LONG,
            // 检查是否key已经被占用，如果没有则设置超时时间和唯一标识，初始化value=1
            "if (redis.call('exists', KEYS[1]) == 0) then " +
                "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                "return nil; " +
            "end; " +
            // 如果锁重入,需要判断锁的key field 都一直情况下 value 加一
            "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; " +
            // 返回剩余的过期时间
            "return redis.call('pttl', KEYS[1]);",
            Collections.singletonList(getName()), internalLockLeaseTime, getLockName(threadId));

解锁

在Redisson中解锁需要以下五个参数：

KEYS[1] ：需要加锁的key，这里需要是字符串类型。

KEYS[2] ：redis消息的ChannelName,一个分布式锁对应唯一的一个channelName:“redisson_lock__channel__” + getName() + “”

ARGV[1] ：reids消息体，这里只需要一个字节的标记就可以，主要标记redis的key已经解锁，再结合redis的Subscribe，能唤醒其他订阅解锁消息的客户端线程申请锁。

ARGV[2] ：锁的超时时间，防止死锁

ARGV[3] ：锁的唯一标识，也就是刚才介绍的 id（UUID.randomUUID()） + “:” + threadId

public void unlock() 
        Boolean opStatus = commandExecutor.evalWrite(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
        // 如果key已经不存在，说明已经被解锁，直接发布（publihs）redis消息
        "if (redis.call('exists', KEYS[1]) == 0) then " +
        "redis.call('publish', KEYS[2], ARGV[1]); " +
        "return 1; " +
        "end;" +
        // key和field不匹配，说明当前客户端线程没有持有锁，不能主动解锁。
        "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
        "return nil;" +
        "end; " +
        "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
        // 如果counter>0说明锁在重入，不能删除key
        "if (counter > 0) then " +
        "redis.call('pexpire', KEYS[1], ARGV[2]); " +
        "return 0; " +
        "else " +
        // 删除key并且publish 解锁消息
        "redis.call('del', KEYS[1]); " +
        "redis.call('publish', KEYS[2], ARGV[1]); " +
        "return 1; "+
        "end; " +
        "return nil;",
        Arrays.asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(Thread.currentThread().getId()));
        if (opStatus == null) 
                throw new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
                        + id + " thread-id: " + Thread.currentThread().getId());
        
        // 解锁成功之后取消更新锁expire的时间任务
        if (opStatus) 
                cancelExpirationRenewal();
        
    

注意点

Redisson 默认的 CommandExecutor 实现是通过 eval 命令来执行 Lua 脚本，所以要求 Redis 的版本必须为 2.6 或以上，否则可能要自己来实现