redis 分布式缓存详解

Redis命令操作、Redis持久化、Spring整合Redis、Redis的分布式缓存、主从复制、哨兵监测、集群搭建、分布式Session实现

Modify_QaQ

3292人浏览 · 2021-12-17 23:05:56

Modify_QaQ · 2021-12-17 23:05:56 发布

1、Redis概述

1.1、NoSQL

NoSQL(Not Only SQL)，意即不仅仅是SQL, 泛指非关系型的数据库。

1.2、Redis安装

首先需要从Redis官网上下载Redis的源码包，将下载的包上传到Linux，之后将gz文件进行解压。

# 解压gz文件
tar -zxvf redis-6.2.6.tar.gz
# 进入目录
cd redis-6.2.6
# 进行编译源码
make MALLOC=libc
# 将redis进行安装
make install PREFIX=/usr/redis
# 进入到bin目录下 
cd usr/redis/bin
# 进行启动服务
./redis-server
# 连接
./redis-cli -h localhost 6379

加载配置启动，在之前的源码包当中有一个redis.conf文件，这是官方给进行提供的一个配置文件。

# 将配置文件复制到部署目录下
cp redis.conf /usr/redis
# 加载配置文件进行启动
./redis-server ../redis-conf

2、Redis 指令操作

2.1、key操作

# 创建key设置值
set name niao
# 获取值
get name
set name1 huang
set name2 yue
# 获取所有key
keys *
# 进行匹配搜索， * 任意字符		？ 一个字符 		[01] 0或者1的其中一个
keys name*
keys name?
keys name[01]

# 判断key是否存在
exists name

# 切换库
select 1
select 0
# 将key移动到其他库
move name 1
select 1
keys *

# 设置key存在时间
expire name1 5000

# 查看key还有多少时间过期
ttl name1  -1 永久  -2 不存在
pttl name1

# 随机获取一个key
randomkey

# 修改key的名称
rename name2 Name

# 查看key的类型
type Name

2.2、String 类型操作

# 清除所有key
flushall

# 一次设置多个key
mset name huang age 18 bir 2000-01-03

# 一次获取多个key的值
mget name age bir

# 获取键值重新复制
getset name niao

# 获取值的长度
strlen name

# 在键值后面进行追加内容
append name wang

# 对键值进行切片
getrange name 5 8

# 设置过期时间
setex age 100 19
psetex bir 50000 2012-05-01
setnx age 18
# 原子操作 对应给定的keys到他们相应的values上。只要有一个key已经存在，一个操作都不会执行
msetnx name huang content content

# 进行加减操作
decr age
decrby age 5
incr age
incrby age 5

# 向上取整保留17位小数
incrbyfloat age 12.6

2.3、List 类型操作

# 从左边开始进行push加入元素
lpush llist zhangsan lisi wangwu zhaoliu
rpush rlist zhangsan lisi wangwu zhaoliu

# 遍历集合
lrange rlist 0 -1

# 从左边加入，且key值存在
lpushx name aaa
rpushx name aaa

# 从左边弹出元素
lpop rlist
rpop rlist

# 集合长度
llen rlist

# 重新赋值
lset rlist 0 huangyueyue

# 根据索引获取
lindex rlist 2

# 删除元素
lrem rlist 2 wangwu
# 截取list
ltrim rlist 0 2

# 插入元素
linsert rlist after lisi  niao
linsert rlist before lisi  niao

2.4、Set 类型操作

# 往set集合加入内容
sadd sets zhangsan lisi wangwu zhaoliu zhangsan

# 获取所有元素
smembers sets

# 获取长度
scard sets

# 弹出元素
spop sets 1

sadd age 18 19 20
# 移动元素
smove sets age zhangsan

# 删除元素
srem age zhangsan
sismember sets zhangsan
srandmember sets 3

# 去除sets集合当中含有age的所有元素
sdiff sets age

# 求交集
sinter sets age

# 求并集
sunion sets age

2.5、ZSet 类型操作

zadd zsets 10 zhangsan 16 lisi 25 wangwu 30 zhangsan
type zsets
zrange zsets 0 -1
zrange zsets 0 -1 withscores
zrange zsets 0 -1 withscores limit 0 3

zcard zsets

zrank zsets zhangsan
zrevrank zsets zhangsan

zrem zsets zhangsan

zincrby zsets 1 zhangsan

2.6、Hash 类型操作

