springboot2.x整合Redission_springboot redission

一、概述

从 spring-boot 2.x 版本开始，spring-boot-data-redis 默认使用 Lettuce 客户端操作数据。

Reddissin也是一个redis客户端，其在提供了redis基本操作的同时，还具备其他客户端一些不具备的高精功能，例如 限流 分布式锁+看门狗、分布式限流、远程调用等等

二、接入Spring-Boot项目

引入依赖

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.13.6</version>
</dependency>
1
2
3
4
5

注意:引入次依赖后，无需再引入spring-boot-starter-data-redis,其redisson-spring-boot-starter内部已经进行了引入，且排除了Redis的Luttuce 以及 Jedis客户端

redission配置

引入配置可使用程序化配置以及YML配置两种方式

程序化配置

@Configuration
public class RedissonConfig {
    @Bean
    public RedissonClient redissonClient() throws IOException {
        // 默认连接地址 127.0.0.1:6379
        RedissonClient redisson = Redisson.create();

        Config config = new Config();
        config.useSingleServer().setAddress("myredisserver:6379");
        return Redisson.create(config);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12

YML配置

我这里仅列出单节点配置，集群或者哨兵模式请访问我yml中的官网

# redis单节点配置方式
singleServerConfig:
  # 连接空闲超时，单位：毫秒
  idleConnectionTimeout: 10000
  # 连接超时，单位：毫秒
  connectTimeout: 10000
  # 命令等待超时，单位：毫秒 默认3000
  timeout: 3000
  # 命令失败重试次数
  retryAttempts: 3
  # 命令重试发送时间间隔，单位：毫秒
  retryInterval: 1500
  # 无密码则设置 null
  password: "123456a"
  # 单个连接最大订阅数量
  subscriptionsPerConnection: 5
  # 客户端名称
  clientName: null
  # redis 节点地址
  address: "redis://127.0.0.1:6379"
  # 从节点发布和订阅连接的最小空闲连接数
  subscriptionConnectionMinimumIdleSize: 1
  # 发布和订阅连接池大小
  subscriptionConnectionPoolSize: 50
  # 发布和订阅连接的最小空闲连接数
  connectionMinimumIdleSize: 32
  # 发布和订阅连接池大小
  connectionPoolSize: 64
  # 数据库编号
  database: 10
  # DNS监测时间间隔，单位：毫秒  在启用该功能以后，Redisson将会监测DNS的变化情况
  dnsMonitoringInterval: 5000
threads: 0
nettyThreads: 0
codec: !<org.redisson.codec.JsonJacksonCodec> {}
transportMode: "NIO"

# 官方文档:https://github.com/redisson/redisson/wiki/2.-%E9%85%8D%E7%BD%AE%E6%96%B9%E6%B3%95#26-%E5%8D%95redis%E8%8A%82%E7%82%B9%E6%A8%A1%E5%BC%8F
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

加载配置文件

@Configuration
public class RedissonConfig {
    @Bean
    public RedissonClient redissonClient() throws IOException {
        Config config = Config.fromYAML(RedissonConfig.class.getClassLoader().getResource("redission-config.yml"));
        return Redisson.create(config);
    }
}
1
2
3
4
5
6
7
8

---

在项目使用Redission时，我们一般会使用 RedissonClient 进行数据操作,但有朋友或许觉得RedissonClient操作不方便，或者更喜欢使用 RedisTemplate进行操作，其实这两者也是可以共存的，我们只需要再定义RedisTemplate的配置类即可.

项目中同时使用RedisTemplate

我们只需要定义一个配置类，且创建RedisTemplate自定义配置Bean即可

package com.leilei.config;

import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

/**
 * @author lei
 * @create 2021-09-09 15:19
 * @desc redis配置
 **/
@Configuration
@AutoConfigureAfter(RedisAutoConfiguration.class)
public class RedisConfig {
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
        //deBug 会发现其 connectionFactory 实例为 redission
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(connectionFactory);
        redisTemplate.setDefaultSerializer(new GenericJackson2JsonRedisSerializer());
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        redisTemplate.setKeySerializer(stringRedisSerializer);
        redisTemplate.setHashKeySerializer(stringRedisSerializer);
        return redisTemplate;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

发现项目引入Redission后，RedisTemplate底层所用的连接工厂也是Redission

三、限流

我们是有面临高并发下需要对接口或者业务逻辑限流的问题，我们可以采用Guaua依赖下的RateLimiter 实现，实际上，Redisssion也有类似的限流功能

RateLimiter 被称为令牌桶限流，此类限流是首先定义好一个令牌桶，指明在一定时间内生成多少个令牌，每次访问时从令牌桶获取指定数量令牌，如果获取成功，则设为有效访问。

Redission中的令牌桶限流使用:

1.获取限流实例

需要使用redissonClient 来获取一个RRateLimiter限流实例

    /**
     * Returns rate limiter instance by <code>name</code>
     * 
     * @param name of rate limiter
     * @return RateLimiter object
     */
    RRateLimiter getRateLimiter(String name);
1
2
3
4
5
6
7

2.设置令牌桶规则

设置令牌桶规则，例如 1分钟秒内，生成6个有效令牌

   /**
     * Updates RateLimiter's state and stores config to Redis server.
     *
     * @param mode - rate mode
     * @param rate - rate
     * @param rateInterval - rate time interval
     * @param rateIntervalUnit - rate time interval unit
     */
    void setRate(RateType mode, long rate, long rateInterval, RateIntervalUnit rateIntervalUnit);
1
2
3
4
5
6
7
8
9

3.对限流的业务进行令牌获取尝试

// 尝试获取令牌 底层默认是获取一个令牌
boolean tryAcquire();

// 尝试获取指定令牌
boolean tryAcquire(long permits);

// 一定时间内尝试获取1个令牌
boolean tryAcquire(long timeout, TimeUnit unit);
1
2
3
4
5
6
7
8

Redission 限流底层是使用的Lua脚本

4.限流实战

在我们定义好限流实例与设置好限流规则后启动项目，发现指定数据库存在了一个缓存KEY,KEY的名字就是我们设置的限流实例名字

RateType.OVERALL 表示针对所有客户端

RRateLimiter rateLimiter;  
@PostConstruct
public void initRateLimiter(){
    RRateLimiter ra = redissonClient.getRateLimiter("rate-limiter");
    ra.setRate(RateType.OVERALL, 6, 1, RateIntervalUnit.MINUTES);
    rateLimiter = ra;
}
---------
@GetMapping("/rate/limiter")
public String testRateLimiter() {
    return lockService.testRateLimiter();
}
---------
public String testRateLimiter() {
    boolean b = rateLimiter.tryAcquire();
    if (b) {
        return "ok";
    }
    return "fail";
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

接着，我们来模拟并发访问

页面访问url

http://localhost:8080/rate/limiter
1

redis生成了两个新的缓存KEY,一个是Zset数据类型，一个是String类型

Zset类型缓存{限流名}:permits里存储的是每一次访问时间

String类型缓存{限流名}:value缓存的是还剩余可访问次数

当我们连续访问六次都返回ok，第七次访问时，我的接口返回了fall，此时查看我们的缓存KEY,发现Zset集合存在六个访问

5.规则设置注意事项

规则设置有以下两种模式:

    /**
     * Updates RateLimiter's state and stores config to Redis server.
     *
     * @param mode - rate mode
     * @param rate - rate
     * @param rateInterval - rate time interval
     * @param rateIntervalUnit - rate time interval unit
     */
    void setRate(RateType mode, long rate, long rateInterval, RateIntervalUnit rateIntervalUnit);

    /**
     * Initializes RateLimiter's state and stores config to Redis server.
     * 
     * @param mode - rate mode
     * @param rate - rate
     * @param rateInterval - rate time interval
     * @param rateIntervalUnit - rate time interval unit
     * @return {@code true} if rate was set and {@code false}
     *         otherwise
     */
    boolean trySetRate(RateType mode, long rate, long rateInterval, RateIntervalUnit rateIntervalUnit);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

区别:

// setRate 我们项目服务重启,就会强制重置之前的限流配置与状态,以当前为准
ra.setRate(RateType.OVERALL, 6, 1, RateIntervalUnit.MINUTES);

// trySetRate 我们项目服务重启,不会更新限流配置与限流状态,但参数更改后亦不会生效!比如之前是十分钟内颁布令牌100个，更改为5分钟内颁布令牌30个并不会生效
ra.trySetRate(RateType.OVERALL, 7, 2, RateIntervalUnit.MINUTES);
1
2
3
4
5

setRate 演示

快速访问三次

模拟我们服务器重启

trySetRate演示

设置两分钟后内7个有效令牌

ra.trySetRate(RateType.OVERALL, 7, 2, RateIntervalUnit.MINUTES);
1

我们快速访问几次，出发限流缓存

修改配置后，重新启动我们的项目

ra.trySetRate(RateType.OVERALL, 70, 1, RateIntervalUnit.MINUTES);
1

针对这两种特性，我们可以根据自己的需求进行选择!

四、分布式锁

有点经验的同学一提到使用分布式锁便联想到了redis，那redis如何实现分布式锁呢？

分布式锁本质上要实现的目标就是在Redis中占一个坑（简单的说，就是萝卜占坑的道理),当别的进程也要来占坑时，发现那个坑里已经有一个颗大萝卜时，就只好放弃或者稍后重试。

分布式锁常用手段：

1.使用setNx命令

这个命令的详细描述是（set if not exists)，如果指定key不存在则设置(成功占坑)，在业务执行完成后，调用del命令删该key（释放坑）

ex:

# set 锁名 值
setnx vehicle-lock  111

// dosoming

del  vehicle-lock
1
2
3
4
5
6

但这个命令存在一个问题，如果执行逻辑中出现问题，可能导致del指令无法执行，那么该锁就会成为死锁了。

可能有小伙伴贴心的想到了，我们可以给这个key再设置一个过期时间呀。

比如

setnx vehicle-lock  111

expire vehicle-lock  10

// dosoming

del  vehicle-batch
1
2
3
4
5
6
7

即使这样操作后，该逻辑仍有问题，由于setnx 与expire 是两条命令，如果在 setnx与 expire之间，redis服务器挂了，就会导致expire不会执行，从而过期时间设置失败，该锁仍会成为死锁

根源是 setnx与expire两条命令并不是原子命令

且redis的事物也无法解决 setnx 与expire的问题，因为expire是依赖于setnx的执行结果的，如果setnx没有成功，expire则不应该执行。事物又无法进行if else判断，，，顾 setnx +expire方式实现分布式锁，并不是优解

2.使用setNx Ex 命令

上方已经说了 setNx+ expire的问题，Redis官方为了解决这个问题，在2.8版本时引入了 set指令的扩展参数

使得 setnx 与 expire命令可以一起执行

ex:

# set 锁名 值 ex 过期时间（单位:秒) nx
set vehicle-lock 111 ex 5 nx
// doSomthing
del vehicle-lock
1
2
3
4

从逻辑上来讲，setNx Ex 已是优解了，不会使该分布式锁成为死锁

但在我们开发中，或许仍会出现问题，为什么呢？

由于我们一开始为此锁设置了一个过期时间，那假如我们的业务逻辑执行耗时超过了设置的过期时间呢？就会出现一个线程未执行完毕，第二个线程可能持有了这个分布式锁的情况。

所以呢，如果使用 setNx Ex 组合，必须要确保自己的锁的超时时间大于占锁后的业务执行时间

3.使用Redission

上方介绍的 setNx 与 setNx Ex 命令，都是Redis 服务器为我们提供的原生命令，也或多或少的存在着一部分问题，为解决setNx Ex命令存在着业务逻辑大于锁超时时间的问题，Redisson内部提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期。默认情况下，看门狗的检查锁的超时时间是30秒钟（就是续期30s)，也可以通过修改Config.lockWatchdogTimeout来另行指定，锁的初始过期时间默认也是30s

ex:

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

// 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
if (res) {
   try {
     ...
   } finally {
       lock.unlock();
   }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

尝试获取锁

尝试续期

分布式锁使用详情示例

获取到锁的执行，未获取到的放弃操作

public String threadLock() {
    RLock lock = redissonClient.getLock("vehicle-lock");
    System.out.println("当前线程:" + Thread.currentThread().getName());
    //加锁
    try {
        if (!lock.tryLock(1, TimeUnit.SECONDS)) {
            String message = "线程:" + Thread.currentThread().getName() + "未获取到锁,直接返回";
            System.out.println(message);
            return message;
        }
        System.out.println("线程:" + Thread.currentThread().getName() + "获取到锁!");
        Thread.sleep(4000_0);
        System.out.println("线程:" + Thread.currentThread().getName() + "业务结束!");
    } catch (InterruptedException e) {
        throw new RuntimeException(String.format("出现异常：%s",e.getMessage()));
    } finally {
        //判断 拿到了锁的才释放锁,否则会报错!
        if (lock.isLocked() && lock.isHeldByCurrentThread()) {
            System.out.println("线程:" + Thread.currentThread().getName() + "释放锁!");
            lock.unlock();
        }
    }
    return "线程:" + Thread.currentThread().getName() + "业务结束!";
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

访问接口后，快速查看redis，发现罗卜已成功占坑

由于我们设置了睡眠40s(模拟业务耗时大于分布式锁过期时间)，我们可以不断刷新缓存KEYvehicle-lock测试看门狗续期

当锁已占用10s，看门狗便会触发一次锁续期

整个流程

当前线程:http-nio-8080-exec-1
线程:http-nio-8080-exec-1获取到锁!
线程:http-nio-8080-exec-1业务结束!
线程:http-nio-8080-exec-1释放锁!
1
2
3
4

高并发模拟:

获取到锁的执行，未获取到待锁释放后再争抢获取锁执行

public String threadLock() {
    RLock lock = redissonClient.getLock("vehicle-lock");
    System.out.println("当前线程:" + Thread.currentThread().getName());
    //加锁
    lock.lock();
    try {
        System.out.println("线程:" + Thread.currentThread().getName() + "获取到锁!");
        Thread.sleep(10000);
        System.out.println("线程:" + Thread.currentThread().getName() + "业务结束!");
    } catch (InterruptedException e) {
        throw new RuntimeException(String.format("出现异常：%s",e.getMessage()));
    } finally {
        //判断 拿到了锁的才释放锁,否则会报错!
        if (lock.isLocked() && lock.isHeldByCurrentThread()) {
            System.out.println("线程:" + Thread.currentThread().getName() + "释放锁!");
            lock.unlock();
        }
    }
    return "线程:" + Thread.currentThread().getName() + "业务结束!";
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

Redission提供了多种类型的分布式锁，比如 可重入锁（Reentrant Lock）、公平锁（Fair Lock）、联锁（MultiLock）、红锁（RedLock）、 读写锁（ReadWriteLock）

公平锁使用详情示例

公平锁它保证了当多个Redisson客户端线程同时请求加锁时，优先分配给先发出请求的线程。所有请求线程会在一个队列中排队等待

    /**
     * Returns Lock instance by name.
     * <p>
     * Implements a <b>fair</b> locking so it guarantees an acquire order by threads.
     * <p>
     * To increase reliability during failover, all operations wait for propagation to all Redis slaves.
     * 
     * @param name - name of object
     * @return Lock object
     */
    RLock getFairLock(String name);
1
2
3
4
5
6
7
8
9
10
11

public String threadFairLock() {
    RLock lock = redissonClient.getFairLock("vehicle-fair-lock");
    System.out.println("当前线程:" + Thread.currentThread().getName());
    //加锁
    lock.lock();
    try {
        System.out.println("线程:" + Thread.currentThread().getName() + "获取到锁!");
        Thread.sleep(1000_0);
        System.out.println("线程:" + Thread.currentThread().getName() + "业务结束!");
    } catch (InterruptedException e) {
        throw new RuntimeException(String.format("出现异常：%s",e.getMessage()));
    } finally {
        //判断 拿到了锁的才释放锁,否则会报错!
        if (lock.isLocked() && lock.isHeldByCurrentThread()) {
            System.out.println("线程:" + Thread.currentThread().getName() + "释放锁!");
            lock.unlock();
        }
    }
    return "线程:" + Thread.currentThread().getName() + "业务结束!";
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

当前线程:http-nio-8080-exec-1
线程:http-nio-8080-exec-1获取到锁!
当前线程:http-nio-8080-exec-2
当前线程:http-nio-8080-exec-3
当前线程:http-nio-8080-exec-4
当前线程:http-nio-8080-exec-5
当前线程:http-nio-8080-exec-6
当前线程:http-nio-8080-exec-7
当前线程:http-nio-8080-exec-8
当前线程:http-nio-8080-exec-9
当前线程:http-nio-8080-exec-10
线程:http-nio-8080-exec-1业务结束!
线程:http-nio-8080-exec-1释放锁!
线程:http-nio-8080-exec-2获取到锁!
线程:http-nio-8080-exec-2业务结束!
线程:http-nio-8080-exec-2释放锁!
线程:http-nio-8080-exec-3获取到锁!
线程:http-nio-8080-exec-3业务结束!
线程:http-nio-8080-exec-3释放锁!
线程:http-nio-8080-exec-4获取到锁!
线程:http-nio-8080-exec-4业务结束!
线程:http-nio-8080-exec-4释放锁!
线程:http-nio-8080-exec-5获取到锁!
线程:http-nio-8080-exec-5业务结束!
线程:http-nio-8080-exec-5释放锁!
线程:http-nio-8080-exec-6获取到锁!
线程:http-nio-8080-exec-6业务结束!
线程:http-nio-8080-exec-6释放锁!
线程:http-nio-8080-exec-7获取到锁!
线程:http-nio-8080-exec-7业务结束!
线程:http-nio-8080-exec-7释放锁!
线程:http-nio-8080-exec-8获取到锁!
线程:http-nio-8080-exec-8业务结束!
线程:http-nio-8080-exec-8释放锁!
线程:http-nio-8080-exec-9获取到锁!
线程:http-nio-8080-exec-9业务结束!
线程:http-nio-8080-exec-9释放锁!
线程:http-nio-8080-exec-10获取到锁!
线程:http-nio-8080-exec-10业务结束!
线程:http-nio-8080-exec-10释放锁!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

我们可以根据自己的业务场景来灵活选择使用哪一种锁，以及是否让未获取到锁的线程等待执行或者放弃执行任务

Redission还有其他特别强大的功能，后续继续加更…