如果你之前是在用 Redis 的话，那使用 Redisson 的话将会事半功倍，Redisson 提供了使用 Redis的最简单和最便捷的方法。
 
Redisson的宗旨是促进使用者对 Redis 的关注分离（Separation of Concern），从而让使用者能够将精力更集中地放在处理业务逻辑上。
 
Redisson 是一个在 Redis 的基础上实现的 Java 驻内存数据网格（In-Memory Data Grid）。
 
 
 
Netty 框架：Redisson采用了基于NIO的Netty框架，不仅能作为Redis底层驱动客户端，具备提供对Redis各种组态形式的连接功能，对Redis命令能以同步发送、异步形式发送、异步流形式发送或管道形式发送的功能，LUA脚本执行处理，以及处理返回结果的功能
 
 
基础数据结构：将原生的Redis Hash，List，Set，String，Geo，HyperLogLog等数据结构封装为Java里大家最熟悉的映射（Map），列表（List），集（Set），通用对象桶（Object Bucket），地理空间对象桶（Geospatial Bucket），基数估计算法（HyperLogLog）等结构，
 
 
分布式数据结构：这基础上还提供了分布式的多值映射（Multimap），本地缓存映射（LocalCachedMap），有序集（SortedSet），计分排序集（ScoredSortedSet），字典排序集（LexSortedSet），列队（Queue），阻塞队列（Blocking Queue），有界阻塞列队（Bounded Blocking Queue），双端队列（Deque），阻塞双端列队（Blocking Deque），阻塞公平列队（Blocking Fair Queue），延迟列队（Delayed Queue），布隆过滤器（Bloom Filter），原子整长形（AtomicLong），原子双精度浮点数（AtomicDouble），BitSet等Redis原本没有的分布式数据结构。
 
 
分布式锁：Redisson还实现了Redis文档中提到像分布式锁Lock这样的更高阶应用场景。事实上Redisson并没有不止步于此，在分布式锁的基础上还提供了联锁（MultiLock），读写锁（ReadWriteLock），公平锁（Fair Lock），红锁（RedLock），信号量（Semaphore），可过期性信号量（PermitExpirableSemaphore）和闭锁（CountDownLatch）这些实际当中对多线程高并发应用至关重要的基本部件。正是通过实现基于Redis的高阶应用方案，使Redisson成为构建分布式系统的重要工具。
 
 
节点：Redisson作为独立节点可以用于独立执行其他节点发布到分布式执行服务和分布式调度服务里的远程任务。
 
使用 RList 操作 Redis 列表
 
下面的代码简单演示了如何在 Redisson 中使用 RList 对象。RList 是 Java 的 List 集合的分布式并发实现。考虑以下代码：
 
import org.redisson.Redisson;
import org.redisson.api.RList;
import org.redisson.api.RedissonClient;

public class ListExamples {
    public static void main(String[] args) {
        // 默认连接上 127.0.0.1:6379
        RedissonClient client = Redisson.create();

        // RList 继承了 java.util.List 接口
        RList<String> nameList = client.getList("nameList");
        nameList.clear();
        nameList.add("bingo");
        nameList.add("yanglbme");
        nameList.add("https://github.com/yanglbme");
        nameList.remove(-1);

        boolean contains = nameList.contains("yanglbme");
        System.out.println("List size: " + nameList.size());
        System.out.println("Is list contains name 'yanglbme': " + contains);
        nameList.forEach(System.out::println);

        client.shutdown();
    }
}
复制代码
 
运行上面的代码时，可以获得以下输出：
 
List size: 2
Is list contains name 'yanglbme': true
bingo
yanglbme
复制代码
 
使用 RMap 操作 Redis 哈希
 
Redisson 还包括 RMap，它是 Java Map 集合的分布式并发实现，考虑以下代码：
 
import org.redisson.Redisson;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;

public class MapExamples {
    public static void main(String[] args) {
        // 默认连接上127.0.0.1:6379
        RedissonClient client = Redisson.create();
        // RMap 继承了 java.util.concurrent.ConcurrentMap 接口
        RMap<String, String> map = client.getMap("personalInfo");
        map.put("name", "yanglbme");
        map.put("address", "Shenzhen");
        map.put("link", "https://github.com/yanglbme");

        boolean contains = map.containsKey("link");
        System.out.println("Map size: " + map.size());
        System.out.println("Is map contains key 'link': " + contains);
        String value = map.get("name");
        System.out.println("Value mapped by key 'name': " + value);
        boolean added = map.putIfAbsent("link", "https://doocs.github.io") == null;
        System.out.println("Is value mapped by key 'link' added: " + added);
        client.shutdown();
    }
}
复制代码
 
运行上面的代码时，将会看到以下输出：
 
Map size: 3
Is map contains key 'link': true
Value mapped by key 'name': yanglbme
Is value mapped by key 'link' added: false
复制代码
 
使用 RLock 实现 Redis 分布式锁
 
RLock 是 Java 中可重入锁的分布式实现，下面的代码演示了 RLock 的用法：
 
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;

