分布式锁全家桶

Redis

1. 基本实现

借助于redis中的命令setnx(key, value)，key不存在就新增，存在就什么都不做。同时有多个客户端发送setnx命令，只有一个客户端可以成功，返回1（true）；其他的客户端返回0（false）。

1. 多个客户端同时获取锁（setnx）

2. 获取成功，执行业务逻辑，执行完成释放锁（del）

3. 其他客户端等待重试


@Service
public class StockService {
@Autowired
private StockMapper stockMapper;
@Autowired
private LockMapper lockMapper;
@Autowired
private StringRedisTemplate redisTemplate;
 public void checkAndLock() {
    // 加锁，获取锁失败重试
     while (!this.redisTemplate.opsForValue().setIfAbsent("lock","xxx")){
      try {
        // 避免栈溢出
        Thread.sleep(100);
         
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
     }
         // 先查询库存是否充足
     Stock stock = this.stockMapper.selectById(1L);
     // 再减库存
     if (stock != null && stock.getCount() > 0){
     stock.setCount(stock.getCount() - 1);
     this.stockMapper.updateById(stock);
      }
      // 释放锁
     this.redisTemplate.delete("lock");
 }
}

2. 防死锁

解决：给锁设置过期时间，自动释放锁。

设置过期时间两种方式：

1. 通过expire设置过期时间（缺乏原子性：如果在setnx和expire之间出现异常，锁也无法释放）

2. 使用set指令设置过期时间：set key value ex 3 nx（既达到setnx的效果，又设置了过期时间）

问题：可能会释放其他服务器的锁。

场景：如果业务逻辑的执行时间是7s。执行流程如下

1. index1业务逻辑没执行完，3秒后锁被自动释放。index2请求就会执行

2. index2获取到锁，执行业务逻辑，3秒后锁被自动释放。index3请求就会执行

3. index3获取到锁，执行业务逻辑

4. index1业务逻辑执行完成，开始调用del释放锁，这时释放的是index3的锁，导致index3的业务只执行1s就被别人释放。最终等于没锁的情况。

解决：setnx获取锁时，设置一个指定的唯一值（例如：uuid）；释放前获取这个值，判断是否自己的锁

3. 防误删

实现如下：

问题：删除操作缺乏原子性。

场景：

1. index1执行删除时，查询到的lock值确实和uuid相等

2. index1执行删除前，lock刚好过期时间已到，被redis自动释放

3. index2获取了lock

4. index1执行删除，此时会把index2的lock删除

解决方案：没有一个命令可以同时做到判断 + 删除，所有只能通过其他方式实现（LUA脚本）

4. redis中的lua脚本保证原子性

1. 现实问题

redis采用单线程架构，可以保证单个命令的原子性，但是无法保证一组命令在高并发场景下的原子性。

如果redis客户端通过lua脚本把3个命令一次性发送给redis服务器，那么这三个指令就不会被其他客户端指令打断。Redis 也保证脚本会以原子性(atomic)的方式执行：当某个脚本正在运行的时候，不会有其他脚本或 Redis 命令被执行。这和使用 MULTI/ EXEC 包围的事务很类似。

但是MULTI/ EXEC方法来使用事务功能，将一组命令打包执行，无法进行业务逻辑的操作。这期间有某一条命令执行报错（例如给字符串自增），其他的命令还是会执行，并不会回滚。