hset maps name zhangsan
type maps
hget maps name
hgetall maps
hdel maps name
hexists maps name
hvals maps 
hmset maps name lisi age 18
hmget maps 
hsetnx maps name wangwu
hincrby maps age 5
hincrbyfloat maps age 15.326

# 开启redis的远程连接
bind 0.0.0.0
# 加载配置文件重启服务
./redis-server ../redis.conf

3、Redis 持久化

持久化方式：

快照持久化
AOF append only file 只追加日志文件，将所有的写入操作记入到日志文件当中，当服务突然宕机之后，重启时只需要在执行一遍日志文件，这样数据还是会持久化回来。

3.1、快照持久化

3.1.1、快照持久化的特点：

快照持久化可以将某一时刻的所有数据写入到硬盘当中，当然这也是redis会默认进行开启，保存的文件是以rdb形式结尾的文件，因此这种方式也被称为RDB方式。

3.1.2、快照的创建：

可以使用BGSAVE或者SAVE命令来进行创建快照

Save 命令执行一个同步保存操作，将当前 Redis 实例的所有数据快照(snapshot)以 RDB 文件的形式保存到硬盘。
BGSAVE 命令执行之后立即返回 OK ，然后 Redis fork 出一个新子进程，原来的 Redis 进程(父进程)继续处理客户端请求，而子进程则负责将数据保存到磁盘，然后退出。客户端可以通过 LASTSAVE 命令查看相关信息，判断 BGSAVE 命令是否执行成功。

3.1.3、SAVE和BGSAVE的区别：

SAVE 保存是阻塞主进程，客户端无法连接redis，等SAVE完成后，主进程才开始工作，客户端可以连接
BGSAVE 是fork一个save的子进程，在执行save过程中，不影响主进程，客户端可以正常链接redis，等子进程fork执行save完成后，通知主进程，子进程关闭。

在redis当中，默认会开启快照持久化，在当redis通过shutdown指令接收到关闭服务器的请求时，会执行save命令。阻塞所有的客户端，不再执行客户端发送的任何命令。并且save命令执行完成之后关闭服务器。

3.1.4、快照配置：

# 用来指定快照的文件名
dbfilename dump.rdb
# 快照生成的目录
dir ./

3.1.5、存在问题：

突然宕机导致部分未创建的快照的数据丢失

3.2、AOF 持久化

3.2.1、AOF的特点：

这种方式可以将所有客户端执行的写命令记录到日志文件当中，AOF持久化会将被执行的写命令到AOF的文件末尾，以此用来记录数据发生的变化，因此只要redis从头到尾执行一次AOF文件所包含的所有命令，就可以恢复AOF文件的记录的数据集。

3.2.2、AOF的配置：

appendonly yes
appendfilename appendonly.aof

同步频率

always 每个 Redis 命令都要同步写入硬盘。这样会严重降低 Redis 的性能
everysec 每秒执行一次同步，显式地将多个写命令同步到硬盘
no 让操作系统来决定应该何时进行同步

3.2.3、AOF重写：

AOF持久化会把redis所有的写命令都保存到AOF文件，使得AOF文件越来越大，大大占用磁盘使用空间，也给还原redis数据很大时间。AOF是记录redis执行的写命令也就是一些重复执行的写命令都会记录到AOF文件内，所以所删除重复的写入命令可以适当的缩小AOF文件大小。

直接在客户端当中执行 BGREWRITEAOF 命令进行重写AOF文件，但是当aof文件很大的时候执行 BGREWRITEAOF 会导致性能问题，为此redis也提供了对应策略：

# 配置当 AOF 文件需要比旧 AOF 文件增大多少时才进行AOF重写，100代表增大了百分百，即一倍
auto-aof-rewrite-percentage 100
# 配置当 AOF 文件需要达到多大体积时才进行 AOF 重写。只有这两个配置的条件都达到时，才会进行AOF 重写
auto-aof-rewrite-min-size 20M

3.2.4、AOF重写原理：

redis会从整个内存当中的数据库内容的方式重写了一个新的AOF文件，替换原有的文件。

AOF的重写流程

redis调用fork，创建父子进程，子进程根据内存的数据库快照，往临时文件当中写入数据库状态命令
父进程继续处理客户端的请求，除了把命令写入到原来的AOF文件当中，同时把收到的命令缓存起来，这样是用来保证如果子进程重写失败，原来的AOF文件也不会出现问题。
当子进程把快照的内容写入以命令方式写入到临时文件当中后，子进程通知父进程，父进程收到后，将缓存的新命令写入临时文件。
最后父进程直接将临时文件替换掉原本的AOF文件，并重命名，后面收到的命令开始往新的AOF文件当中追加。