public class LockExamples {
    public static void main(String[] args) throws InterruptedException {
        // 默认连接上127.0.0.1:6379
        RedissonClient client = Redisson.create();
        // RLock 继承了 java.util.concurrent.locks.Lock 接口
        RLock lock = client.getLock("lock");

        lock.lock();
        System.out.println("lock acquired");

        Thread t = new Thread(() -> {
            RLock lock1 = client.getLock("lock");
            lock1.lock();
            System.out.println("lock acquired by thread");
            lock1.unlock();
            System.out.println("lock released by thread");
        });

        t.start();
        t.join(1000);

        lock.unlock();
        System.out.println("lock released");

        t.join();

        client.shutdown();
    }
}
复制代码
 
此代码将产生以下输出：
 
lock acquired
lock released
lock acquired by thread
lock released by thread
复制代码
 
使用 RAtomicLong 实现 Redis 原子操作
 
RAtomicLong 是 Java 中 AtomicLong 类的分布式“替代品”，用于在并发环境中保存长值。以下示例代码演示了 RAtomicLong 的用法：
 
import org.redisson.Redisson;
import org.redisson.api.RAtomicLong;
import org.redisson.api.RedissonClient;

public class AtomicLongExamples {
    public static void main(String[] args) {
        // 默认连接上127.0.0.1:6379
        RedissonClient client = Redisson.create();
        RAtomicLong atomicLong = client.getAtomicLong("myLong");

        System.out.println("Init value: " + atomicLong.get());

        atomicLong.incrementAndGet();
        System.out.println("Current value: " + atomicLong.get());

        atomicLong.addAndGet(10L);
        System.out.println("Final value: " + atomicLong.get());

        client.shutdown();
    }
}
复制代码
 
此代码的输出将是：
 
Init value: 0
Current value: 1
Final value: 11
复制代码
 
 
 
可重入锁测试
 
基于Redis的Redisson分布式可重入锁RLockJava 对象实现了java.util.concurrent.locks.Lock接口。同时还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。
 
RLock lock = redisson.getLock("anyLock");
// 最常见的使用方法
lock.lock();
复制代码
 
我们用 passjava 这个开源项目测试下可重入锁的两个点：
 
（1）多个线程抢占锁，后面锁需要等待吗？
（2）如果抢占到锁的线程所在的服务停了，锁会不会被释放？
3.1.1 验证一：可重入锁是阻塞的吗？
 
为了验证以上两点，我写了个 demo 程序：代码的流程就是设置WuKong-lock锁，然后加锁，打印线程 ID，等待 10 秒后释放锁，最后返回响应：“test lock ok”。
 
@ResponseBody
@GetMapping("test-lock")
public String TestLock() {
    // 1.获取锁，只要锁的名字一样，获取到的锁就是同一把锁。
    RLock lock = redisson.getLock("WuKong-lock");

    // 2.加锁
    lock.lock();
    try {
        System.out.println("加锁成功，执行后续代码。线程 ID：" + Thread.currentThread().getId());
        Thread.sleep(10000);
    } catch (Exception e) {
        //TODO
    } finally {
        lock.unlock();
        // 3.解锁
        System.out.println("Finally，释放锁成功。线程 ID：" + Thread.currentThread().getId());
    }

    return "test lock ok";
}
复制代码
 
先验证第一个点，用两个 http 请求来测试抢占锁。
 
请求的 URL：
 
http://localhost:11000/question/v1/redisson/test/test-lock
复制代码
 
 
第一个线程对应的线程 ID 为 86，10秒后，释放锁。在这期间，第二个线程需要等待锁释放。
 
第一个线程释放锁之后，第二个线程获取到了锁，10 秒后，释放锁。
 
画了一个流程图，帮助大家理解。如下图所示：
 
 
第一步：线程 A 在 0 秒时，抢占到锁，0.1 秒后，开始执行等待 10 s。
第二步：线程 B 在 0.1 秒尝试抢占锁，未能抢到锁（被 A 抢占了）。
第三步：线程 A 在 10.1 秒后，释放锁。
第四步：线程 B 在 10.1 秒后抢占到锁，然后等待 10 秒后释放锁。
由此可以得出结论，Redisson 的可重入锁（lock）是阻塞其他线程的，需要等待其他线程释放的。
 
3.1.2 验证二：服务停了，锁会释放吗？
 
如果线程 A 在等待的过程中，服务突然停了，那么锁会释放吗？如果不释放的话，就会成为死锁，阻塞了其他线程获取锁。
 
我们先来看下线程 A 的获取锁后的，Redis 客户端查询到的结果，如下图所示：
 
 
WuKong-lock 有值，而且大家可以看到 TTL 在不断变小，说明 WuKong-lock 是自带过期时间的。
 
通过观察，经过 30 秒后，WuKong-lock 过期消失了。说明 Redisson 在停机后，占用的锁会自动释放。
 
 
那这又是什么原理呢？这里就要提一个概念了，看门狗。
 
 
3.2 看门狗原理
 
如果负责储存这个分布式锁的 Redisson 节点宕机以后，而且这个锁正好处于锁住的状态时，这个锁会出现锁死的状态。为了避免这种情况的发生，Redisson内部提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期。
 
默认情况下，看门狗的检查锁的超时时间是30秒钟，也可以通过修改Config.lockWatchdogTimeout来另行指定。
 
