[版权申明] 非商业目的注明出处可自由转载
出自：shusheng007

概述

SpringBoot集成Redis步骤这块非常简单，没有太多内容，集成后如何使用的思路可以略微谈一谈，我认为这也算是集成的一部分吧。

Redis简述

官方介绍

The open source, in-memory data store used by millions of developers as a database, cache, streaming engine, and message broker.

Redis是一个基于内存的开源数据存储，被数百万开发者用做数据库，缓存，流引擎，以及消息中间件

Redis本质是一个基于内存的数据存储，但是随着它的发展野心也变得越来越大...但现目前在大部分时候其被用作数据库和缓存，另外两个有专门的工具负责，例如Kafka。Kafka太优秀了，以至于谁都要想来致敬它，前有Rocketmq,后有Redis Stream...

Redis服务端

由于Redis在提供了无与伦比的高吞吐量的同时，还提供了很多实用的数据结构，例如5大常用数据结构（string,list,set,hash,zset）。除此之外还有像geohash,bitmap,hyperloglog等，关键它还支持插件的功能，这样就可以很容易扩展新的功能了。

由于其基本已经大一统了分布式缓存这个领域，在微服务盛行的当下，导致每个项目几乎都会部署这货。所以它就有机会不断的扩展自己的功能，连消息中间件它都想染指...

关于Redis其实有很多内容，可以找相关资料进行学习

Redis Java客户端

我们在SpringBoot中使用Redis是通过Redis客户端的，目前较为流行的有两个：Jedis与Lettuce，现在SpringBoot默认使用Lettuce，建议在新项目使用Lettuce即可。但一般情况下我们会使用spring-data-redis，所以大部分时间我们都在使用RedisTemplate这个对内部客户端抽象了的组件。

SpringBoot整合Redis

记住SpringBoot整合第三方技术基本就是3步：引入依赖-配置-使用

引入依赖

在SpringBoot中我们一般会使用spring-data-redis,而不会直接使用Redis的客户端。直接使用可以吗？当然可以了，你比如说在访问数据库方面，MyBatis就比较强势，所以很多人不会去使用spring-data-jpa。

在pom文件中引入如下依赖，那个commons-pool2是为了使用连接池，如果不使用连接池也可以不配置。

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>

    <dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-pool2</artifactId>
    </dependency>
</dependencies>

配置

配置服务端连接信息

像这种要访问远端服务的都需要进行连接信息配置，例如你访问mysql，访问rabbitmq，访问kafka等等

在application.yml中配置如下信息，其中那个连接池配置需要引入commons-pool2才生效

spring:
  redis:
    #redis数据库，其有16个数据库
    database: 0
    #redis服务器地址
    host: localhost
    #redis服务器端口号
    port: 6379
    #连接池配置
    lettuce:
      pool:
        enabled: true
        max-active: 8
        max-wait: 10s

配置代码

如果我们要使用spring-data-redis给我们抽象出来的RedisTemplate，一般需要配置一下它的序列化器。因为我们要把Java对象保存到Redis中，所以保存的时候的告诉它以什么形式保存，例如以json的形式保存，那么我们就需要在保存的时候先把对象给序列化成json，然后从redis读取的时候再把json反序列化为java对象。

/**
 * Created by shusheng007
 *
 * @author benwang
 * @date 2022/10/7 11:50
 * @description:
 */

@Configuration
public class RedisConfig {
    @Bean
    public RedisTemplate<String,Object> redisTemplate(RedisConnectionFactory redisConnectionFactory){
        RedisTemplate<String,Object> template = new RedisTemplate<>();
        template.setConnectionFactory(redisConnectionFactory);

        StringRedisSerializer keySerializer = new StringRedisSerializer();
        GenericJackson2JsonRedisSerializer valueSerializer = new GenericJackson2JsonRedisSerializer();

        template.setKeySerializer(keySerializer);
        template.setValueSerializer(valueSerializer);
        template.setHashKeySerializer(keySerializer);
        template.setHashValueSerializer(valueSerializer);
        template.afterPropertiesSet();
        return template;
    }
}