3.2.5、AOF 的优点：

AOF 持久化的方法提供了多种的同步频率，即使使用默认的同步频率每秒同步一次，Redis 最多也就丢失 1 秒的数据而已。（保证数据完整性，对数据要求高的建议使用）
AOF 文件使用 Redis 命令追加的形式来构造，因此，即使 Redis 只能向 AOF 文件写入命令的片断，使用 redis-check-aof 工具也很容易修正 AOF 文件。（容易修改写入的命令）
AOF 文件的格式可读性较强，这也为使用者提供了更灵活的处理方式。例如，如果我们不小心错用了 FLUSHALL 命令，在重写还没进行时，我们可以手工将最后的 FLUSHALL 命令去掉，然后再使用 AOF 来恢复数据。

3.2.6、AOF 的缺点：

对于具有相同数据的的 Redis，AOF 文件通常会比 RDB 文件体积更大。
虽然 AOF 提供了多种同步的频率，默认情况下，每秒同步一次的频率也具有较高的性能。但在 Redis 的负载较高时，RDB 比 AOF 具好更好的性能保证。
RDB 使用快照的形式来持久化整个 Redis 数据，而 AOF 只是将每次执行的命令追加到 AOF 文件中，因此从理论上说，RDB 比 AOF 方式更健壮。官方文档也指出，AOF 的确也存在一些 BUG，这些 BUG 在 RDB 没有存在

4、使用 Java 操作 Redis

4.1、通过 API 操作Redis

在使用java来操作Redis，需要使用到jedis这个jar包，直接构建一个maven项目，在pom文件当中加入依赖。

        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>2.9.0</version>
        </dependency>

编写java代码连接redis服务进行操作。在这里需要注意关闭远程服务器的防火墙，以及加载配置文件启动redis，让远程机器可以连接到redis。

        Jedis jedis = new Jedis("192.168.101.128",6379);
        // 默认使用0号库
        jedis.select(0);
        Set<String> keys = jedis.keys("*");
        for(String key : keys){
            System.out.println(key);
        }
        // 清空当前库
        // jedis.flushDB();
        // 清空所有库
        // jedis.flushAll();

        // 释放资源
        jedis.close();

而对于redis里面的一些键、数据类型的操作和redis的命令基本一致，这里只列出了部分方法，对于其余方法直接查看Jedis这个类当中提供的方法即可。部分代码如下：

    public void keyTest(){
        // 删除key
        jedis.del("name");
        jedis.del("name","age");

        // 设置超时时间
        jedis.expire("age",100);
        jedis.ttl("age");

        // 随机key
        jedis.randomKey();

        // 重命名
        jedis.rename("age","newAge");
        // 类型
        jedis.type("age");
    }

    private void stringTest(){
        jedis.set("name","yueyue");
        jedis.get("name");

        jedis.mset("age","18","address","长沙");

        jedis.mget("name","age","address");
    }

    private void listTest(){
        jedis.lpush("llist","zhangsan","lisi");
        jedis.rpush("rlist","zhangsan","lisi");
        List<String> llist = jedis.lrange("llist",0,-1);
        for (String s : llist){
            System.out.println(s);
        }
        jedis.lpop("llist");
        jedis.lindex("llist",2);
        jedis.linsert("llist", BinaryClient.LIST_POSITION.AFTER,"zhangsan","zhangsanDemo");
    }

4.2、Spring 整合 Redis

首先直接快速初始化一个springboot项目，并且导入相对应的依赖。

		<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>

并且在配置文件当中对redis服务进行配置，指定redis的服务主机、端口号、使用的库

server.port=8080

# redis 配置
spring.redis.host=192.168.101.128
spring.redis.port=6379
spring.redis.database=0

直接在Test当中来对redis进行操作。在导入的spring-redis的依赖当中，提供了两个用来操作redis的类，StringRedisTemplate和RedisTemplate。

而两者管理的的数据是不共通的；也就是说StringRedisTemplate只能管理StringRedisTemplate里面的数据，RedisTemplate只能管理RedisTemplate中的数据。其实他们两者之间的区别主要在于他们使用的序列化类:

RedisTemplate使用的是JdkSerializationRedisSerializer 存入数据会将数据先序列化成字节数组然后在存入Redis数据库。
StringRedisTemplate使用的是StringRedisSerializer

这里直接用Spring的Test单元测试进行编写代码：直接将两者都进行注入。

package com.lzq;

import com.lzq.entity.User;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.redis.connection.DataType;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.util.*;

@SpringBootTest
class SpringredisApplicationTests {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private RedisTemplate redisTemplate;

    @Test
    void keyTest() {
    	// 随机key
        String randomKey = stringRedisTemplate.randomKey();
        System.out.println(randomKey);

		// key的类型
        DataType nameType = stringRedisTemplate.type("name");
        System.out.println(nameType);
		
		// 所有key
        Set<String> keys = stringRedisTemplate.keys("*");
        for (String key : keys) {
            System.out.println(key);
        }
		
		// 创建key并设置超时时间
        Long nameExpire = stringRedisTemplate.getExpire("name");
        System.out.println(nameExpire);
        // stringRedisTemplate.expire("name",);
    }

    @Test
    void stringTest() {
    	// 设置一个name
        stringRedisTemplate.opsForValue().set("name", "zhangsan");
		
		// 获取键值
        String age = stringRedisTemplate.opsForValue().get("age");
        String name = stringRedisTemplate.opsForValue().get("name");
        System.out.println(name);
        System.out.println(age);
    }

    @Test
    void listTest() {
    	// 从坐插入一个元素
        stringRedisTemplate.opsForList().leftPush("llist", "zhangsan");
        
        // 从左插入多个元素
        stringRedisTemplate.opsForList().leftPushAll("llist", "wangwu", "zhaoliu");
        
        // 获取所有元素
        List<String> llist = stringRedisTemplate.opsForList().range("llist", 0, -1);
        for (String s : llist) {
            System.out.println(s);
        }
        
        // 从右边开始插入
        stringRedisTemplate.opsForList().rightPush("llist", "right");
    }

    @Test
    void setTest() {
    	// 创建set
        stringRedisTemplate.opsForSet().add("set", "zhangsan", "lisi", "wu", "zhangsan");
        
        // 获取所有元素
        Set<String> set = stringRedisTemplate.opsForSet().members("set");
        for (String s : set) {
            System.out.println(s);
        }
        
        // 随机元素
        String random = stringRedisTemplate.opsForSet().randomMember("set");
        
        // set长度
        Long size = stringRedisTemplate.opsForSet().size("set");
    }

    @Test
    void ZsetTest(){
    	// 创建zset
        stringRedisTemplate.opsForZSet().add("zset","zhangsan",15);

		// 获取所有元素（获取不到分数）
        Set<String> zset = stringRedisTemplate.opsForZSet().range("zset",0,-1);
        for(String s : zset){
            System.out.println(s);
        }
        System.out.println("==============================================");
        
        // 获取所有元素 包含分数
        Set<ZSetOperations.TypedTuple<String>> zset1 = stringRedisTemplate.opsForZSet().rangeByScoreWithScores("zset", 0, 100);
        for (ZSetOperations.TypedTuple type : zset1){
            System.out.println(type.getScore());
            System.out.println(type.getValue());
        }
    }

    @Test
    void hashTest(){
    	// 创建hash
        stringRedisTemplate.opsForHash().put("hash","name","zhangsan");
        
        // 往hash里面插入多个键值对
        Map<String,String> map = new HashMap<>();
        map.put("age","18");
        map.put("address","长沙");
        stringRedisTemplate.opsForHash().putAll("hash",map);
        
        // 获取hash的指定键值
        Object name = stringRedisTemplate.opsForHash().get("hash", "name");
        System.out.println(name);
		
		// 获取所有的键和值
        Set<Object> keys = stringRedisTemplate.opsForHash().keys("hash");
        List<Object> values = stringRedisTemplate.opsForHash().values("hash");
        for (Object key : keys){
            System.out.println("key = " + key);
        }
        for (Object value : values){
            System.out.println("value = " + value);
        }
    }