如果我们未制定 lock 的超时时间，就使用 30 秒作为看门狗的默认时间。只要占锁成功，就会启动一个定时任务：每隔 10 秒重新给锁设置过期的时间，过期时间为 30 秒。
 
如下图所示：
 
 
当服务器宕机后，因为锁的有效期是 30 秒，所以会在 30 秒内自动解锁。（30秒等于宕机之前的锁占用时间+后续锁占用的时间）。
 
如下图所示：
 
 
3.3 设置锁过期时间
 
我们也可以通过给锁设置过期时间，让其自动解锁。
 
如下所示，设置锁 8 秒后自动过期。
 
lock.lock(8, TimeUnit.SECONDS);
复制代码
 
如果业务执行时间超过 8 秒，手动释放锁将会报错，如下图所示：
 
 
所以我们如果设置了锁的自动过期时间，则执行业务的时间一定要小于锁的自动过期时间，否则就会报错。
 
Redis的分布式锁实现
 
1. 利用setnx+expire命令 (错误的做法)
 
Redis的SETNX命令，setnx key value，将key设置为value，当键不存在时，才能成功，若键存在，什么也不做，成功返回1，失败返回0 。 SETNX实际上就是SET IF NOT Exists的缩写
 
因为分布式锁还需要超时机制，所以我们利用expire命令来设置，所以利用setnx+expire命令的核心代码如下：
 
public boolean tryLock(String key,String requset,int timeout) {
    Long result = jedis.setnx(key, requset);
    // result = 1时，设置成功，否则设置失败
    if (result == 1L) {
        return jedis.expire(key, timeout) == 1L;
    } else {
        return false;
    }
}
复制代码
 
实际上上面的步骤是有问题的，setnx和expire是分开的两步操作，不具有原子性，如果执行完第一条指令应用异常或者重启了，锁将无法过期。
 
一种改善方案就是使用Lua脚本来保证原子性（包含setnx和expire两条指令）
 
2. 使用Lua脚本（包含setnx和expire两条指令）
 
代码如下
 
public boolean tryLock_with_lua(String key, String UniqueId, int seconds) {
    String lua_scripts = "if redis.call('setnx',KEYS[1],ARGV[1]) == 1 then" +
            "redis.call('expire',KEYS[1],ARGV[2]) return 1 else return 0 end";
    List<String> keys = new ArrayList<>();
    List<String> values = new ArrayList<>();
    keys.add(key);
    values.add(UniqueId);
    values.add(String.valueOf(seconds));
    Object result = jedis.eval(lua_scripts, keys, values);
    //判断是否成功
    return result.equals(1L);
}
复制代码
 
3. 使用 set key value [EX seconds][PX milliseconds][NX|XX] 命令 (正确做法)
 
Redis在 2.6.12 版本开始，为 SET 命令增加一系列选项：
 
SET key value[EX seconds][PX milliseconds][NX|XX]
复制代码
 
EX seconds: 设定过期时间，单位为秒
PX milliseconds: 设定过期时间，单位为毫秒
NX: 仅当key不存在时设置值
XX: 仅当key存在时设置值
set命令的nx选项，就等同于setnx命令，代码过程如下：
 
public boolean tryLock_with_set(String key, String UniqueId, int seconds) {
    return "OK".equals(jedis.set(key, UniqueId, "NX", "EX", seconds));
}
复制代码
 
value必须要具有唯一性，我们可以用UUID来做，设置随机字符串保证唯一性，至于为什么要保证唯一性？假如value不是随机字符串，而是一个固定值，那么就可能存在下面的问题：
 
1.客户端1获取锁成功
2.客户端1在某个操作上阻塞了太长时间
3.设置的key过期了，锁自动释放了
4.客户端2获取到了对应同一个资源的锁
5.客户端1从阻塞中恢复过来，因为value值一样，所以执行释放锁操作时就会释放掉客户端2持有的锁，这样就会造成问题
所以通常来说，在释放锁时，我们需要对value进行验证
 
释放锁的实现
 
释放锁时需要验证value值，也就是说我们在获取锁的时候需要设置一个value，不能直接用del key这种粗暴的方式，因为直接del key任何客户端都可以进行解锁了，所以解锁时，我们需要判断锁是否是自己的，基于value值来判断，代码如下：
 
public boolean releaseLock_with_lua(String key,String value) {
    String luaScript = "if redis.call('get',KEYS[1]) == ARGV[1] then " +
            "return redis.call('del',KEYS[1]) else return 0 end";
    return jedis.eval(luaScript, Collections.singletonList(key), Collections.singletonList(value)).equals(1L);
}
复制代码
 
这里使用Lua脚本的方式，尽量保证原子性。
 
使用 set key value [EX seconds][PX milliseconds][NX|XX] 命令 看上去很OK，实际上在Redis集群的时候也会出现问题，比如说A客户端在Redis的master节点上拿到了锁，但是这个加锁的key还没有同步到slave节点，master故障，发生故障转移，一个slave节点升级为master节点，B客户端也可以获取同个key的锁，但客户端A也已经拿到锁了，这就导致多个客户端都拿到锁。
 