一般情况下，我们会将key的序列化器设置为String序列化器，把value设置为Json的序列化器，如代码所示。

至此已经配置好了，可以愉快的使用RedisTemplate了。

使用

使用比较简单了，注入RedisTemplate然后调用其方法即可。

@Service
@Slf4j
@RequiredArgsConstructor
public class RedisOpsService {
    private final RedisTemplate<String, Object> redisTemplate;

    public void testValue() {
        ValueOperations<String, Object> valueOperations = redisTemplate.opsForValue();
        valueOperations.set("string", "I am a string");
        log.info("string:{}", valueOperations.get("string"));

    }
}

虽说RedisTemplate抽象了Redis的操作，但是你要正确使用也必须先大概搞清楚其底层的数据结构及操作命令才好下手使用，不然也是不行的。

下图展示了RedisTemplate对Redis常用功能的分装。

至于如何使用，请看文章后面的源码示例。

多一点

Redis事务

事务应该都不陌生了，一系列操作要么都执行，要么都不执行。但是Redis事务比较扯，例如一个事务有2条命令，第一条报错了，但第二条却可以执行成功，而不是在报错的时候回滚。（通过Lua脚本就不会这样，当其中某条命令发生错误时，后面的命令不会被执行。）

Redis的事务原理是将多条命令先放到一个队列中，然后统一执行。使用multi命令启动事务，使用exec命令执行事务，使用discard放弃执行，还可以使用watch来监视一个key，如果这个key的值被修改过，则放弃执行整个事务。

public void testTransaction() {
    ValueOperations<String, Object> valueOperations = redisTemplate.opsForValue();
    valueOperations.set("trx_key", "watch");
    //包含事务中每条命令的执行结果
    List<Object> trxResult = redisTemplate.execute(new SessionCallback<List<Object>>() {
        @Override
        public List<Object> execute(RedisOperations operations) throws DataAccessException {
            //启动watch
            operations.watch("trx_key");
            operations.multi();
            operations.opsForValue().set("key1", "value1");
            Object value1 = operations.opsForValue().get("key1");
            log.info("key1的值为:{}", value1);
//                operations.opsForValue().increment("key1");
            operations.opsForValue().set("key2", "value2");

            //测试的时候在此处打断点，然后使用redis客户端修改watch的值后再执行，看看能否执行成功
            return operations.exec();
        }
    });
    log.info("事务执行结果:{}", trxResult);
}

管道（Pipeline）

使用管道可以极大的提高Redis的qps，管道其实就是批量执行命令，就和mysql批量执行sql命令差不多。因为一条一条执行命令的话就会有很多时间花费在客户端与服务器的传输上，而Redis服务器却吃不饱。

好比六x龄童去某小学签售自己的破书骗钱，这爷们儿签名手速比我撸管时候的手速都快，一分钟能签100多本，开始小学生都是一个一个排着队过来找他签名，一个签完拿着书走回去另一个再过来。老六说了，这么签下去，晚上的酒局还组不组了？一次来10个，把书都展开并排放好，他刷刷画了几下，10本就处理完了，然后10个小朋友拿着书走了，下一批花朵进来了...

public void testPipeline() {
    List<String> results = new ArrayList<>();

    List<Object> pipeResult = redisTemplate.executePipelined(new SessionCallback<Object>() {
        @Override
        public Object execute(RedisOperations operations) throws DataAccessException {
            for (int i = 0; i < 10; i++) {
                operations.opsForValue().set("pip:key" + i, "value" + i);
                results.add(String.valueOf(operations.opsForValue().get("pip:key" + i)));
            }
            log.info("命令只是进入队列，还没有执行, 结果:{}", results);
            return null;
        }
    });
    log.info("执行结果类型:{}，结果:{}", pipeResult.getClass(), pipeResult);
    log.info("读取数据结果:{}", results);
}

值得注意的是results里保存的都是null，你知道这是为什么吗？

Lua 脚本

简介

Lua 是一种轻量小巧的脚本语言，用标准C语言编写并以源代码形式开放，其设计目的是为了嵌入应用程序中，从而为应用程序提供灵活的扩展和定制功能。