    @Test
    void ObjectTest(){
        // 设置键的序列化方式
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        // 设置hash类型的key的序列化
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        
        User user = new User();
        user.setName("yueyue");
        user.setAge(18);
        user.setAddress("长沙");
        user.setBir(new Date());
		
		// 给string、list、set、zset、hash类型 设置对象，这里的User也必须要实现序列化接口
        redisTemplate.opsForValue().set("ovalue",user);
        redisTemplate.opsForList().rightPush("list",user);
        redisTemplate.opsForSet().add("set",user);
        redisTemplate.opsForZSet().add("zset",user,100);
        redisTemplate.opsForHash().put("hash","user",user);
		
		// 获取对象值
        User user1 = (User) redisTemplate.opsForValue().get("ovalue");
        System.out.println(user1.toString());
    }
}
    @Test
    void boundTest(){
    	// 对同一个对象进行多次操作，可以将这个key进行绑定
        BoundValueOperations<String, String> name = stringRedisTemplate.boundValueOps("name");
        name.append("is good man");
        System.out.println(name.get());
    }

总结：

对于处理键值都是string使用StringRedisTemplate
对于键值存在对象使用RedisTemplate
对于同一个key多次操作可以使用boundXXXOps value、list、set、hash

5、Redis 应用场景

利用string类型完成项目中手机验证码的实现
利用string类型完成具有失效性的业务功能，如：抢票、订单等付款时间限制
利用分布式集群当中 session的共享
利用zset类型完成分数、榜单之类的功能
利用分布式缓存完成
利用redis的超时完成token存储
解决分布式锁的问题

6、使用 Redis 实现分布式缓存

6.1、缓存

6.1.1、缓存是什么？

是计算机内存的一段数据

6.1.2、缓存的特点

读写快
断电立即丢失

6.1.3、缓存解决了什么问题

提高网站吞吐量，提高网站运行效率
用来减轻对数据库访问压力，这里使用缓存的数据很少发生改变

6.1.4、本地缓存和分布式缓存的区别

本地缓存（Local Cache）：存在应用服务器内存之中的数据
分布式缓存（Distribute Cache）：存储在当前应用服务器内存之外的数据

6.2、Mybatis 缓存

首先，这里还是直接导入相关依赖，并且编写对应的配置信息，这里的单个简单的整个Mybatis也就不过多的赘述了，

        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid</artifactId>
            <version>1.0.20</version>
        </dependency>

        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.19</version>
            <scope>compile</scope>
        </dependency>
        
        <dependency>
            <groupId>org.mybatis.spring.boot</groupId>
            <artifactId>mybatis-spring-boot-starter</artifactId>
            <version>2.1.4</version>
        </dependency>

spring.datasource.type=com.alibaba.druid.pool.DruidDataSource
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.url=jdbc:mysql://localhost:3306/springboot?characterEncoding=UTF-8
spring.datasource.username=root
spring.datasource.password=root

这里主要还是来进行说明mybatis的缓存，这里还是编写xml文件，在xml文件当中加上一个cache标签，也就表示着将查询获取的数据存到缓存当中，而后面的相同的查询，就直接会去缓存当中获取数据，不会再执行查询了。

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.lzq.mapper.UserMapper">
	<cache></cache>
    <select id="findAll" resultType="com.lzq.entity.User">
        select * from user
    </select>
</mapper>

查看cache标签缓存实现：Mybatis底层默认使用的是org.apache.ibatis.cache.ipml.PerpetulCache实现

6.3、Redis 缓存的实现

那这里我们的Redis的缓存就可以参考Mybatis的缓存来进行实现。

自定义 Cache 类实现 Cache 接口，并对里面的方法进行实现
使用<cache type="xom.xxx.cache.RedisCache">

首先我们添加这个自定义的cahce类，用来实现Cache接口

public class RedisCache implements Cache {
}

这里先不做任何处理，直接启动项目，查看报错信息，表示必须要一个带String 类型 id 的参数的构造函数，并且对getId方法的返回值进行修改，直接返回id即可。

private final String id;

public RedisCache(String id){
	this.id = id;
}

并且可以发现，在mybatis的cache当中的putObject和getObject方法就是获取数据以及将数据写入到缓存当中，到这里就简单了，我们直接用RedisTemplate将数据写入缓存，获取也是同理进行获取。

    @Override
    public void putObject(Object o, Object o1) {
        System.out.println(o+""+o1);
    }

    @Override
    public Object getObject(Object o) {
        return o;
    }

但是在这里又会有一个问题，这里是通过Xml文件直接导入的类，而这里又不能直接将RedisTemplate进行注入，这里又该如何处理呢?可以先通过实现ApplicationContextAware接口用来对单个Bean进行获取，这里的getBean方法就可以直接获取工厂当中的所有的Bean对象。