所以针对Redis集群这种情况，还有其他方案
 
4. Redlock算法 与 Redisson 实现
 
Redis作者 antirez基于分布式环境下提出了一种更高级的分布式锁的实现Redlock，原理如下：
 
 
 
下面参考文章Redlock：Redis分布式锁最牛逼的实现 和 redis.io/topics/dist…
 
 
假设有5个独立的Redis节点（注意这里的节点可以是5个Redis单master实例，也可以是5个Redis Cluster集群，但并不是有5个主节点的cluster集群）：
 
获取当前Unix时间，以毫秒为单位
依次尝试从5个实例，使用相同的key和具有唯一性的value(例如UUID)获取锁，当向Redis请求获取锁时，客户端应该设置一个网络连接和响应超时时间，这个超时时间应用小于锁的失效时间，例如你的锁自动失效时间为10s，则超时时间应该在5~50毫秒之间，这样可以避免服务器端Redis已经挂掉的情况下，客户端还在死死地等待响应结果。如果服务端没有在规定时间内响应，客户端应该尽快尝试去另外一个Redis实例请求获取锁
客户端使用当前时间减去开始获取锁时间（步骤1记录的时间）就得到获取锁使用的时间，当且仅当从大多数(N/2+1，这里是3个节点)的Redis节点都取到锁，并且使用的时间小于锁失败时间时，锁才算获取成功。
如果取到了锁，key的真正有效时间等于有效时间减去获取锁所使用的时间（步骤3计算的结果）
如果某些原因，获取锁失败（没有在至少N/2+1个Redis实例取到锁或者取锁时间已经超过了有效时间），客户端应该在所有的Redis实例上进行解锁（即便某些Redis实例根本就没有加锁成功，防止某些节点获取到锁但是客户端没有得到响应而导致接下来的一段时间不能被重新获取锁）
Redisson实现简单分布式锁
 
对于Java用户而言，我们经常使用Jedis，Jedis是Redis的Java客户端，除了Jedis之外，Redisson也是Java的客户端，Jedis是阻塞式I/O，而Redisson底层使用Netty可以实现非阻塞I/O，该客户端封装了锁的，继承了J.U.C的Lock接口，所以我们可以像使用ReentrantLock一样使用Redisson，具体使用过程如下。
 
首先加入POM依赖
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.10.6</version>
</dependency>
复制代码
 