不知道同学你第一眼看到Lua这个名字有没有想歪，我见到它的第一眼就感慨怎么叫了这么个名字（撸啊），这的多少伤身体，估计是我比较闷骚...。Lua脚本有自己的语法，如果要写Lua脚本需要先学习一下。

Redis执行lua脚本的时候是由统一的一个解析器进行执行，脚本执行的过程中是不会有其它脚本和指令执行，意思就是说一个脚本的执行过程不会被其他指令打断，所以能保证脚本里的操作具备原子性。

下面我们用Lua脚本写个转账的小例子，看一看在SpringBoot中怎么使用Lua脚本。

案例需求：从账号from给账号to转50块钱。

假设我们使用hash保存两个账户，这个hash的key为account， key1="from"/value1 = 250 ，key2="to"/value2 = 200,转账后，value1= 200， value2 = 250.

Lua脚本

我们先写一个Lua脚本，放在resources/scripts/moneyTransfer.lua下

--转账
local account = 'account'
--获取from账号金额
local fromBalance = tonumber(redis.call('HGET', account, KEYS[1]))
--获取to账号金额
local toBalance = tonumber(redis.call('HGET', account, KEYS[2]))
--获取转账金额
local transAmount = tonumber(ARGV[1])
--如果from账号金额大于转账金额则进行转账
if fromBalance >= transAmount
then
    redis.call('HSET', account, KEYS[1], fromBalance - transAmount)
    redis.call('HSET', account, KEYS[2], toBalance + transAmount)
    return true
end
return false

构建Lua脚本相应的Java对象

@Configuration
public class RedisConfig {

    @Bean
    public RedisScript<Boolean> moneyTransferScript(){
        Resource resource = new ClassPathResource("scripts/moneyTransfer.lua");
        return RedisScript.of(resource,Boolean.class);
    }
}

可以看到那个返回值与Lua脚本的返回值类型对应，都是Boolean。

执行脚本

执行Lua脚本主要是使用RedisTemplate的execte的两个带有RedisScript参数的重载版本

private final RedisScript<Boolean> moneyTransferScript;

public Boolean testLuaScript() {
    BoundHashOperations<String, Object, Object> accounts = redisTemplate.boundHashOps("account");
    String from = "from";
    accounts.put(from, 250);
    String to = "to";
    accounts.put(to, 200);
    RedisSerializer<String> strSerializer = new StringRedisSerializer();
    RedisSerializer<Boolean> boolSerializer = new RedisSerializer<Boolean>() {
        @Override
        public byte[] serialize(Boolean aBoolean) throws SerializationException {
            return String.valueOf(aBoolean).getBytes(StandardCharsets.UTF_8);
        }

        @Override
        public Boolean deserialize(byte[] bytes) throws SerializationException {
            return Boolean.valueOf(new String(bytes, StandardCharsets.UTF_8));
        }
    };

    Boolean result = redisTemplate.execute(moneyTransferScript,
            strSerializer, boolSerializer,
            Arrays.stream(new String[]{from, to}).collect(Collectors.toList()),
            "50");
    log.info("转账结果:{}", result);
    return result;
}

路子就是这么个路子，至于如何写Lua脚本还是要学习它的语法。上面的代码我还自己实现了一个Boolean类型的RedisSerializer，是不是也很简单呢？

直接使用Redis客户端

我们知道RedisTemplate底层使用了Lettuce，当RedisTemplate不能满足我们的需求时是否可以直接使用Lettuce客户端呢，当然可以拉，只是你又需要去学习它的用法拉。其支持同步，异步和响应式编程，老强大拉，你有兴趣就去研究一下吧。

如下代码所示：最关键的步骤为通过RedisTemplate获取到Lettuce的StatefulRedisConnection。