  public void checkAndLock() {
        // 加锁，获取锁失败重试
        String uuid = UUID.randomUUID().toString();
        while (!this.redisTemplate.opsForValue().setIfAbsent("lock", uuid, 3,
                TimeUnit.SECONDS)) {
            try {
                Thread.sleep(50);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 先查询库存是否充足
        Stock stock = this.stockMapper.selectById(1L);
         // 再减库存
        if (stock != null && stock.getCount() > 0) {
            stock.setCount(stock.getCount() - 1);
            this.stockMapper.updateById(stock);
        }
        // 释放锁
        //redis.call('get', KEYS[1]) 获取 lock 值 uuid
        //KEYS[1]  取出KEYS列 第一个值  lua 索引从1开始
        //ARGV[1]   ARGV 第一个值
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end ";
        //执行lua脚本
        //参数1 执行脚本, Long.class 返回值类型,如果不指定会报异常
        //参数二 KEYS 列表
        //参数三 ARGV 不固定参数 ...object
        this.redisTemplate.execute(new DefaultRedisScript<>(script, Long.class), Arrays.asList("lock"), uuid);
    }

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），从而让使用者能够将精力更集中地放在处理业务逻辑上。

官方文档地址：https://github.com/redisson/redisson/wiki

1. 可重入锁（Reentrant Lock）拼运气

基于Redis的Redisson分布式可重入锁 RLock Java对象实现了 java.util.concurrent.locks.Lock 接口。

大家都知道，如果负责储存这个分布式锁的Redisson节点宕机以后，而且这个锁正好处于锁住的状态时，这个锁会出现锁死的状态。为了避免这种情况的发生，Redisson内部提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期。默认情况下，看门狗检查锁的超时时间是30秒钟，也可以通过修改 Config.lockWatchdogTimeout 来另行指定。 RLock 对象完全符合Java的Lock规范。也就是说只有拥有锁的进程才能解锁，其他进程解锁则会抛出 IllegalMonitorStateException 错误。另外Redisson还通过加锁的方法提供了 leaseTime 的参数来指定加锁的时间。超过这个时间后锁便自动解开了。


public void deduct() {
        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();
            }
        }
    }

watch dog 的自动延期机制


private void redissonDoc() throws InterruptedException {
    //1. 普通的可重入锁
    RLock lock = redissonClient.getLock("generalLock");
 
    // 拿锁失败时会不停的重试
    // 具有Watch Dog 自动延期机制 默认续30s 每隔30/3=10 秒续到30s
    lock.lock();
 
    // 尝试拿锁10s后停止重试,返回false
    // 具有Watch Dog 自动延期机制 默认续30s
    boolean res1 = lock.tryLock(10, TimeUnit.SECONDS);
 
    // 拿锁失败时会不停的重试
    // 没有Watch Dog ，10s后自动释放
    lock.lock(10, TimeUnit.SECONDS);
 
    // 尝试拿锁100s后停止重试,返回false
    // 没有Watch Dog ，10s后自动释放
    boolean res2 = lock.tryLock(100, 10, TimeUnit.SECONDS);
 
    //2. 公平锁 保证 Redisson 客户端线程将以其请求的顺序获得锁
    RLock fairLock = redissonClient.getFairLock("fairLock");
 
    //3. 读写锁 没错与JDK中ReentrantLock的读写锁效果一样
    RReadWriteLock readWriteLock = redissonClient.getReadWriteLock("readWriteLock");
    readWriteLock.readLock().lock();
    readWriteLock.writeLock().lock();
}

如何启动Redisson的看门狗机制

如果你想让Redisson启动看门狗机制，你就不能自己在获取锁的时候，定义超时释放锁的时间，无论，你是通过lock() （void lock(long leaseTime, TimeUnit unit);）还是通过tryLock获取锁，只要在参数中，不传入releastime，就会开启看门狗机制，
就是这两个方法不要用： boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException
和 void lock(long leaseTime, TimeUnit unit);，因为它俩都传release

1. 引入依赖


 
        <dependency>
            <groupId>org.redissongroupId>
            <artifactId>redissonartifactId>
            <version>3.16.1version>
        dependency>

2. 添加配置


@Configuration
public class RedissonConfig {
  @Bean
  public RedissonClient redissonClient(){
   //初始化一个配置对象
   Config config = new Config();
   // 可以用"rediss://"来启用SSL连接
   //useSingleServer 单机模式,还有其他模式
   config.useSingleServer().setAddress("redis://172.16.116.100:6379");
   return Redisson.create(config);
 }
}

3. 代码中使用


@Autowired
private RedissonClient redissonClient;
public void checkAndLock() {
  // 加锁，获取锁失败重试
  RLock lock = this.redissonClient.getLock("lock");
  lock.lock();
  // 先查询库存是否充足
  Stock stock = this.stockMapper.selectById(1L);
  // 再减库存
  if (stock != null && stock.getCount() > 0){
  stock.setCount(stock.getCount() - 1);
  this.stockMapper.updateById(stock);
  }
  // 释放锁
  lock.unlock();
}

.2. 公平锁（Fair Lock）拼手速,拼网速

基于Redis的Redisson分布式可重入公平锁也是实现了 java.util.concurrent.locks.Lock 接口的一种 RLock 对象。同时还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。它保证了当多个Redisson客户端线程同时请求加锁时，优先分配给先发出请求的线程。所有请求线程会在一个队列中排队，当某个线程出现宕机时，Redisson会等待5秒后继续下一个线程，也就是说如果前面有5个线程都处于等待状态，那么后面的线程会等待至少25秒。