使用Redisson，代码如下(与使用ReentrantLock类似）
// 1. 配置文件
Config config = new Config();
config.useSingleServer()
        .setAddress("redis://127.0.0.1:6379")
        .setPassword(RedisConfig.PASSWORD)
        .setDatabase(0);
//2. 构造RedissonClient
RedissonClient redissonClient = Redisson.create(config);

//3. 设置锁定资源名称
RLock lock = redissonClient.getLock("redlock");
lock.lock();
try {
    System.out.println("获取锁成功，实现业务逻辑");
    Thread.sleep(10000);
} catch (InterruptedException e) {
    e.printStackTrace();
} finally {
    lock.unlock();
}
复制代码
 
 

  
 
 

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），从而让使用者能够将精力更集中地放在处理业务逻辑上。
 
 
 
官网： https://redisson.org/
 
github: https://github.com/redisson/redisson
 
中文文档： https://github.com/redisson/redisson/wiki/%E7%9B%AE%E5%BD%95
 
 
 
配置支持： 集群、哨兵、主从、单节点、云服务
 
支持对象： 分布式对象、分布式集合、分布式锁、分布式服务
 
集成整合： 可以整合 spring、spring-cache、 hibernate、 jcache、 Tomcat-session、spring-session
 
 

Redisson实现分布式锁(1)原理
 
有关Redisson作为实现分布式锁，总的分3大模块来讲。
 
1、Redisson实现分布式锁原理
2、Redisson实现分布式锁的源码解析
3、Redisson实现分布式锁的项目代码（可以用于实际项目中）
 
本文只介绍Redisson如何实现分布式锁的原理。其它的会在接下来的博客讲，最后有关Redisson实现分布式锁的项目代码的博客中会放上项目源码到GitHub上。
 
一、高效分布式锁
 
当我们在设计分布式锁的时候，我们应该考虑分布式锁至少要满足的一些条件，同时考虑如何高效的设计分布式锁，这里我认为以下几点是必须要考虑的。
 
1、互斥
 
在分布式高并发的条件下，我们最需要保证，同一时刻只能有一个线程获得锁，这是最基本的一点。
 
2、防止死锁
 
在分布式高并发的条件下，比如有个线程获得锁的同时，还没有来得及去释放锁，就因为系统故障或者其它原因使它无法执行释放锁的命令,导致其它线程都无法获得锁，造成死锁。
 
所以分布式非常有必要设置锁的有效时间，确保系统出现故障后，在一定时间内能够主动去释放锁，避免造成死锁的情况。
 
3、性能
 
对于访问量大的共享资源，需要考虑减少锁等待的时间，避免导致大量线程阻塞。
 
所以在锁的设计时，需要考虑两点。
 
1、锁的颗粒度要尽量小。比如你要通过锁来减库存，那这个锁的名称你可以设置成是商品的ID,而不是任取名称。这样这个锁只对当前商品有效,锁的颗粒度小。
 
2、锁的范围尽量要小。比如只要锁2行代码就可以解决问题的，那就不要去锁10行代码了。
 
4、重入
 
我们知道ReentrantLock是可重入锁，那它的特点就是：同一个线程可以重复拿到同一个资源的锁。重入锁非常有利于资源的高效利用。关于这点之后会做演示。
 
针对以上Redisson都能很好的满足，下面就来分析下它。
 
二、Redisson原理分析
 
为了更好的理解分布式锁的原理，画张图来分析。
 
 
1、加锁机制
 
线程去获取锁，获取成功: 执行lua脚本，保存数据到redis数据库。
 
线程去获取锁，获取失败: 一直通过while循环尝试获取锁，获取成功后，执行lua脚本，保存数据到redis数据库。
 
2、watch dog自动延期机制
 
在一个分布式环境下，假如一个线程获得锁后，突然服务器宕机了，那么这个时候在一定时间后这个锁会自动释放，你也可以设置锁的有效时间(不设置默认30秒），这样的目的主要是防止死锁的发生。
 
但在实际开发中会有下面一种情况:
 
//设置锁1秒过去
redissonLock.lock("redisson", 1);
/**
* 业务逻辑需要咨询2秒
*/
redissonLock.release("redisson");

/**
* 线程1 进来获得锁后，线程一切正常并没有宕机，但它的业务逻辑需要执行2秒，这就会有个问题，在 线程1 执行1秒后，这个锁就自动过期了，
* 那么这个时候 线程2 进来了。那么就存在 线程1和线程2 同时在这段业务逻辑里执行代码，这当然是不合理的。
* 而且如果是这种情况，那么在解锁时系统会抛异常，因为解锁和加锁已经不是同一线程了，具体后面代码演示。
*/
 
所以这个时候看门狗就出现了，它的作用就是 线程1 业务还没有执行完，时间就过了，线程1 还想持有锁的话，就会启动一个watch dog后台线程，不断的延长锁key的生存时间。
 
注意 正常这个看门狗线程是不启动的，还有就是这个看门狗启动后对整体性能也会有一定影响，所以不建议开启看门狗。
 
3、为啥要用lua脚本呢？
 
这个不用多说，主要是如果你的业务逻辑复杂的话，通过封装在lua脚本中发送给redis，而且redis是单线程的，这样就保证这段复杂业务逻辑执行的原子性。
 
4、可重入加锁机制
 
Redisson可以实现可重入加锁机制的原因，有两点：
 
1、Redis存储锁的数据类型是 Hash类型
2、Hash数据类型的key值包含了当前线程信息。
 
下面是redis存储的数据
 
 这里表面数据类型是Hash类型,Hash类型相当于我们java的 <key,<key1,value>> 类型,这里key是指 ‘redisson’
 
它的有效期还有9秒，我们再来看里们的key1值为078e44a3-5f95-4e24-b6aa-80684655a15a:45它的组成是:
 
guid + 当前线程的ID。后面的value是就和可重入加锁有关。
 
举图说明
 
 
上面这图的意思就是可重入锁的机制，它最大的优点就是相同线程不需要在等待锁，而是可以直接进行相应操作。
 
5、Redis分布式锁的缺点
 
Redis分布式锁会有个缺陷，就是在Redis哨兵模式下:
 
客户端1 对某个master节点写入了redisson锁，此时会异步复制给对应的 slave节点。但是这个过程中一旦发生 master节点宕机，主备切换，slave节点从变为了 master节点。
 
这时客户端2 来尝试加锁的时候，在新的master节点上也能加锁，此时就会导致多个客户端对同一个分布式锁完成了加锁。
 
这时系统在业务语义上一定会出现问题，导致各种脏数据的产生。
 
缺陷在哨兵模式或者主从模式下，如果 master实例宕机的时候，可能导致多个客户端同时完成加锁。
 
Redisson实现分布式锁(2)—RedissonLock
 
Redisson分布式锁的实现是基于RLock接口，RedissonLock实现RLock接口。
 
一、RLock接口
 
1、概念
 
public interface RLock extends Lock, RExpirable, RLockAsync
 
很明显RLock是继承Lock锁，所以他有Lock锁的所有特性，比如lock、unlock、trylock等特性,同时它还有很多新特性：强制锁释放，带有效期的锁,。
 
2、RLock锁API
 
这里针对上面做个整理，这里列举几个常用的接口说明
 
public interface RRLock {
    //----------------------Lock接口方法-----------------------

    /**
     * 加锁 锁的有效期默认30秒
     */
    void lock();
    /**
     * tryLock()方法是有返回值的，它表示用来尝试获取锁，如果获取成功，则返回true，如果获取失败（即锁已被其他线程获取），则返回false .
     */
    boolean tryLock();
    /**
     * tryLock(long time, TimeUnit unit)方法和tryLock()方法是类似的，只不过区别在于这个方法在拿不到锁时会等待一定的时间，
     * 在时间期限之内如果还拿不到锁，就返回false。如果如果一开始拿到锁或者在等待期间内拿到了锁，则返回true。
     *
     * @param time 等待时间
     * @param unit 时间单位 小时、分、秒、毫秒等
     */
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
    /**
     * 解锁
     */
    void unlock();
    /**
     * 中断锁 表示该锁可以被中断 假如A和B同时调这个方法，A获取锁，B为获取锁，那么B线程可以通过
     * Thread.currentThread().interrupt(); 方法真正中断该线程
     */
    void lockInterruptibly();

    //----------------------RLock接口方法-----------------------
    /**
     * 加锁 上面是默认30秒这里可以手动设置锁的有效时间
     *
     * @param leaseTime 锁有效时间
     * @param unit      时间单位 小时、分、秒、毫秒等
     */
    void lock(long leaseTime, TimeUnit unit);
    /**
     * 这里比上面多一个参数，多添加一个锁的有效时间
     *
     * @param waitTime  等待时间
     * @param leaseTime 锁有效时间
     * @param unit      时间单位 小时、分、秒、毫秒等
     */
    boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException;
    /**
     * 检验该锁是否被线程使用，如果被使用返回True
     */
    boolean isLocked();
    /**
     * 检查当前线程是否获得此锁（这个和上面的区别就是该方法可以判断是否当前线程获得此锁，而不是此锁是否被线程占有）
     * 这个比上面那个实用
     */
    boolean isHeldByCurrentThread();
    /**
     * 中断锁 和上面中断锁差不多，只是这里如果获得锁成功,添加锁的有效时间
     * @param leaseTime  锁有效时间
     * @param unit       时间单位 小时、分、秒、毫秒等
     */
    void lockInterruptibly(long leaseTime, TimeUnit unit);  
}
 
RLock相关接口，主要是新添加了 leaseTime 属性字段，主要是用来设置锁的过期时间,避免死锁。
 
二、RedissonLock实现类
 
public class RedissonLock extends RedissonExpirable implements RLock
 
RedissonLock实现了RLock接口，所以实现了接口的具体方法。这里我列举几个方法说明下
 
1、void lock()方法
 
@Override
public void lock() {
    try {
        lockInterruptibly();
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}
 
发现lock锁里面进去其实用的是lockInterruptibly（中断锁，表示可以被中断）,而且捕获异常后用 Thread.currentThread().interrupt()来真正中断当前线程，其实它们是搭配一起使用的。
 
具体有关lockInterruptibly()方法讲解推荐一个博客。博客：Lock的lockInterruptibly()
 
接下来执行流程,这里理下关键几步
 
   /**
     * 1、带上默认值调另一个中断锁方法
     */
    @Override
    public void lockInterruptibly() throws InterruptedException {
        lockInterruptibly(-1, null);
    }
    /**
     * 2、另一个中断锁的方法
     */
    void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException 
    /**
     * 3、这里已经设置了锁的有效时间默认为30秒  （commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout()=30）
     */
    RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
    /**
     * 4、最后通过lua脚本访问Redis,保证操作的原子性
     */
    <T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        internalLockLeaseTime = unit.toMillis(leaseTime);

        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                "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; " +
                        "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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
    }
 
那么void lock(long leaseTime, TimeUnit unit)方法其实和上面很相似了，就是从上面第二步开始的。
 
2、tryLock(long waitTime, long leaseTime, TimeUnit unit)
 
接口的参数和含义上面已经说过了，现在我们开看下源码，这里只显示一些重要逻辑。
 
@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);
        //1、 获取锁同时获取成功的情况下，和lock(...)方法是一样的 直接返回True，获取锁False再往下走
        if (ttl == null) {
            return true;
        }
        //2、如果超过了尝试获取锁的等待时间,当然返回false 了。
        time -= System.currentTimeMillis() - current;
        if (time <= 0) {
            acquireFailed(threadId);
            return false;
        }

        // 3、订阅监听redis消息，并且创建RedissonLockEntry，其中RedissonLockEntry中比较关键的是一个 Semaphore属性对象,用来控制本地的锁请求的信号量同步，返回的是netty框架的Future实现。
        final RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
        //  阻塞等待subscribe的future的结果对象，如果subscribe方法调用超过了time，说明已经超过了客户端设置的最大wait time，则直接返回false，取消订阅，不再继续申请锁了。
        //  只有await返回true，才进入循环尝试获取锁
        if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) {
            if (!subscribeFuture.cancel(false)) {
                subscribeFuture.addListener(new FutureListener<RedissonLockEntry>() {
                    @Override
                    public void operationComplete(Future<RedissonLockEntry> future) throws Exception {
                        if (subscribeFuture.isSuccess()) {
                            unsubscribe(subscribeFuture, threadId);
                        }
                    }
                });
            }
            acquireFailed(threadId);
            return false;
        }

       //4、如果没有超过尝试获取锁的等待时间，那么通过While一直获取锁。最终只会有两种结果
        //1)、在等待时间内获取锁成功 返回true。2）等待时间结束了还没有获取到锁那么返回false。
        while (true) {
            long currentTime = System.currentTimeMillis();
            ttl = tryAcquire(leaseTime, unit, threadId);
            // 获取锁成功
            if (ttl == null) {
                return true;
            }
           //   获取锁失败
            time -= System.currentTimeMillis() - currentTime;
            if (time <= 0) {
                acquireFailed(threadId);
                return false;
            }
        }
    }
 