public void testRedisClient() {
    Object nativeConnection = redisTemplate.getConnectionFactory().getConnection().getNativeConnection();
    RedisAsyncCommandsImpl redisAsyncCommands = null;
    if (nativeConnection instanceof RedisAsyncCommandsImpl) {
        redisAsyncCommands = (RedisAsyncCommandsImpl) nativeConnection;
    }
    if (Objects.isNull(redisAsyncCommands)) {
        return;
    }
    //顺利拿到了Lettuce的StatefulRedisConnection，接下来就可以使用Lettuce客户端的各种操作了
    StatefulRedisConnection<String, Object> lettuceCon = redisAsyncCommands.getStatefulConnection();

    //同步操作
    RedisStringCommands<String,Object> sync = lettuceCon.sync();
    sync.set("l:key1","value1");
    log.info("lettuce sync get:{}", sync.get("l:key1"));

    //异步操作
    RedisStringAsyncCommands<String, Object> async = lettuceCon.async();
    RedisFuture<String> set = async.set("l:key2", "value2");
    RedisFuture<Object> get = async.get("l:key2");

    get.thenAccept(new Consumer<Object>() {
        @Override
        public void accept(Object o) {
            log.info("lettuce async get:{}", o);
        }
    });

    //Reactive用法
    RedisStringReactiveCommands<String, Object> reactive = lettuceCon.reactive();
    Mono<String> reactSet = reactive.set("l:key3", "value3");
    Mono<Object> reactGet = reactive.get("l:key3");

    reactGet.subscribe(new Consumer<Object>() {
        @Override
        public void accept(Object o) {
            log.info("lettuce react get:{}", o);
        }
    });
}

分布式锁

什么是分布式锁？为什么需要分布式锁？如何实现呢？

首先得理解什么是锁？编程领域这个锁和现实中的锁很相似，但是呢含义却更广，类比成看门狗可能更加合适。

假如you穿越回了古代成了一位通晓未来的大师，那么来找你问道的人就会络绎不绝。你为了避免人群一窝蜂上来把你给踩死（资源状态被破坏），一次只接待一位客人，于是你雇了一个管家（锁）。张三来问道，先得过管家这一关，管家一看李四正在里面问道，就让张三等一下(互斥锁)，然后李四半天不出来（超时），管家就把李四轰出来了（锁超时释放），把张三放进去了（其他线程获得锁）。突然有个人性问题很难，你思考良久不语，张三就和管家说：我先去外面上个厕所，一会回来（可重入锁）...

那分布式锁又是什么呢？以前的应用都是单体应用，所以使用语言提供的锁机制即可，例如synchronized。但现在都是分布式架构，同一个服务的多个实例，或者多个不同的服务都有可能并发访问某个资源。假如下单OrderService部署了两个实例,OrderService#A，OrderService#B要同时修改数据库的一个值，你说你要把锁加在里面有用吗？没用啊，两个不同的实例，可能被部署在两个不同的docker容器中，线程直接怎么可能协调呢。有的同学要说了，你傻啊，当然是加在数据库上啊，不得不说你真的很聪明。所以说我们还是要找一个中间人来协调各方的请求，刚好Redis也符合这个角色，Zookeeper,以及数据库都符合。

那么不使用中心化组件有可能实现协调吗？其实也有，你看区块链，大家互相都不鸟对方，就看谁的算力强，一定时间内搞出的链子谁最长就认谁是正统，其他人的都丢弃，然后在从此开始新的比赛。这样挺好的，就是有点废电，不过除了这个工作量算法，好像还有其他省电的算法...这有点扯远了

总而言之，只要中心化的组件（其他服务都要访问的组件）提供了可以适合修改状态的的机制即可用来做分布式锁。

分布式锁一般有如下指标：

互斥性：同一时刻只能有一个线程持有锁
可重入性：同一节点上的同一个线程如果获取了锁之后能够再次获取锁
锁超时：支持锁超时释放，防止死锁
高性能和高可用：加锁和解锁需要高效，同时也需要保证高可用，防止分布式锁失效
具备阻塞和非阻塞性：当一个线程没有获得锁可以阻塞等待，当锁被释放后立刻去抢

分布式锁的实现方式

一般实现分布式锁有以下几种方式：

基于数据库

基于表的唯一性索引实现，弱势非常明显，不具备生产可用性。

大体思路如下：