@Component
public class ApplicationContextUtil implements ApplicationContextAware {
    private ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    public Object getBean(String name){
        return applicationContext.getBean(name);
    }
}

这里都有了，后面的就简单了，只需要对RedisCache这个类的putObject和getObject方法进行改写，将获取的数据和缓存的数据直接向redis操作即可。

import org.apache.ibatis.cache.Cache;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.util.concurrent.locks.ReadWriteLock;

/**
 * @author huangyueyue
 * @date 2021/12/14 16:12
 */
public class RedisCache implements Cache {
    private final String id;

    ApplicationContextUtil applicationContextUtil = new ApplicationContextUtil();
    RedisTemplate redisTemplate = (RedisTemplate) applicationContextUtil.getBean("RedisTemplate");

    public RedisCache(String id) {
        this.id = id;
    }

    @Override
    public String getId() {
        return id;
    }

    @Override
    public void putObject(Object o, Object o1) {
        System.out.println(o+""+o1);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.opsForHash().put(id.toString(),o,o1);
    }

    @Override
    public Object getObject(Object o) {
        redisTemplate.opsForHash().get(id.toString(),o);
        return o;
    }

    @Override
    public Object removeObject(Object o) {
        return null;
    }

    @Override
    public void clear() {
        redisTemplate.delete(id.toString());
    }

    @Override
    public int getSize() {
        return redisTemplate.opsForHash().size(id.toString()).intValue();
    }

    @Override
    public ReadWriteLock getReadWriteLock() {
        return null;
    }
}

存在问题： 在这里对缓存进行存储的时候，使用的是单个dao的名称来作为hash的键，而单个dao进行了增删改操作时，cache会调用clear方法，将当前dao对应的缓存给清除掉，而当业务当时时多表进行关联的时候，这个时候每次操作只能删除一个dao对应的缓存，另一个模块产生的缓存无法清除，这就会导致信息混乱。

解决方案： 在mybatis当中对于关联关系的数据，可以使用cache-ref标签，用来指定多个dao层的数据进行缓存共享。只需要在对应的xml文件当中加上<cache-ref namespace="com.lzq.UserDao"></cache-ref>，表示当前dao生成的缓存给加入到UserDao当中。

6.4、Redis 缓存优化

6.4.1、对键值进行优化

对放入缓存当中的key进行优化，key的长度不能太长，需要对key进行设计，使用MD5算法；

一切文件字符串经过MD5处理之后都会生成32位16进制字符串
不同内容文件经过MD5加密之后，加密结果一定不一致
相同内容文件多次经过MD5加密之后生成的结果始终一致

在这里进行优化，直接使用SpringBoot提供的MD5加密的工具类进行实现。

    public String getKeyToMD5(String key){
        return DigestUtils.md5DigestAsHex(key.getBytes());
    }

后续在cache当中的对键进行MD5加密

    @Override
    public void putObject(Object o, Object o1) {
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.opsForHash().put(applicationContextUtil.getKeyToMD5(id.toString()), applicationContextUtil.getKeyToMD5(o.toString()), o1);
    }

    @Override
    public Object getObject(Object o) {
        return redisTemplate.opsForHash().get(applicationContextUtil.getKeyToMD5(id.toString()), applicationContextUtil.getKeyToMD5(o.toString()));
    }

6.4.2、缓存穿透

在进行查询的时候，查询一条不存在的数据，这种方式就叫做缓存穿透，也就是一直查id为-1的数据，一直访问进行恶意攻击，缓存利用不大，请求大规模的涌向数据库，容易导致数据库宕机崩溃。

在mybatis的cache当中解决了缓存穿透：将查数据库当中没有的记录，数据库也将查询的数据作为一个null进行存储到缓存当中。

6.4.2、缓存雪崩

在系统运行的某一时刻，突然系统中的缓存全部失效（这是在进行设置缓存的时候会给缓存设置一个超时时间，恰好在某一时刻全部失效），刚好在这一时刻，服务端涌来大量请求，导致缓存无法利用，大量请求涌向数据库，导致数据库宕机崩溃。

解决方案：

不设置超时时间（不推荐）
避免多个模块的缓存设置的超时时间相同

7、Redis 主从复制

主从复制架构仅仅是用来解决数据的冗余备份，从节点仅仅用来同步数据。