重点 tryLock一般用于特定满足需求的场合，但不建议作为一般需求的分布式锁，一般分布式锁建议用void lock(long leaseTime, TimeUnit unit)。因为从性能上考虑，在高并发情况下后者效率是前者的好几倍
 
3、unlock()
 
解锁的逻辑很简单。
 
@Override
    public void unlock() {
        // 1.通过 Lua 脚本执行 Redis 命令释放锁
        Boolean opStatus = commandExecutor.evalWrite(getName(), LongCodec.INSTANCE,
                RedisCommands.EVAL_BOOLEAN,
                "if (redis.call('exists', KEYS[1]) == 0) then " +
                        "redis.call('publish', KEYS[2], ARGV[1]); " +
                        "return 1; " +
                        "end;" +
                        "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                        "return nil;" +
                        "end; " +
                        "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
                        "if (counter > 0) then " +
                        "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                        "return 0; " +
                        "else " +
                        "redis.call('del', KEYS[1]); " +
                        "redis.call('publish', KEYS[2], ARGV[1]); " +
                        "return 1; "+
                        "end; " +
                        "return nil;",
                Arrays.<Object>asList(getName(), getChannelName()),
                LockPubSub.unlockMessage, internalLockLeaseTime,
                getLockName(Thread.currentThread().getId()));
        // 2.非锁的持有者释放锁时抛出异常
        if (opStatus == null) {
            throw new IllegalMonitorStateException(
                    "attempt to unlock lock, not locked by current thread by node id: "
                            + id + " thread-id: " + Thread.currentThread().getId());
        }
        // 3.释放锁后取消刷新锁失效时间的调度任务
        if (opStatus) {
            cancelExpirationRenewal();
        }
    }
 