新建一张表，然后建立一个唯一索引，各个线程抢锁的时候就往表里面插入一条数据，因为有唯一性索引，可以保证只能有一个线程插入成功（加锁），当操作完业务后再把那条数据给删了（释放锁）

会遇到很多问题：数据库单点，超时不释放，不可重入，无法阻塞等等问题都需要解决

基于zookeeper

基于ZooKeeper的临时有序节点实现

大体思路如下：

创建一个持久化节点 /lock。
假设有两个客户端，C1，C2来抢锁。假如C1先来的，那么它会在节点下创建临时有序子节点:/lock/00000001，检查自己是否是/lock下最小的子节点，发现还真是，于是就加锁成功了，开始执行执行业务了。
C2也来加锁，它也在 /lock01 下创建了一个临时子节点/lock/00000002，查看一下自己创建的子节点是否为当前列表中序号最小的子节点，发现不是，于是它就watch了/lock/00000001节点，等待它被干掉，自己就是最小的了，就可以上位；
C1执行完业务后就把/lock/00000001节点给删除了，因为C2一直监听着它，所以C2立马发现/lock/00000002是当前列表中最小的子节点了，于是它就加锁开始执行业务。

下面的图致敬至：ZK（ZooKeeper）分布式锁实现

ZooKeeper分布式锁一般使用curator这个ZooKeeper客户端来实现，有需求直接调查它即可。

<dependency> 
   <groupId>org.apache.curator</groupId> 
   <artifactId>curator-recipes</artifactId> 
   <version>x.x.x</version> 
</dependency>

基于Redis

Redis原生提供的加锁命令为

SET key value [EX seconds] [PX milliseconds] [NX|XX]

EX seconds ：将键的过期时间设置为 seconds 秒。执行 SET key value EX seconds 的效果等同于执行 SETEX key seconds value 。
PX milliseconds ：将键的过期时间设置为 milliseconds 毫秒。执行 SET key value PX milliseconds 的效果等同于执行 PSETEX key milliseconds value 。
NX ：只在键不存在时，才对键进行设置操作。执行 SET key value NX 的效果等同于执行 SETNX key value 。
XX ：只在键已经存在时，才对键进行设置操作。

例如下面这句命令的意思是：当lock这个key不存在时将其value设置为123456并设置其过期时间为10秒

SET lock 123456 EX 10 NX

虽然这个命令很简单，但是要实现一把Redis分布式锁却需要处理非常多的细节，具体推荐看
# Redis实现分布式锁的8大坑！切记！这篇文章。

下面我们简单开发一把简单的Redis分布式锁，抛砖引玉

public void testDistributeLock(){
    final String lockKey = "lock";
    final String lockValue = UUID.randomUUID().toString();
    try {
        Boolean isSuccess = redisTemplate.opsForValue().setIfAbsent(lockKey,lockValue,30, TimeUnit.SECONDS);
        log.info("线程[{}]加锁状态:{}",Thread.currentThread().getName(),isSuccess);
        if(!isSuccess){
            long startTime = System.currentTimeMillis();
            boolean isTrySuccess = false;
            //下面这一段代码是让线程自旋10秒，也就是说当线程加锁的时候发现锁被人占了，它就在10秒内不断的尝试，10秒后放弃
            while (true){
                long currentTime = System.currentTimeMillis();
                if(currentTime-startTime>10*1000){
                    log.info("线程[{}]自旋加锁失败");
                    break;
                }
                Thread.sleep(1000);
                isTrySuccess = redisTemplate.opsForValue().setIfAbsent(lockKey,lockValue,30, TimeUnit.SECONDS);
                log.info("线程[{}]自旋加锁状态:{}",Thread.currentThread().getName(),isTrySuccess);
                if(isTrySuccess){
                    break;
                }
            }
            if(!isTrySuccess){
                return;
            }
        }
        //业务代码
        business();
    } catch (Exception e) {
        log.error("加锁异常",e);
    } finally {
        releaseLock(lockKey, lockValue) ;
    }
}