这里还是使用伪分布式来进行配置三个redis服务器。直接到redis的目录当中复制一份没做修改的redis.conf文件。在redis的启动目录当中新建三个文件夹，每个文件夹作为一个节点，存放一个配置文件，后续进行启动的时候，只需要加载对应的配置文件进行启动即可，这也就是伪分布式进行配置。

cd /usr/redis
mkdir master slave1 slave2
cd /tools/redis/redis-6.2.6
cp redis.conf /usr/redis/master
cp redis.conf /usr/redis/slave1
cp redis.conf /usr/redis/slave2

在配置文件都复制到对应的目录下之后，也就是三个节点，这个时候只需要修改对应的配置文件，分别给三个节点指定端口，以及bind，让各个节点之间可以进行通信，而在从节点上需要指定 slaveof 指向 master节点的ip和端口。

// master 
port 6379
bind 0.0.0.0

// slave1
port 6380
bind 0.0.0.0
slaveof 192.168.101.128 6379

// slave2
port 6381
bind 0.0.0.0
slaveof 192.168.101.128 6379

最后直接加载对应的配置文件进行启动redis以及分别连上主节点和从节点

ps aux|grep redis
cd bin
./redis-server ../master/redis.conf
./redis-server ../salve1/redis.conf
./redis-server ../slave2/redis.conf

./redis-cli -p 6379
./redis-cli -p 6380
./redis-cli -p 6381

在这里的从节点是只读的，当往从节点进行set元素，会进行提示，但是这里是默认只读，也可在配置文件当中修改read-only 进行调整。并且这里当主节点突然宕机，从节点也依旧是从节点，无法替换主节点。

8、Redis Sentinel 哨兵机制

8.1、哨兵机制概述

Sentinel（哨兵）是Redis 的高可用性解决方案：由一个或多个 Sentinel 实例组成的 Sentinel 系统可以监视任意多个主服务器，以及这些主服务器属下的所有从服务器，并在被监视的主服务器进入下线状态时，自动将下线主服务器属下的某个从服务器升级为新的主服务器。简单的说哨兵就是带有自动故障转移功能的主从架构。

无法解决:

单节点并发压力问题
单节点内存和磁盘物理上限

8.2、哨兵机制原理

故障转移

在搭建了redis的主从复制架构之后，master节点突然宕机而其他从节点不能替换master，这样就会导致整个架构不可用，而故障转移也就会将从节点升为主节点，让架构继续提供服务。

原理：

在配置了哨兵之后，哨兵用来监听各个节点直接的服务心跳，用来确保各个节点的服务都运行正常，而这个时候，当master节点突然宕机了，这个时候哨兵收不到主节点的心跳，哨兵就会先对主从节点的复制进行暂停，然后在现有的从节点重新推举一个新的节点作为master节点，而后续这个宕机的master节点重新加入将作为从节点继续进行数据同步。

8.3、哨兵的配置

只需要创建一个配置文件sentinel.conf文件，在文件内容当中写入：

# sentinel monitor 哨兵名称 监听ip 端口 哨兵个数
sentinel monitor master 192.168.101.128 6379 1

之后就可以使用 redis-sentinel脚本来加载配置文件进行启动哨兵即可。

./redis-sentinel ../master/sentinel.conf

可以看到在这里哨兵进行启动，哨兵也相当于一个服务，也有自己的端口服务等等，当第一次加载配置文件启动哨兵之后，对应的sentinel.conf文件也会自动补齐相对应的配置。

8.4、微服务配置

spring.redis.sentinel.master=master
spring.redis.sentinel.nodes=192.168.101.128:26379

然后运行项目，会发现报错了，这是因为哨兵作为一个服务，而这个服务没有开启远程连接，这里和redis的开始一样，直接在配置文件当中加入

bind 0.0.0.0

9、Redis 集群

在进行安装Redis集群之前首先需要安装Ruby

yum install -y ruby rubygems
# 并且下载一个redis-3.2.1.gem 进行安装
# 下载地址
# https://rubygems.org/gems/redis/versions/3.2.2
gem install redis-3.2.1

之后复制六份配置文件出来，分别用7001，7002，7003，7004，7005，7006六个文件夹即六个配置文件用来启动多个服务，修改对应的配置文件，这里以7001为例，后续的配置直接用7001的配置文件copy一份过去，将配置文件当中的7001改为对应的数字即可。