使用 EVAL 命令执行 Lua 脚本来释放锁：
 
key 不存在，说明锁已释放，直接执行 publish 命令发布释放锁消息并返回 1。
key 存在，但是 field 在 Hash 中不存在，说明自己不是锁持有者，无权释放锁，返回 nil。
因为锁可重入，所以释放锁时不能把所有已获取的锁全都释放掉，一次只能释放一把锁，因此执行 hincrby 对锁的值减一。
释放一把锁后，如果还有剩余的锁，则刷新锁的失效时间并返回 0；如果刚才释放的已经是最后一把锁，则执行 del 命令删除锁的 key，并发布锁释放消息，返回 1。
 
注意这里有个实际开发过程中，容易出现很容易出现上面第二步异常，非锁的持有者释放锁时抛出异常。比如下面这种情况
 
//设置锁1秒过去
redissonLock.lock("redisson", 1);
/**
* 业务逻辑需要咨询2秒
*/
redissonLock.release("redisson");
/**
* 线程1 进来获得锁后，线程一切正常并没有宕机，但它的业务逻辑需要执行2秒，这就会有个问题，在 线程1 执行1秒后，这个锁就自动过期了，
* 那么这个时候 线程2 进来了。在线程1去解锁就会抛上面这个异常（因为解锁和当前锁已经不是同一线程了）
*/
 
Redisson实现分布式锁(3)—项目落地实现
 
一、项目概述
 
1、技术架构
 
项目总体技术选型
 
SpringBoot2.1.5 + Maven3.5.4 + Redisson3.5.4 + lombok(插件)
 
2、加锁方式
 
该项目支持 自定义注解加锁 和 常规加锁 两种模式
 
自定义注解加锁
 
@DistributedLock(value="goods", leaseTime=5)
public String lockDecreaseStock(){
    //业务逻辑
}
 
常规加锁
 
//1、加锁
redissonLock.lock("redisson", 10);
//2、业务逻辑
//3、解锁
redissonLock.unlock("redisson");
 
3、Redis部署方式
 
该项目支持四种Redis部署方式
 
1、单机模式部署
2、集群模式部署
3、主从模式部署
4、哨兵模式部署
 
该项目已经实现支持上面四种模式，你要采用哪种只需要修改配置文件application.properties，项目代码不需要做任何修改。
 
4、项目整体结构
 
redis-distributed-lock-core # 核心实现
|
---src
      |
      ---com.jincou.redisson
                           |# 通过注解方式 实现分布式锁
                           ---annotation
                           |# 配置类实例化RedissonLock
                           ---config
                           |# 放置常量信息
                           ---constant
                           |# 读取application.properties信息后，封装到实体
                           ---entity    
                           |# 支持单机、集群、主从、哨兵 代码实现
                           ---strategy

redis-distributed-lock-web-test # 针对上面实现类的测试类
|
---src
      |
      ---java
            |
            ---com.jincou.controller
                                 |# 测试 基于注解方式实现分布式锁
                                 ---AnnotatinLockController.java
                                 |# 测试 基于常规方式实现分布式锁
                                 ---LockController.java
      ---resources                
           | # 配置端口号 连接redis信息(如果确定部署类型，那么将连接信息放到core项目中)
            ---application.properties
 
二、测试
 