 public void deduct(){
 
        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();
 
 
 
    }

.3. 联锁（MultiLock）

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

首先要有三个redis服务器

不建议使用,有一台redis宕机就上锁失败


public void deduct(){
 
        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();
        
    }

4. 红锁（RedLock）(了解)

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

半数以上加锁成功,就成功


public void deduct(){
 
        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();
 
 
    }

.5. 读写锁（ReadWriteLock）

基于Redis的Redisson分布式可重入读写锁 RReadWriteLock Java对象实现了 java.util.concurrent.locks.ReadWriteLock 接口。其中读锁和写锁都继承了RLock接口。

分布式可重入读写锁允许同时有多个读锁和一个写锁处于加锁状态。


public void deduct(){
 
        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();
 
 
 
    }

写-写不可以并发

读-写-不可以并发

读-读-可以并发

添加StockController方法：


@GetMapping("test/read")
public String testRead(){
String msg = stockService.testRead();
return "测试读";
}
 
@GetMapping("test/write")
public String testWrite(){
String msg = stockService.testWrite();
return "测试写";
}

添加StockService方法：


public String testRead() {
  RReadWriteLock rwLock = this.redissonClient.getReadWriteLock("rwLock");
  rwLock.readLock().lock(10, TimeUnit.SECONDS);
  System.out.println("测试读锁。。。。");
  // rwLock.readLock().unlock();
  return null;
}
 
public String testWrite() {
  RReadWriteLock rwLock = this.redissonClient.getReadWriteLock("rwLock");
  rwLock.writeLock().lock(10, TimeUnit.SECONDS);
  System.out.println("测试写锁。。。。");
  // rwLock.writeLock().unlock();
  return null;
}

打开开两个浏览器窗口测试：

同时访问写：一个写完之后，等待一会儿（约10s），另一个写开始

同时访问读：不用等待

先写后读：读要等待（约10s）写完成

先读后写：写要等待（约10s）读完成

6. 信号量（Semaphore）(请求资源做限流)

基于Redis的Redisson的分布式信号量（Semaphore）Java对象 RSemaphore 采用了与 java.util.concurrent.Semaphore 相似的接口和用法。同时还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。


RSemaphore semaphore = redisson.getSemaphore("semaphore");
semaphore.acquire();
semaphore.release();

在StockController添加方法：


@GetMapping("test/semaphore")
  public String testSemaphore(){
  this.stockService.testSemaphore();
  return "测试信号量";
}

在StockService添加方法：


public void testSemaphore() {
//初始化对象semaphore作为key存放在redis中
  RSemaphore semaphore = this.redissonClient.getSemaphore("semaphore");
//限制请求量个数,redis一旦申明,在想修改需要手动删掉 redis 中的semaphore
  semaphore.trySetPermits(3);
  try {
//获取资源,获取成功的会继续处理业务,否则会被阻塞住
  semaphore.acquire();
 
  TimeUnit.SECONDS.sleep(5);
  System.out.println(System.currentTimeMillis());
 
//释放资源
  semaphore.release();
  } catch (InterruptedException e) {
  e.printStackTrace();
  }
}

添加测试用例：并发10次，循环一次

控制台效果：


控制台1：
1606960790234
1606960800337
1606960800443
1606960805248
控制台2：
1606960790328
1606960795332
1606960800245
控制台3：
1606960790433
1606960795238
1606960795437

由此可知：

1606960790秒有3次请求进来：每个控制台各1次

1606960795秒有3次请求进来：控制台2有1次，控制台3有2次

1606960800秒有3次请求进来：控制台1有2次，控制台2有1次

1606960805秒有1次请求进来：控制台1有1次

可过期信号量(用处不大)

7. 闭锁（CountDownLatch）(到计数器)

数字减为零才执行

适合:一个事件等待一组事件执行结束后在执行

基于Redisson的Redisson分布式闭锁（CountDownLatch）Java对象 RCountDownLatch 采用了与 java.util.concurrent.CountDownLatch 相似的接口和用法。


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

需要两个方法：一个等待，一个计数countDown

给StockController添加测试方法：


@GetMapping("test/latch")
public String testLatch(){
  this.stockService.testLatch();
  return "班长锁门。。。";
}
 
@GetMapping("test/countdown")
  public String testCountDown(){
  this.stockService.testCountDown();
  return "出来了一位同学";
}

给StockService添加测试方法：


public void testLatch() {
  RCountDownLatch latch = this.redissonClient.getCountDownLatch("latch");
  latch.trySetCount(6);
  try {
//阻塞,当latch的值是0的时候才会放行
  latch.await();
  } catch (InterruptedException e) {
  e.printStackTrace();
  }
}
 
public void testCountDown() {
  RCountDownLatch latch = this.redissonClient.getCountDownLatch("latch");
 //减1
  latch.countDown();
}

启测试，打开两个页面：当第二个请求执行6次之后，第一个请求才会执行。

相关阅读:
hbase操作学习
 shell编程扩展-----bash编程
 How to resolve jre-openjdk and jre-openjdk-headless conflicts?
【CCNA实验分享】三层交换机Vlan间路由
 Netty入门指南之基础介绍
 GBASE 8s dbspace配置参数
 使用IntelliJ IDEA创建Maven项目并上传项目至gitlab等远程仓库
 shiro会话管理
 Vue2、Vue3知识总结---完整版✨
算法-链表-合并两个升序链表
原文地址：https://blog.csdn.net/yuzheh521/article/details/126675883