private void releaseLock(String lockKey, String lockValue) {
        //只有持有锁的线程才能释放锁，释放锁这两步，比较和删除需要原子性，应该使用Lua脚本实现
//        if (lockValue.equals(redisTemplate.opsForValue().get(lockKey))) {
//            log.info("线程[{}]释放锁:{}", Thread.currentThread().getName(), lockValue);
//            redisTemplate.delete(lockKey);
//        }

        String delLua = new StringBuilder()
                .append("if redis.call('get',KEYS[1]) == ARGV[1]")
                .append("\n")
                .append("then")
                .append("\n")
                .append(" return redis.call('del',KEYS[1])")
                .append("\n")
                .append("else")
                .append("\n")
                .append(" return 0")
                .append("\n")
                .append("end")
                .toString();
        log.info("删除的lua脚本:\n{}", delLua);
        RedisScript<Long> rs = RedisScript.of(delLua, Long.class);
        //不传参数和返回值的序列化器则使用template的value的序列化器
        Long result = redisTemplate.execute(rs, Arrays.asList(lockKey), lockValue);
        if (result > 0) {
            log.info("线程[{}]释放锁:{}", Thread.currentThread().getName(), lockValue);
        }
    }

private void business() {
    log.info("线程[{}]开始处理业务...",Thread.currentThread().getName());
    try {
        Thread.sleep(5000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    log.info("线程[{}]结束处理业务...",Thread.currentThread().getName());
}

上面的分布式锁有几点需要说明一下：

加锁

使用下面命令加锁，保证值设置和过期时间操作的原子性

SET lock uuid EX 30 NX

释放锁

加锁时value存放的是全局唯一的uuid，在释放锁的时候比较这个值，防止线程把其他线程的锁释放了。

如何重现这种情况？

例如锁的过期时间是30秒，线程1执行耗时40秒的业务，那么在其主动释放锁之前锁就过期被删除了。假设在35秒的时候，线程2获得了锁，然后开始执行耗时10秒的业务，5秒后线程1执行完了自己的业务，主动去释放锁，也就是删除lock那个key，如果不加以验证，它就会把线程2创建的锁给删了，关键是线程2还没用完呢。

锁阻塞

当一个线程尝试加锁不成功时，不是直接返回而是等待一定的时间，在此期间不断的尝试加锁

我们来验证一下，使用PostMan开两个窗口，连续发起请求，输入如下：

: 线程[http-nio-8080-exec-1]加锁状态:false
: 线程[http-nio-8080-exec-4]加锁状态:true
: 线程[http-nio-8080-exec-4]开始处理业务...
: 线程[http-nio-8080-exec-1]自旋加锁状态:false
: 线程[http-nio-8080-exec-1]自旋加锁状态:false
: 线程[http-nio-8080-exec-1]自旋加锁状态:false
: 线程[http-nio-8080-exec-1]自旋加锁状态:false
: 线程[http-nio-8080-exec-4]结束处理业务...
: 线程[http-nio-8080-exec-1]自旋加锁状态:false
: 线程[http-nio-8080-exec-4]释放锁:726f079b-ba6d-4a7d-b5ca-ed8ca82dd06c
: 线程[http-nio-8080-exec-1]自旋加锁状态:true
: 线程[http-nio-8080-exec-1]开始处理业务...
: 线程[http-nio-8080-exec-1]结束处理业务...
: 线程[http-nio-8080-exec-1]释放锁:e40bd110-a36f-4c18-a216-0e9de1fd1fd1

从输出可可以看到，线程exec-4先抢到了锁，其锁里保存的value是726f079b-ba6d-4a7d-b5ca-ed8ca82dd06c。在其执行业务的同时，线程exec-1在不断自旋尝试获取锁，在线程exec-4完成业务并释放锁后，线程exec-1终于获取锁成功，其锁里保存的value是e40bd110-a36f-4c18-a216-0e9de1fd1fd1

根据你的业务需求，会对锁提出各种各样的要求，例如：锁重入，锁延期，锁分段，集群下失败等等问题，还是要具体业务具体分析了。

如果在生产中使用还是推荐直接使用Redis它儿子:Redisson，里面实现了很多锁。