模拟1秒内100个线程请求接口，来测试结果是否正确。同时测试3中不同的锁:lock锁、trylock锁、注解锁。
 
1、lock锁
 
/**
* 模拟这个是商品库存
*/
public static volatile Integer TOTAL = 10;

@GetMapping("lock-decrease-stock")
public String lockDecreaseStock() throws InterruptedException {
    redissonLock.lock("lock", 10);
    if (TOTAL > 0) {
        TOTAL--;
    }
    Thread.sleep(50);
    log.info("======减完库存后,当前库存===" + TOTAL);
    //如果该线程还持有该锁，那么释放该锁。如果该线程不持有该锁，说明该线程的锁已到过期时间，自动释放锁
    if (redissonLock.isHeldByCurrentThread("lock")) {
        redissonLock.unlock("lock");
    }
    return "=================================";
}
 
压测结果
 
 没问题，不会超卖！
 
2、tryLock锁
 
/**
* 模拟这个是商品库存
*/
public static volatile Integer TOTAL = 10;

@GetMapping("trylock-decrease-stock")
public String trylockDecreaseStock() throws InterruptedException {
    if (redissonLock.tryLock("trylock", 10, 5)) {
        if (TOTAL > 0) {
            TOTAL--;
        }
        Thread.sleep(50);
        redissonLock.unlock("trylock");
        log.info("====tryLock===减完库存后,当前库存===" + TOTAL);
    } else {
        log.info("[ExecutorRedisson]获取锁失败");
    }
    return "===================================";
}
 
测试结果
 
 没有问题 ，不会超卖！
 
3、注解锁
 
/**
* 模拟这个是商品库存
*/
public static volatile Integer TOTAL = 10;

@GetMapping("annotatin-lock-decrease-stock")
@DistributedLock(value="goods", leaseTime=5)
public String lockDecreaseStock() throws InterruptedException {
    if (TOTAL > 0) {
        TOTAL--;
    }
    log.info("===注解模式=== 减完库存后,当前库存===" + TOTAL);
    return "=================================";
}
 
测试结果
 
 没有问题 ，不会超卖！
 
通过实验可以看出，通过这三种模式都可以实现分布式锁，然后呢？哪个最优。
 
三、三种锁的选择
 
观点 最完美的就是lock锁，因为
 
1、tryLock锁是可能会跳过减库存的操作，因为当过了等待时间还没有获取锁，就会返回false,这显然很致命！
2、注解锁只能用于方法上，颗粒度太大，满足不了方法内加锁。
 
1、lock PK tryLock 性能的比较
 
模拟5秒内1000个线程分别去压测这两个接口，看报告结果！
 
1）lock锁
 
压测结果 1000个线程平均响应时间为31324。吞吐量 14.7/sec
 
 
2）tryLock锁
 
压测结果 1000个线程平均响应时间为28628。吞吐量 16.1/sec
 
 这里只是单次测试，有很大的随机性。从当前环境单次测试来看，tryLock稍微高点。
 
2、常见异常 attempt to unlock lock, not ······
 
在使用RedissonLock锁时，很容易报这类异常，比如如下操作
 
//设置锁1秒过去
redissonLock.lock("redisson", 1);
/**
* 业务逻辑需要咨询2秒
*/
redissonLock.release("redisson");
 
上面在并发情况下就会这样
 
 造成异常原因：
 
线程1 进来获得锁后，但它的业务逻辑需要执行2秒，在线程1执行1秒后，这个锁就自动过期了，那么这个时候 线程2进来了获得了锁。在线程1去解锁就会抛上面这个异常（因为解锁和当前锁已经不是同一线程了）
 
所以我们需要注意，设置锁的过期时间不能设置太小，一定要合理，宁愿设置大点。
 
正对上面的异常，可以通过isHeldByCurrentThread()方法，
 
//如果为false就说明该线程的锁已经自动释放，无需解锁
if (redissonLock.isHeldByCurrentThread("lock")) {
    redissonLock.unlock("lock");
}
 
好了，这篇博客就到这了！
 
解决SpringBoot项目repackage failed: Unable to find main class Maven打包 install的问题
 
部署这个项目的时候遇到的问题：
 
一个通用的工具类项目，在进行打包时，出现repackage failed: Unable to find main class提示找不到主类问题，这个项目是一个通用的工具类，不是一个SpringBoot Web工程，根本启动类哪里来的main方法，到底是什么原因导致这个问题，后来看了一下依赖的父项目的pom文件，
 
发现父项目有一个打包的插件
 
<build>
    <plugins>
        <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
            <executions>
                <execution>
                    <goals>
                        <goal>repackage</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>
 
这时，问题就出现了，报打包失败错误！！！
 
解决：如果你的项目是一个放置通用工具类的工程，那么该项目中，就不能包括上面这个打包插件，如果你这个工具类工程依赖有父工程，那么父工程中也不能包括该打包插件，只有你的项目是一个web项目时，含有Main方法的程序入口类，要加该打包插件，放在了父工程的pom文件中，那就是代表了所有子模块都有这个打包插件，所以报错，解决就是去掉这个插件 ，只在web工程中加入这个打包插件！
 
代码地址：https://github.com/kongfanyu/spring-boot-distributed-redisson