【Redis】之 Redisson 分布式锁

骑猪看日落 2022-12-16 09:27 206阅读 0赞

# 三、Redisson #

--------------------

### 1、分布式锁原理 ###

分布式锁解决的问题是：在分布式系统下，能够对所有不同服务器下的服务进行加锁。对于本地锁而已，本地锁由于是运行在自己服务进程上的锁，只能对自己的服务请求进行加锁；而分布式锁则可以保证所有服务只有一个服务可以加锁。

在 Redis 中原始的加锁的方式是通过 `NX（数据唯一）` 和 `EX（设置过期时间）` 这两个参数实现的。`NX` 可以保证缓存数据唯一，设置成功了则表示获得锁；`EX` 则可以避免死锁问题。

分布式锁加锁流程如下：

# 所有服务进行抢锁
    set lock 123456 NX EX 30
    
    # 获得锁的服务执行完业务后释放锁
    del lock

当然，在分布式系统下，lock 的值需要设置为随机值，并且加锁和释放锁的操作都必须是原子操作的，所以转换为 Java 代码如下：

public String getRedisLock() { 
    	// 1、占分布式锁，去redis占坑，设置过期时间必须和加锁是同步的，保证原子性（避免死锁）
        String uuid = UUID.randomUUID().toString();
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", uuid,300,TimeUnit.SECONDS);
        if (lock) { 
            System.out.println("获取分布式锁成功...");
            try { 
                // 加锁成功...执行业务
                return "finish！";
            } finally { 
                // 执行 lua 脚本释放锁来保证释放锁是原子性的
                String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
                stringRedisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class), Arrays.asList("lock"), uuid);
            }
        } else { 
            System.out.println("获取分布式锁失败...等待重试...");
            // 加锁失败...重试机制，休眠一百毫秒
            try {  TimeUnit.MILLISECONDS.sleep(100); } catch (InterruptedException e) {  e.printStackTrace(); }
            // 自旋的方式
            return getRedisLock();     
        }
    }

但项目一般不会使用这种原生复杂的方式进行分布式加锁，一般都是使用 Redisson 来实现分布式加锁。

### 2、Redisson 简介 ###

Redisson 是一个 Redis 客户端，只是 Redisson 功能更强大，可以使用 Redisson 来实现 Redis 的分布式锁。