port 7001
bind 0.0.0.0
daemonize yes
dbfilename dump-7001.rdb
appendonly yes
appendfilename "appendonly-7001.aof"
cluster-enabled yes
cluster-config-file nodes-7001.conf
cluster-node-timeout 5000

修改对应6个节点的配置文件之后，就可以直接加载对应的配置文件进行启动redis了。

./redis-server ../7001/redis.conf
./redis-server ../7002/redis.conf
./redis-server ../7003/redis.conf
./redis-server ../7004/redis.conf
./redis-server ../7005/redis.conf
./redis-server ../7006/redis.conf

然后可以使用命令查看6个redis服务启动是否成功，ps aux|grep redis 当服务都启动好了之后，我们就可以直接来进行创建集群了，在这里使用 redis-trib.rb进行集群创建，首先去redis的解压目录下的src当中拷贝一份redis-trib.rb文件过来，之后直接执行命令：

./redis-trib.rb create --replicas 1 192.168.101.128:7001 192.168.101.128:7002 192.168.101.128:7003 
		192.168.101.128:7004 192.168.101.128:7005 192.168.101.128:7006

在进行构建集群的时候会出现如下警告：

在这里插入图片描述
这其实也不是错，不要慌，只是因为从redis5.0开始，建议使用redis-cli作为创建集群的命令，不推荐再使用redis-trib.rb来创建集群了，毕竟使用redis-trib.rb还要安装Ruby程序，比redis-cli麻烦的多。只需要使用redis-cli命令进行启动redis集群即可。

./redis-cli --cluster create 192.168.101.128:7001 192.168.101.128:7002 192.168.101.128:7003 
		192.168.101.128:7004 192.168.101.128:7005 192.168.101.128:7006

启动集群之后，我们可以连接到集群上的每一个节点，

./redis-cli -p 7003 -c

集群操作命令

# 查看集群节点状态
./redis-cli check 192.168.101.128:7003

# 将新的节点加入到集群当中
./redis-cli add-node 192.168.101.128:7007 192.168.101.128.7001

10、Redis 实现分布式 Session

管理机制：

redis的session管理是利用spring提供的session管理解决方案，将一个应用session交给redis进行存储。整个应用中所有的session请求都会去redis中获取对应的session数据。

首先构建一个SpringBoot项目，加入依赖

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

        <dependency>
            <groupId>org.springframework.session</groupId>
            <artifactId>spring-session-data-redis</artifactId>
        </dependency>

以及对项目用到的redis服务进行配置

server.port=8080
server.servlet.context-path=/redissession

spring.redis.cluster.nodes=192.168.101.128:7001,192.168.101.128:7002,192.168.101.128:7003,192.168.101.128:7004,192.168.101.128:7005,192.168.101.128:7006

在这里我们将所有的session交由给redis进行管理，直接加一个配置类。

package com.lzq.config;

import org.springframework.context.annotation.Configuration;
import org.springframework.session.data.redis.config.annotation.web.http.EnableRedisHttpSession;

/**
 * @author huangyueyue
 * @version 1.0
 * @date 2021-12-17 22:23
 */
@Configuration
@EnableRedisHttpSession
public class redisSessionConfig {
}

最后直接整个控制器来进行测试：启动项目访问当前接口，可以发现这里一直刷新而size的大小始终是2，这是为什么呢？在这里我们没获取到的session，将session设置进去，也就相当于将当前的session设置到了redis当中，而每一次取的session也是从redis当中进行获取的，而这里并没有对redis当中session进行持续更新，所以每一次存到redis当中的session一直是最新的，之后再获取session，只有一个，再add一个到list当中，这list的size就永远都是2.

package com.lzq.controller;

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @author huangyueyue
 * @version 1.0
 * @date 2021-12-17 22:23
 */
@RestController
public class HelloController {

    @GetMapping("/hello")
    public void hello(HttpServletRequest request , HttpServletResponse response) throws IOException {
        List<String> list = (List<String>) request.getSession().getAttribute("list");
        if(list == null){
            list = new ArrayList<>();
            request.getSession().setAttribute("list",list);
        }
        list.add("xxx");
		// 每一次add之后要重新设置session
		// request.getSession().setAttribute("list",list);
        response.getWriter().println("size = " + list.size() );
        response.getWriter().println("seesionId = " + request.getSession().getId());
    }
    
    @GetMapping("/logout")
    public void logout(HttpServletRequest request){
        request.getSession().invalidate();
    }
}