更多关于 Redisson 介绍可以参考官方文档：[https://github.com/redisson/redisson/wiki/%E7%9B%AE%E5%BD%95][https_github.com_redisson_redisson_wiki_E7_9B_AE_E5_BD_95]

### 3、项目整合 Redisson ###

1）、引入依赖

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

2）、配置 Redisson 客户端对象

@Configuration
    public class MyRedissonConfig { 
    
        /** * 所有对Redisson的使用都是通过RedissonClient * @return * @throws IOException */
        @Bean(destroyMethod="shutdown")
        public RedissonClient redisson() throws IOException { 
            // 1、创建配置
            Config config = new Config();
            // Redis url should start with redis:// or rediss://
            config.useSingleServer().setAddress("redis://192.168.56.10:6379");f
    
            // 2、根据Config创建出RedissonClient实例
            RedissonClient redissonClient = Redisson.create(config);
            return redissonClient;
        }
    
    }

配置好后就可以通过依赖注入方式创建 Redisson 客户端对象：

@Autowired
    private RedissonClient redissonClient;

### 4、Redisson 锁 ###

**1）、普通加锁**

@ResponseBody
        @GetMapping(value = "/hello")
        public String hello() { 
    
            // 1、获取一把锁，只要锁的名字一样，就是同一把锁
            RLock myLock = redisson.getLock("my-lock");
    
            // 2、加锁，阻塞式等待，默认加的锁都是30s
            myLock.lock(); 
            try { 
                System.out.println("加锁成功，执行业务..." + Thread.currentThread().getId());
                try {  TimeUnit.SECONDS.sleep(20); } 
                catch (InterruptedException e) {  e.printStackTrace(); }
            } catch (Exception ex) { 
                ex.printStackTrace();
            } finally { 
                // 3、解锁 假设解锁代码没有运行，Redisson会不会出现死锁
                System.out.println("释放锁..." + Thread.currentThread().getId());
                myLock.unlock();
            }
    
            return "hello";
        }

在 Redisson 的加锁中，不同担心出现死锁问题，哪怕加锁后没有释放锁也不会进入死锁状态，因为锁都是默认设置为 30 秒的过期时间：

*  1）、锁的自动续期，如果业务超长，运行期间自动锁上新的30s，不用担心业务时间长，锁自动过期被删掉；
 *  2）、加锁的业务只要运行完成，就不会给当前锁续期，即使不手动解锁，锁默认会在30s内自动过期，不会产生死锁问题

设置锁的到期时间：

// 设置10秒钟自动解锁,自动解锁时间一定要大于业务执行时间
    myLock.lock(10,TimeUnit.SECONDS);

问题：在锁时间到了以后，不会自动续期

*  1）、如果我们传递了锁的超时时间，就会发送给 redis 执行脚本，进行占锁，默认超时就是我们制定的时间；
 *  2）、如果我们指定锁的超时时间，就使用 lockWatchdogTimeout = 30 \* 1000 【看门狗默认时间】，只要占锁成功，就会启动一个定时任务【重新给锁设置过期时间，新的过期时间就是看门狗的默认时间】，每隔10秒都会自动的再次续期，续成30秒：internalLockLeaseTime =【看门狗时间】 / 3 = 10s

**2）、读写锁**

// 测试写锁
    public String writeValue() { 
        String s = "";
        RReadWriteLock readWriteLock = redisson.getReadWriteLock("rw-lock"); // 构造读写锁
        RLock rLock = readWriteLock.writeLock(); // 获取写锁
        try { 
            //1、改数据加写锁，读数据加读锁
            rLock.lock();
            s = UUID.randomUUID().toString();
            ValueOperations<String, String> ops = stringRedisTemplate.opsForValue();
            ops.set("writeValue",s);
            TimeUnit.SECONDS.sleep(10);
        } catch (InterruptedException e) { 
            e.printStackTrace();
        } finally { 
            rLock.unlock();
        }
    
        return s;
    }
    
    // 测试读锁
    public String readValue() { 
        String s = "";
        RReadWriteLock readWriteLock = redisson.getReadWriteLock("rw-lock");
        
        RLock rLock = readWriteLock.readLock();
        try { 
            // 加读锁
            rLock.lock();
            ValueOperations<String, String> ops = stringRedisTemplate.opsForValue();
            s = ops.get("writeValue");
            try {  TimeUnit.SECONDS.sleep(10); } catch (InterruptedException e) {  e.printStackTrace(); }
        } catch (Exception e) { 
            e.printStackTrace();
        } finally { 
            rLock.unlock();
        }
    
        return s;
    }

读写锁保证一定能读到最新数据，修改期间，写锁是一个排它锁（互斥锁、独享锁），读锁是一个共享锁，写锁没释放读锁必须等待：

*  读锁 + 读操作：相当于无锁，并发读，只会在Redis中记录好，所有当前的读锁，他们都会同时加锁成功；
 *  写锁 + 读操作：必须等待写锁释放；
 *  写锁 + 写操作：阻塞方式；
 *  读锁 + 写操作：有读锁，写也要等待。

总结就是：加读锁，可以读不能写；加写锁，既不能读也不能写。

**3）、信号量**

/** * 信号量也可以做分布式限流 */
    public String park() throws InterruptedException { 
    
        // 构建信号量
        RSemaphore park = redisson.getSemaphore("park");
        
        // 获取一个信号、获取一个值，redis 中的 park 值减一
        park.acquire();   				 // 阻塞 
        boolean flag = park.tryAcquire(); // 非阻塞
    
        if (flag) { 
            //执行业务
        } else { 
            return "error";
        }
    
        return "ok=>" + flag;
    }
    
    public String go() { 
        RSemaphore park = redisson.getSemaphore("park");
        park.release();     // 释放一个信号量，redis 中的 park 值加一
        return "ok";
    }

**4）、闭锁**

/** * 放假、锁门 * 1班没人了 * 5个班，全部走完，我们才可以锁大门 * 分布式闭锁 */
    public String lockDoor() throws InterruptedException { 
    
        RCountDownLatch door = redisson.getCountDownLatch("door");
        door.trySetCount(5);
        door.await();       //等待闭锁完成
    
        return "放假了...";
    }
    
    public String gogogo(@PathVariable("id") Long id) { 
        RCountDownLatch door = redisson.getCountDownLatch("door");
        door.countDown();       //计数-1
    
        return id + "班的人都走了...";
    }

更多关于锁的资料可以参考官方文档：[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]。

### 5、缓存一致性问题 ###

缓存一致性问题即：缓存里的数据如何与数据库保持一致？

解决方案有：

*  双写模式：改完数据库的同时更新缓存中对应的数据。
 *  失效模式：修改完数据库的数据时把缓存中对应的数据删除，下次请求该数据时就会主动从数据库中查询并更新到缓存中。

这两个方案在高并发的情况下都会出现脏数据的问题：

*  双写模式：在1号机器写完数据库但还没来得及更新缓存数据过程中，2号机器对数据进行了读取，这样就出现了数据不一致而出现了脏数据。但这是暂时性的脏数据问题，当主动更新完缓存之后或缓存失效主动从数据库中查询时就可以得到最新的正确数据了。
 *  失效模式：在1号机器修改完数据库但还没有来得及删除缓存中的数据时，2号机器对数据进行了读取，这时还是出现了数据不一致问题，同样，这种也是暂时性的问题。

我们系统使用的一致性解决方案为：

*  1、缓存的所有数据都有过期时间，数据过期下一次查询就会触发主动更新
 *  2、读写数据的时候，加上分布式的读写锁

一般来讲，对于实时性和一致性要求高的数据不应该放入缓存中。

[https_github.com_redisson_redisson_wiki_E7_9B_AE_E5_BD_95]: https://github.com/redisson/redisson/wiki/%E7%9B%AE%E5%BD%95
[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