【分布式缓存】全新Redis6全部知识点，零基础入门

1.分布式缓存Redis6安装

1.1.缓存和队列简介

高并发必备两大“核心技术”

（1）什么是队列（MQ消息中间件）

全称：MessageQueue，主要用于程序与程序之间的通信（异步+解耦）。
核心应用：
(1)解耦：订单系统->物流系统
(2)异步：用户注册同时发送优惠劵，和初始化操作
(3)削峰：秒杀、日志处理
1
2
3
4
5

（2）什么是缓存

程序需要经常调用的数据放在内存中，因为内存中的响应非常快，使其快速调用，避免去数据库持久层去查。
主要就是提高性能 DNS缓存、前端缓存、代理缓存服务器Nginx、应用程序缓存、数据库缓存
1
2

1.2.本地缓存和分布式缓存介绍

（1）分布式缓存

与应用隔离的缓存组件或服务，与本地服务隔离的一个独立的缓存服务，多个服务可共享这一个缓存，多个节点共享缓存，需要考虑带宽。
1

常见的分布式缓存：Redis、Memcached
1

（2）本地缓存

和业务程序一起的缓存，例如mybatis的一二级缓存，只能由服务本身调用，不能多节点共享，不需要考虑带宽。
1

常见的本地缓存：guava、redis也可以做本地缓存、SpringCache
1

（3）本地缓存和分布式缓存的选择

结合业务去选择缓存，高并发的项目一般分布式缓存和本地缓存都存在。
1

（4）热点key的解决方案

热点key一般都放在本地缓存中，因为不需要带宽，效率很高，先去本地缓存中去查找，没有的话再去分布式缓存中查找。
应用：热点新闻、热卖商品、大V明星结婚
1
2

1.3.Nosql和Redis简介

• 什么是Redis
  • 其两者最重要的区别是NoSQL不使用SQL作为查询语言。
  • NoSQL数据存储可以不需要固定的表格模式
  • 键 - 值对存储，列存储，文档存储，图形数据库
  • NoSql：redis、memcached、mongodb、Hbase

一个开源的使用 ANSI C 语言编写、遵守 BSD 协议、支持网络、可基于内存、分布式、可选持久性的键值对(Key-Value)存储数据库，并提供多种语言的 API
1

高性能：Redis能读的速度是110000次/s，写的速度是81000次/s
1

内存中的存储结构，它可以做为消息中间件、缓存、数据库。如：lists（列表）、hashs（散列）、sorted sets（有序集合）、sets（集合）、strings（字符串）
1

1.4.Linux源码安装Redis6

（1）源码安装Redis上传到linux服务器（先安装升级gcc新版才能编译）

#安装gcc
yum install -y gcc-c++ autoconf automake

#centos7默认的gcc是4.8.5版本，版本小于5.3无法编译，要先安装gcc新版才能编译
gcc -v（查看gcc当前版本）

#升级新版gcc，配置永久生效
yum -y install centos-release-scl
yum -y install devtoolset-9-gcc devtoolset-9-gcc-c++ devtoolset-9-binutils

scl enable devtoolset-9 bash #从gcc4.8.5切换到gcc9编译器
echo "source /opt/rh/devtoolset-9/enable" >>/etc/profile

#解压redis安装包
tar -xvf redis.6.2.1.tar.gz
mv redis.6.2.1 redis6

#编译redis
cd redis6
make

#安装到指定的目录
mkdir -p /usr/local/redis
make PREFIX = /usr/local/redis install
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

• 注意：安装编译redis6需要升级gcc，默认自带的gcc版本比较老

• redis-server：redis启动文件

• redis-cli：redis客户端

• redis.conf：redis配置文件

1.5.Docker容器化部署Redis6

云计算+容器化是当下的主流，也是未来的趋势, docker就是可
以快速部署启动应⽤
实现虚拟化，完整资源隔离，⼀次编写，四处运⾏
但有⼀定的限制，⽐如Docker是基于Linux 64bit的，⽆法在
32bit的linux/Windows/unix环境下使⽤
1
2
3
4
5

（1）Docker安装

#安装并运行Docker
yum install -y docker-io 

#启动docker
systemctl start docker

#检查安装结果
docker info

#启动使用docker
systemctl start docker 	#运行Docker守护进程
systemctl stop docker  	#停止Docker守护进程
systemctl restart docker#重启Docker守护进程

docker ps	#查看容器
docker stop 容器id #停掉某个容器

#修改镜像文件
vim /etc/docker/daemon.json

{
"debug":true,"experimental":true,
"registry-mirrors":["https://pb5bklzr.mirror.aliyuncs.com","https://hubmirror.c.163.com","https://docker.mirrors.ustc.edu.cn"]
}

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

（2）Docker部署redis并配置密码

docker run -itd --name xdclass-redis -p 6379:6379 redis --requirepass 123456

-i:以交互模式运行容器，通常与-t同事使用。
-d:后台运行容器，并返回容器ID。
1
2
3
4

1.6.分布式缓存Redis6核心配置

（1）redis.conf配置文件的核心配置

daemonize yes	#配置偶后台运行，默认是no
bind ip号	   #绑定指定ip访问，0.0.0.0是不限制，配置多个ip用空格隔开，bind 192.168.10.1 192.168.10.2
port 端口号	  #端口号，默认6379
requirepass		#密码配置
dbfilename		#配置redis持久化文件名称
dir				#配置redis持久化文件存储地址
save			#配置redis持久化机制
1
2
3
4
5
6
7

（2）在redis安装目录创建log、data、conf目录

日志：/usr/local/redis/log
数据：/usr/local/redis/data
配置文件：/usr/local/redis/conf
1
2
3

（3）在/usr/local/redis/conf中创建自定义的配置文件

touch /usr/local/redis/conf/redis.conf
1

vi /usr/local/redis/conf/redis.conf
1

#任何ip都可以访问
bind 0.0.0.0
#守护进程
daemonize yes
#密码
requirepass	123456
#日志文件
logfile "/usr/local/redis/log/redis.log"
#持久化文件名称
dbfilename xdclass.rdb
#持久化文件路径
dir "/usr/local/redis/data"
#持久化策略，10s内有一个key改动，执行快照
save 10 1
1
2
3
4
5
6
7
8
9
10
11
12
13
14

（4）指定配置文件启动redis

/usr/local/redis/bin/./redis-server /usr/local/redis/conf/redis.conf
1

查看日志确定是否启动：tail -f /usr/local/redis/log/redis.log

2.分布式缓存Redis6数据结构

2.1.Redis6常见数据结构

（1）exists 判断key是否存在

exists name	#判断name这个key是否存在
1

（2）del 删除key

del name #删除name这个key
1

（3）type 判断key的类型

type name #判断name是什么类型
1

（4）ttl 查看key的存活时间

ttl name #判断name还有多长时间过期
ttl age	 #判断age还有多长时间过期
1
2

2.2.Redis6数据结构之String类型

• 简介：存储字符串类型的key-value
• 应用场景：验证码、计数器、订单重复提交、用户登录信息、商品详情

常用命令：

（1）set/get 设置和获取key-value

设置key-value：set user:name lixiang
获取key：get user:name
1
2

（2）incr 对指定key的value进行自增1

incr user:age
1

（3）incrby 对指定key的value进行+n操作

incrby user:age 10
1

（4）mget/mset 一次获取多个key值，一次设置多个key-value

设置：mset user:addr tianjin user:phone 1333333333
获取：mget user:addr user:phone
1
2

（5）setex 设置一个key-value带有过期时间的

setex user:code 30 236589 
1

（6）setnx 当key不存在时，才设置key-value，key存在时，不做操作

setnx user:name xxxxx
1

2.3.Redis6数据结构之List类型

• 简介：字符串列表，按照插入顺序排序，双向链表，插入删除时间复杂度为O(1)快，查找为O(n)慢。
• 应用场景：简单队列、最新商品列表、评论列表、非实时排行榜

常用命令：

（1）lpush 将一个或者多个值插入到列表头部，从左边开始插入

lpush phone:rank:daily iphone
1

（2）rpush将一个或者多个值插入到列表尾部，从右边开始插入

rpush phone:rank:daily xiaomi
1

（3）lrange获取指定key下边的范围元素，0代表第一个，1代表第二个，-1代表最后一个

lrange phone:rank:daily 0 -1
1

（4）llen获取当前key的元素个数

llen phone:rank:daily
1

（5）lindex获取当前索引元素的值

lindex phone:rank:daily 2
1

（6）lpop从顶部弹出一个元素，从左边弹出

lpop phone:rank:daily 1
1

（7）rpop从底部弹出一个元素，从右边弹出

rpop phone:rank:daily 1
1

（8）lrem 删除一个元素，可以指定移除个数

lrem word 2 a
1

（9）brpop移除并且获取列表的最后一个元素，如果列表没有元素会阻塞设置的时长或者在规定的时间内弹出元素为止

brpop word 10
1

2.4.Redis6数据结构之Hash类型

• 简介：Hash是一个string类型的field和value的映射表，hash特别适用于存储对象。
• 应用场景：购物车存储、用户个人信息存储、商品详情存储
• 注意：每个hash可以存储2[^32] -1 键值对（40多亿）

常用命令：

（1）hset/hget 设置和获取key中指定字段的值

设置key-value：hset product:daily:1 title iphone
获取key：hset product:daily:1 title iphone
1
2

（2）hdel 删除指定key的指定字段

hdel product:daily:1 title
1

（3）hmset/hmget 一次设置和获取多个key中指定字段的值

hmget product:daily:1 title color
1

（4）hgetall 获取指定key的全部字段的值

hgetall product:daily:1 
1

（5）hincrby 对指定key的指定字段进行+n操作（n可以为正数也可以为负数）

hincrby product:daily:1 price 100
1

（6）hexists 判断指定key的指定字段是否存在

hexits product:daily:1 color
1

2.5.Redis6数据结构之Set类型

• 简介：将一个或者多个成员元素加入到集合中，已经存在的成员元素将被忽略
• 应用场景：去重、社交应用关注、粉丝、共同好友、大数据里面用户画像标签

常用命令：

（1）sadd添加一个或者多个指定的member元素到集合中，若集合中已存在元素，将被忽略

sadd user:tags:1 woman bwn 18-25 beijing
1

（2）smembers 获取当前集合中的所有元素

smembers user:tags:1
1

（3）srem 删除集合中某个元素

srem user:tags:1 bwn
1

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-5PBFpJkt-1667526366898)(图片\3.5（3）.jpg)]

（4）scard 返回集合中所有元素的个数

scard user:tags:1
1

（5）sismember 返回集合中是否存在当前元素

slsmember user:tags:1 bwn
1

（6）sdiff 返回两个集合中的差集

sdiff user:tags:2 user:tags:1
1

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-bJckZ5vW-1667526366899)(图片\3.5（6）.jpg)]

（7）sinter返回两个集合中的交集

siner user:tags:2 user:tags:1
1

（8）sunion 返回两个集合的并集

sunion user:tags:2 user:tags:1
1

2.6.Redis6数据结构之SortedSet类型

• 简介：用于将一个或者多个成员元素及分数值加入到有序集合中
• 应用场景：实时排行榜：商品热销榜、体育类应用热门球队、积分榜、优先级任务、朋友圈 文章点赞-取消
• 注意：底层采用Ziplist压缩列表和“跳跃表”两种存储结构，如果重复添加相同的元素数据，score值将被覆盖，保留最后一次修改的结果。

常用命令：

（1）zadd向有序集合中添加一个或者多个成员，或者更新已经存在的成员的分数

zadd video:rank 90 springcloud 80 springBoot 70 nginx 60 html 50 javascript 40 linux 
1

（2）zcard获取有序集合中成员数

zcard video:rank
1

（3）zcount计算指定分数区间的成员的个数

zcount video:rank 0 50
1

（4）zincrby 有序集合中对指定的成员的分数+n（n可以为正数也可以为负数）

zincrby video:rank 5 linux
1

（5）zrange 返回索引区间的所有元素，从小到大

zrange video:rank 0 -1
1

（6）zrevrange返回索引区间的所有元素，从大到小

zrevrange video:rank 0 -1
1

（7）zrank 返回有序集合中指定成员的索引，从小到大返回

zrank video:rank 1
1

（8）zrevrank返回有序集合中指定成员的索引，从大到小排序

zrevrank video:rank 1
1

（9）zrem移除有序集合中的一个或者多个成员

zrem video:rank linux
1

（10）zscore返回有序集合中，成员的分数值

zscore video:rank springBoot
1

3.SpringBoot2.x整合Redis6客户端

3.1.分布式缓存Redis客户端讲解

• 自带客户端：redis-cli

• java语言客户端

• jedis

  Jedis是直连模式，在多个线程间共享一个Jedis时是线程不安全的，需要使用连接池
  其API提供了比较全面的Redis命令支持，相比其他Redis封装框架更加原生
  Jedis中的方法调用是比较底层的暴漏的Redis的API，Java方法基本和Redis的API保持这一致
  使用阻塞的I/O，方法调用同步，程序流需要等到socket处理完I/O才能执行，不支持异步操作
  1
  2
  3
  4

• lettuce

  高级Redis客户端，用于线程安全同步，异步响应
  基于Netty的的事件驱动，可以在多个线程间并发访问, 通过异步的方式可以更好的利用系统资源
  1
  2

3.2.新版SpringBoot2.x项目创建

（1）相关软件环境

• JDK1.8+以上
• Maven3.5+

（2）加入依赖

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
1
2
3
4
5
6
7
8

注意：

• springBoot2后默认使用Lettuce作为redis的客户端
• 旧版本的Lettuce踩在堆外内存溢出的bug，5.3版本修复了 这个bug。
• 解决：升级版本或者换jedis

（3）SpringDataRedis的RedisTemplate介绍

• RedisTemplate介绍

  • ValueOperations：简单K-V操作
  • SetOperations：set类型数据操作
  • ZSetOperations：zset类型数据操作
  • HashOperations：针对map类型的数据操作
  • ListOperations：针对List类型的数据操作

• RedisTemplate和StringRedisTemplate的区别

  • StringRedisTemplate继承RedisTemplate
  • 两者的数据是不互通的（默认的序列化机制导致key不一样）
  • RedisTemplate默认采用的是JDK的序列化策略，会将数据先序列化成字节数组在存入Redis中
  • 总结：
    • 当Redis数据库里面操作的都是字符串数据的时候，那使用StringRedisTemplate即可
    • 数据是复杂的对象类型，那么使用RedisTemplate是更好的选择

（4）Redis序列化和反序列化机制

• 同个key为啥获取不到值呢，核心就是序列化机制导致key值不一致
• 什么是序列化

  • 把对象转化为字节序列的过程就称为对象的序列化

  • 把字节序列恢复成对象的过程就是反序列化

  • 对象字节序列化主要有两种用途

    • 把对象的字节序列永久的保存在硬盘上，通常放在一个文件上
    • 在网络上传输对象的字节序列

（5）Redis为什么要序列化

• 性能可以提高，不同的序列化方式性能不一样
• 可视化工具更好查看
  • 采用默认的jdk方式会乱码（POJO类需要实现Serializable接口）
  • 采用JSON方式则不用，且可视化工具更好的查看

（6）自定义Redis序列化方式，可以选择多种选择策略

• JdkSerializationRedisSerializer
  • POJO对象的存存储场景，使用JDK本身序列化机制
  • 默认机制ObjectInputStream/ObjectOutputStream进行序列化操作
• StringRedisSerializer
  • key或者value为字符串
• Jackson2JsonRedisSerializer
  • 利用jackson-json工具，将POJO实例序列化成json格式存储
• GenericFastJsonRedisSerializer
  • 另一种javabean与json之间的转化，同时也需要指定Class类型

3.3.RedisTemplate序列化机制配置

@Configuration
public class RedisTemplateConfiguration {

    /**
     * 自定义redisTemplate配置
     * @param redisConnectionFactory
     * @return
     */
    @Bean
    public RedisTemplate<Object,Object> redisTemplate(RedisConnectionFactory redisConnectionFactory){

        //默认使用JDK的序列化方式
        RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory);

        //使用Jackson2JsonRedisSerialize替换默认序列化
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new 		                       	
        Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper objectMapper = new ObjectMapper();
        //指定要序列化的域，field,get和set,以及修饰符范围，ANY是都有包括private和public
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);

        //设置Key和Value的序列化规则
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);

        //设置hashKey和hashValue的的序列化规则
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);

        //设置支持事务
        //redisTemplate.setEnableTransactionSupport(true);

        //初始化RedisTemplate
        redisTemplate.afterPropertiesSet();

        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
33
34
35
36
37
38
39
40
41
42

3.4.Jedis+Lettuce客户端连接池配置

（1）lettuce连接池

 <dependency>
      <groupId>org.apache.commons</groupId>
      <artifactId>commons-pool2</artifactId>
 </dependency>
1
2
3
4

（2）配置application.properties

#连接池最大连接数（使用负数表示没有限制）
spring.redis.lettuce.pool.max-active = 10

#连接池中的最大空闲连接
spring.redis.lettuce.pool.max-idle = 10

#连接池中的最小空闲连接
spring.redis.lettuce.pool.min-idle = 0

#连接池最大阻塞等待时间（使用负数表示没有限制）
spring.redis.lettuce.pool.max-wait = -1ms

#指定客户端
spring.redis.client-type = lettuce
1
2
3
4
5
6
7
8
9
10
11
12
13
14

（3）配置application.yml

server:
  port: 8080
spring:
  redis:
    host: 8.140.116.67
    port: 6379
    password: 123456
    client-type: jedis
    lettuce:
      pool:
        #连接池的最大连接数（负数表示没有限制）
        max-active: 10
        #连接池中的最大空闲连接
        max-idle: 10
        #连接池的最小空闲连接
        min-idle: 0
        #连接池最大阻塞的等待时间（负数表示没有限制）
        max-wait: -1ms
    jedis:
      pool:
        #连接池的最大连接数（负数表示没有限制）
        max-active: 10
        #连接池中的最大空闲连接
        max-idle: 10
        #连接池的最小空闲连接
        min-idle: 0
        #连接池最大阻塞的等待时间（负数表示没有限制）
        max-wait: -1ms
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

（4）Jedis连接池介绍（可以不排除lettuce依赖包）

	<dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-data-redis</artifactId>
      <!-- 可以不排除lettuce依赖包 -->
      <exclusions>
        <exclusion>
          <groupId>io.lettuce</groupId>
          <artifactId>lettuce-core</artifactId>
        </exclusion>
      </exclusions>
    </dependency>
    
    <!--不用指定版本号，本身spring-data-redis里面有-->
    <dependency>
      <groupId>redis.clients</groupId>
      <artifactId>jedis</artifactId>
    </dependency>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

4.String数据结构实战

4.1.图形验证码存入Redis实战

• 背景：
• 注册-登录-修改密码一版需要发送验证码，但是容易被攻击而已调用。
• 如何避免自己的网站被刷呢
  • 增加图形验证码
  • 单IP请求次数限制
  • 限制号码发送

（1）Kaptcha框架介绍

• 验证码的字体/大小/颜色
• 验证码内容的范围（数字、字母、中文汉字）
• 验证码图片的大小，边框，边框粗细，边框颜色
• 验证码的干扰线，验证码的样式

（2）添加Kaptcha依赖

        <!--kaptcha依赖包 (图形验证码)-->
        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>kaptcha-spring-boot-starter</artifactId>
            <version>1.0.0</version>
        </dependency>
1
2
3
4
5
6

（3）代码配置，编写CaptchaConfig类

@Configuration
public class CaptchaConfig {
    /**
     * 验证码配置
     * Kaptcha配置类名
     *
     * @return
     */
    @Bean
    @Qualifier("captchaProducer")
    public DefaultKaptcha kaptcha() {
        DefaultKaptcha kaptcha = new DefaultKaptcha();
        Properties properties = new Properties();
        //验证码个数
        properties.setProperty(Constants.KAPTCHA_TEXTPRODUCER_CHAR_LENGTH, "4");
        //字体间隔
        properties.setProperty(Constants.KAPTCHA_TEXTPRODUCER_CHAR_SPACE,"8");
        //干扰线颜色

        //干扰实现类
        properties.setProperty(Constants.KAPTCHA_NOISE_IMPL, "com.google.code.kaptcha.impl.NoNoise");

        //图片样式
        properties.setProperty(Constants.KAPTCHA_OBSCURIFICATOR_IMPL, "com.google.code.kaptcha.impl.WaterRipple");

        //文字来源
        properties.setProperty(Constants.KAPTCHA_TEXTPRODUCER_CHAR_STRING, "0123456789");
        Config config = new Config(properties);
        kaptcha.setConfig(config);
        return kaptcha;
    }
}
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

（4）编写统一返回工具类

public class JsonData {

    /**
     * 状态码 0 表示成功
     */

    private Integer code;
    /**
     * 数据
     */
    private Object data;
    /**
     * 描述
     */
    private String msg;


    public JsonData(int code,Object data,String msg){
        this.data = data;
        this.msg = msg;
        this.code = code;
    }

    /**
     * 成功，不传入数据
     * @return
     */
    public static JsonData buildSuccess() {
        return new JsonData(0, null, null);
    }

    /**
     *  成功，传入数据
     * @param data
     * @return
     */
    public static JsonData buildSuccess(Object data) {
        return new JsonData(0, data, null);
    }

    /**
     * 失败，传入描述信息
     * @param msg
     * @return
     */
    public static JsonData buildError(String msg) {
        return new JsonData(-1, null, msg);
    }

    public Integer getCode() {
        return code;
    }

    public void setCode(Integer code) {
        this.code = code;
    }

    public Object getData() {
        return data;
    }

    public void setData(Object data) {
        this.data = data;
    }

    public String getMsg() {
        return msg;
    }

    public void setMsg(String msg) {
        this.msg = msg;
    }
}

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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

（5）编写CommonUtil工具类(获取前台请求ip和md5方法)

public class CommonUtil {

    /**
     * 获取ip
     * @param request
     * @return
     */
    public static String getIpAddr(HttpServletRequest request) {
        String ipAddress = null;
        try {
            ipAddress = request.getHeader("x-forwarded-for");
            if (ipAddress == null || ipAddress.length() == 0 || "unknown".equalsIgnoreCase(ipAddress)) {
                ipAddress = request.getHeader("Proxy-Client-IP");
            }
            if (ipAddress == null || ipAddress.length() == 0 || "unknown".equalsIgnoreCase(ipAddress)) {
                ipAddress = request.getHeader("WL-Proxy-Client-IP");
            }
            if (ipAddress == null || ipAddress.length() == 0 || "unknown".equalsIgnoreCase(ipAddress)) {
                ipAddress = request.getRemoteAddr();
                if (ipAddress.equals("127.0.0.1")) {
                    // 根据网卡取本机配置的IP
                    InetAddress inet = null;
                    try {
                        inet = InetAddress.getLocalHost();
                    } catch (UnknownHostException e) {
                        e.printStackTrace();
                    }
                    ipAddress = inet.getHostAddress();
                }
            }
            // 对于通过多个代理的情况，第一个IP为客户端真实IP,多个IP按照','分割
            if (ipAddress != null && ipAddress.length() > 15) {
                // "***.***.***.***".length()
                // = 15
                if (ipAddress.indexOf(",") > 0) {
                    ipAddress = ipAddress.substring(0, ipAddress.indexOf(","));
                }
            }
        } catch (Exception e) {
            ipAddress="";
        }
        return ipAddress;
    }



    public static String MD5(String data)  {
        try {
            java.security.MessageDigest md = MessageDigest.getInstance("MD5");
            byte[] array = md.digest(data.getBytes("UTF-8"));
            StringBuilder sb = new StringBuilder();
            for (byte item : array) {
                sb.append(Integer.toHexString((item & 0xFF) | 0x100).substring(1, 3));
            }
            return sb.toString().toUpperCase();
        } catch (Exception exception) {
        }
        return null;

    }

}

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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

（6）编写生成验证码存入Redis的逻辑

@RestController
@RequestMapping("/api/v1/captcha")
public class CaptchaController{
    
    @Autowired
    private StringRedisTemplate redisTemplate;
    
    @Autowired
    private Producer captchaProduct;
    
    @GetMapping("/get_captcha")
    public void getCaptcha(HttpServletRequest request,HttpServletResponse response){
       /**
         * 获取随机的验证码
         */
        String captchaProducerText = captchaProducer.createText();

        String key = getCaptchaKey(request);
        
        //放在Redis10分钟过期
        redisTemplate.opsForValue().set(key,captchaProducerText,10, TimeUnit.MINUTES);

        BufferedImage image = captchaProducer.createImage(captchaProducerText);
        
        ServletOutputStream outputStream = null;

        try{
            outputStream = response.getOutputStream();

            ImageIO.write(image,"jpg",outputStream);

            outputStream.flush();

            outputStream.close();
        }catch (Exception e){
            e.printStackTrace();
        }
        
    }
    
    @GetMapping("/send_code")
    public JsonData sendCode(@RequestParam(value = "to",required = true)String to,
                             @RequestParam(value = "captcha",required = true) String captcha,
                             HttpServletRequest request){

        String key = getCaptchaKey(request);

        String cacheCaptcha = redisTemplate.opsForValue().get(key);

        if(cacheCaptcha != null && captcha != null && cacheCaptcha.equalsIgnoreCase(captcha)){
            //匹配通过一定要删除当前key
            redisTemplate.delete(key);

            //TODO 发送验证码逻辑

            return JsonData.buildSuccess();
        }else{
            return JsonData.buildError("图形验证码不正确");
        }

    }
    
    /* *
       *  获取存在缓存中的key用请求的ip和请求头，md5加密
       */
    private  String getCaptchaKey(HttpServletRequest request){
        
        String ip = CommonUtil.getIpAddr(request);

        String userAgent = request.getHeader("User-Agent");

        String key = "user-service:captcha:"+CommonUtil.MD5(ip+userAgent);

        return key;
    }
}
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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-32hhuS5r-1667526366900)(图片\5.1（3）.jpg)]

4.2.高并发商品首页热点数据开发实战

（1）热点数据

• 经常会被查询，但是不经产被修改或者删除的数据
• 首页-详情页

（2）链路逻辑

• 检查缓存是否存在
• 缓存不存在则查询数据库
• 查询数据库的结果放到缓存中，设置过期时间
• 下次访问则命中缓存

（3）接口开发

• 实体类编写，商品项，商品卡片

//商品卡片实体类，里面有多个商品
public class VideoCardDO {

    private  String title;

    private int id;

    private int weight;

    List<VideoDO> videoDOList;

    public String getTitle() {
        return title;
    }

    public void setTitle(String title) {
        this.title = title;
    }

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public int getWeight() {
        return weight;
    }

    public void setWeight(int weight) {
        this.weight = weight;
    }

    public List<VideoDO> getVideoDOList() {
        return videoDOList;
    }

    public void setVideoDOList(List<VideoDO> videoDOList) {
        this.videoDOList = videoDOList;
    }
}
//商品实体类
public class VideoDO {

    private int id;

    private String title;

    private String img;

    private int price;

    public VideoDO() {
    }

    public VideoDO(String title, String img, int price) {
        this.title = title;
        this.img = img;
        this.price = price;
    }

    public VideoDO(int id, String title, String img, int price) {
        this.id = id;
        this.title = title;
        this.img = img;
        this.price = price;
    }

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public String getTitle() {
        return title;
    }

    public void setTitle(String title) {
        this.title = title;
    }

    public String getImg() {
        return img;
    }

    public void setImg(String img) {
        this.img = img;
    }

    public int getPrice() {
        return price;
    }

    public void setPrice(int price) {
        this.price = price;
    }
}
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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102

• Dao层编写,这块采用模拟数据库查询

@Repository
public class VideoCardDao {

    public List<VideoCardDO> list(){

        try {
            TimeUnit.MILLISECONDS.sleep(200);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        List<VideoCardDO> cardDOList = new ArrayList<>();

        VideoCardDO videoCardDO1 = new VideoCardDO();

        VideoDO videoDO1 = new VideoDO(1,"SpringCloud","xxxxxxxxxxxxx",1000);
        VideoDO videoDO2 = new VideoDO(2,"Netty","xxxxxxxxxxxxx",234);
        VideoDO videoDO3 = new VideoDO(3,"面试专题视频","xxxxxxxxxxxxx",3564);
        VideoDO videoDO4 = new VideoDO(4,"AlibabaCloud","xxxxxxxxxxxxx",123);
        VideoDO videoDO5 = new VideoDO(5,"Dubbo","xxxxxxxxxxxxx",445);

        videoCardDO1.setId(1);
        videoCardDO1.setTitle("热门视频");
        List<VideoDO> videoDOS = new ArrayList<>();

        videoDOS.add(videoDO1);
        videoDOS.add(videoDO2);
        videoDOS.add(videoDO3);
        videoDOS.add(videoDO4);
        videoDOS.add(videoDO5);

        videoCardDO1.setVideoDOList(videoDOS);

        cardDOList.add(videoCardDO1);

        VideoCardDO videoCardDO2 = new VideoCardDO();

        VideoDO videoDO6 = new VideoDO(1,"SpringCloud","xxxxxxxxxxxxx",1000);
        VideoDO videoDO7 = new VideoDO(2,"Netty","xxxxxxxxxxxxx",234);
        VideoDO videoDO8 = new VideoDO(3,"面试专题视频","xxxxxxxxxxxxx",3564);
        VideoDO videoDO9 = new VideoDO(4,"AlibabaCloud","xxxxxxxxxxxxx",123);
        VideoDO videoDO10 = new VideoDO(5,"Dubbo","xxxxxxxxxxxxx",445);

        videoCardDO1.setId(1);
        videoCardDO1.setTitle("项目实战");
        List<VideoDO> videoDOS2 = new ArrayList<>();

        videoDOS2.add(videoDO6);
        videoDOS2.add(videoDO7);
        videoDOS2.add(videoDO8);
        videoDOS2.add(videoDO9);
        videoDOS2.add(videoDO10);

        videoCardDO2.setVideoDOList(videoDOS2);

        cardDOList.add(videoCardDO2);

        return cardDOList;
    }

}
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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

• service层编写

public interface VideoCardService {

    List<VideoCardDO> list();

}

@Service
public class VideoCardServiceImpl implements VideoCardService {

    @Autowired
    private VideoCardDao videoCardDao;
    
    @Autowired
    private RedisTemplate redisTemplate;
    
    private static final String VIDEO_CARD_CACHE_KEY = "video:card:key";

    @Override
    public List<VideoCardDO> list() {
        
        Object cacheObj = redisTemplate.opsForValue().get(VIDEO_CARD_CACHE_KEY);
        
        if(cacheObj != null){
           return (List<VideoCardDO>)cacheObj;
        }else{
            List<VideoCardDO> list = videoCardDao.list();
            redisTemplate.opsForValue().set(VIDEO_CARD_CACHE_KEY,list,10,TimeUtil.MINUTES);
            return list;
        }
    }
}
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

• Controller层

    @Autowired
    private VideoCardService videoCardService;
   
   /**
     * 缓存查找热点卡片
     * @return
     */
    @RequestMapping("/list_cache")
    public JsonData listCache(){
        List<VideoCardDO> list = videoCardService.list();
     	return JsonData.buildSuccess(list);
    }
1
2
3
4
5
6
7
8
9
10
11
12

4.3.Redis6+Lua脚本实现原生分布式锁

（1）分布式锁简介

• 简介：分布试锁核心知识介绍和注意事项

• 背景：保证同一时间只有一个客户端可以对共享资源进行操作

• 案例：优惠劵领券限制次数、商品库存超卖

• 核心：

  • 为了防止分布式系统中的多个线程之间进行相互干扰，我们需要一种分布式协调技术来对这些进程进行调度

  • 利用互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题

• 避免共享资源并发操作导致数据问题

• 加锁：

  • 本地锁：synchronize、lock等，锁在当前进程内，集群部署下依旧存在问题

  • 分布式锁：redis、zookeeper等实现，虽然还是锁，但是多个进程公用锁标记，可以用Redis、Zookeeper、MySql等都可以

• 设计分布式锁应该考虑的东西
  • 排他性：在分布式应用集群中，同一个方法在同一时间只能被一台机器的一个线程执行
  • 容错性：分布式锁一定能得到释放，比如客户端崩溃或者网络中断
  • 满足可重入、高性能、高可用
  • 注意分布式锁的开销、锁粒度

（2）基于Redis实现分布式锁的几种坑

• 实现分布式锁可以用Redis、Zookeeper、MySql数据库这几种，性能最好的是Redis且最容易理解的

• 分布式锁离不开key -value 设置

key是锁的唯一标识，一版按业务来决定命名，比如想要给一种优惠劵活动加锁，key命名为"coupon:id"，value可以用固定值，比如设置成1
1

• 基于redis实现分布式锁，文档：http://www.redis.cn/commands.html#string

• 加锁setnx key value

setnx 的含义就是SET if Not Exists，有两个参数setnx(key,value),该方法是原子性操作。
如果key不存在，则设置当前key成功，返回1。
如果当前key已经存在，则设置当前key失败，返回0。
1
2
3

• 解锁del(key)

得到锁的线程执行完任务，需要释放锁，以便其他线程可以进入，调用del(key)
1

• 配置锁超时expire(key,30s)

客户端崩溃或者网络中断，资源将永会被锁住，即死锁，因此需要给key配置过期时间，以保证即使没有被显示释放，这把锁也要在一定时间后自动释放。
1

• 综合伪代码

methodA(){
	String key = "coupon_66"
	if(setnx(key,1) == 1){
        //注意设置时间和设置key不是原子性
		expire(key,30,TimeUnit.MILLISECONDS)
            
        try{
            //做对应的业务逻辑
            //查询用户是否已经领卷
        }finally{
            del(key)
        }
	}else{
        //睡眠100毫秒，然后自旋调用本方法
    	methodA()
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

• 存在什么问题

多个命令之间不是原子性操作，如setnx和expire之间，如果setnx成功，但是expire失败，且死机，则就是个死锁。

使用原子性命令：设置和配置过期时间 setnx|setex 
如：set key 1 ex 30 nx

redisTemplate.opsFosValue().setIfAbsent("seckill_1",success,30,TimeUnit.MILLISECONDS)
1
2
3
4

业务超时，存在其他线程误删，key30秒过期，假如线程A执行很慢超过30s，则key就被释放了，其他线程B就得到了锁，这个时候线程A执行完成，而B还没有执行完成，结果A把B加的锁给删掉了。

• 进一步细化误删

可以在del释放锁之前做一个判断，验证当前的锁是不是自己加的锁，那value应该是存当前线程的标识或者uuid

methodA(){
	String key = "coupon_66"
	if(setnx(key,1) == 1){
        //注意设置时间和设置key不是原子性
		expire(key,30,TimeUnit.MILLISECONDS)
            
        try{
            //做对应的业务逻辑
            //查询用户是否已经领卷
        }finally{
            //删除锁操作判断是否为当前线程加的
            if(redisTemplate.get(key).equals(value)){
                //还在当前时间规定内
                del(key)
            }
        }
	}else{
        //睡眠100毫秒，然后自旋调用本方法
    	methodA()
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

• 核心还是判断和删除命令不是原子性操作导致的

• 总结

• 加锁+配置过期时间：保证原子性操作

• 解锁：防止误删除、也要保证原子性操作

• 采用Lua脚本+redis,保证多个命令的原子性

（3）Lua脚本+Redis实现分布式锁的编码实现

//获取lock的值和传递的值⼀样，调⽤删除操作返回1，否则返回0
String script = "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
//Arrays.asList(lockKey)是key列表，uuid是参数
Integer result = redisTemplate.execute(new DefaultRedisScript<>(script, Integer.class),Arrays.asList(lockKey), uuid);
1
2
3
4

@Slf4j
@RestController
@RequestMapping("/api/v1/coupon")
public class CouponController {

    @Autowired
    private RedisTemplate redisTemplate;

    @GetMapping("/add")
    public JsonData save(@RequestParam(value = "coupon_id",required = true) int couponId){

        //防止其他线程误删
        String uuid = UUID.randomUUID().toString();

        String lockKey = "lock:coupon:"+couponId;

        lock(couponId,uuid,lockKey);

        return JsonData.buildSuccess();

    }

    private void lock(int couponId,String uuid,String lockKey) {

        Boolean nativeLock = redisTemplate.opsForValue().setIfAbsent(lockKey, uuid, Duration.ofSeconds(30));

        log.info(uuid+"---加锁状态："+nativeLock);

        //定义Lua脚本
        String script = "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";

        if(nativeLock){

            //加锁成功，做相应的业务逻辑
            try{

                //核心业务逻辑
                TimeUnit.SECONDS.sleep(3L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                //解锁操作
                Object result = redisTemplate.execute(new DefaultRedisScript(script, Long.class), Arrays.asList(lockKey), uuid);
                log.info("解锁结果："+result);
            }

        }else{
            //加锁失败进入睡眠5s，然后在自旋调用
            try {
                log.info("加锁失败，睡眠5s，进入自旋");
                TimeUnit.MILLISECONDS.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            lock(couponId,uuid,lockKey);
        }
    }
}

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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

setIfAbsent():
execute():
1
2

5.List数据结构实战

5.1.昨日热销榜单实战

（1）需求：

• 天猫每天的热销榜单，每天更新一次
• 需要支持人工运营替换榜单的位置

（2）企业中流程：

• 定时任务计算昨天那些商品出售的数量最多
• 晚上12点到1点更新到榜单上
• 预留一个接口，支持人工运营

（3）类似场景：

• 京东：热销手机榜单、电脑榜单等
• 百度：搜索热榜

（4）编码实战

• 开发接口

@RequestMapping("rank")
public JsonData videoRank(){
	List<VideoDO> list = redisTemplate.opsForValue().range(RANK_KEY,0,-1);
    return JsonData.buildSuccess(list);
}
1
2
3
4
5

• 测试数据

@Test
public void rankTest(){
    
    String RANK_KEY = "rank:video";
    
    VideoDO video1 = new VideoDO(3,"PaaS⼯业级微服务⼤课","xdclass.net",1099);
    VideoDO video2 = new VideoDO(5,"AlibabaCloud全家桶实战","xdclass.net",59);
    VideoDO video3 = new VideoDO(53,"SpringBoot2.X+Vue3综合实战","xdclass.net",49);
    VideoDO video4 = new VideoDO(15,"玩转23种设计模式+最近实战","xdclass.net",49);
    VideoDO video5 = new VideoDO(45,"Nginx⽹关+LVS+KeepAlive","xdclass.net",89);
    
    //leftPushAll向左边插入，所以放在最后一位的才是首个
    redisTemplate.opsForList().leftPush(RANK_KEY,video6,video5,video4,video3,video2,video1);
    
    //rightPushAll向右边插入，所以首个就是第一个
    //sTemplate.opsForList().leftPush(RANK_KEY,video1,video2,video3,video4,video5);
    
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

6.Hash数据结构实战

6.1.购物车实现案例实战

（1）背景：

• 电商购物车实现，支持买多见商品，每个商品不同数量
• 支持高性能处理

（2）购物车常见的实现方式：

• 实现方式一：存储到数据库

  • 性能存在瓶颈

• 实现方式二：前端本地存储-localstorage，sessionstorage

  • localstorage在浏览器中存储key/value对，没有过期时间
  • sessionstorage在浏览器中存储key/value对，在关闭会话窗口后将会删除这些数据

• 实现方式三：后端存储到缓存redis

  • 可以开启AOF持久化防止重启丢失（推荐）

（2）购物车数据结构介绍

• 一个购物车里面，存在多个购物项
• 所以购物车是一个双层的Map
  • Map<String,Map<String,String>>
  • 第一层Map，key是用户id
  • 第二层Map，key是购物车商品的id，值是购物项的数据

（3）对应redis里面的存储

（4）编码实战

• 实体类VideoDO、CartItemVO、CartVO

public class CartItemVO {

    /**
     * 商品id
     */
    private Integer productId;

    /**
     * 购买数量
     */
    private Integer buyNum;

    /**
     * 商品标题
     */
    private String productTitle;

    /**
     * 图片
     */
    private String productImg;

    /**
     * 商品单价
     */
    private int price;

    /**
     * 总价格
     */
    private int totalPrice;

    public Integer getProductId() {
        return productId;
    }

    public void setProductId(Integer productId) {
        this.productId = productId;
    }

    public Integer getBuyNum() {
        return buyNum;
    }

    public void setBuyNum(Integer buyNum) {
        this.buyNum = buyNum;
    }

    public String getProductTitle() {
        return productTitle;
    }

    public void setProductTitle(String productTitle) {
        this.productTitle = productTitle;
    }

    public String getProductImg() {
        return productImg;
    }

    public void setProductImg(String productImg) {
        this.productImg = productImg;
    }

    public int getPrice() {
        return price;
    }

    public void setPrice(int price) {
        this.price = price;
    }

    public int getTotalPrice() {
        return totalPrice*buyNum;
    }

    public void setTotalPrice(int totalPrice) {
        this.totalPrice = totalPrice;
    }
}

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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81

public class CartVO {

    private List<CartItemVO> cartItemVOS;

    private Integer totalAmount;

    public List<CartItemVO> getCartItemVOS() {
        return cartItemVOS;
    }

    public void setCartItemVOS(List<CartItemVO> cartItemVOS) {
        this.cartItemVOS = cartItemVOS;
    }

    /**
     * 返回购物车总价格
     * @return
     */
    public Integer getTotalAmount() {
        //jdk8新语法
        return cartItemVOS.stream().mapToInt(CartItemVO::getTotalPrice).sum();
    }

    public void setTotalAmount(Integer totalAmount) {
        this.totalAmount = totalAmount;
    }
}
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

• 模拟dao层，数据库根据id返回数据

@Repository
public class VideoDao {
    private static Map<Integer, VideoDO> map = new HashMap<>();
    static {
        map.put(1,new VideoDO(1,"工业级PaaS云平台SpringCloudAlibaba综合项⽬实战(完结)","https://xdclass.net",1099));
        map.put(2,new VideoDO(2,"玩转新版⾼性能RabbitMQ容器化分布式集群实战","https://xdclass.net",79));
        map.put(3,new VideoDO(3,"新版后端提效神器MybatisPlus+SwaggerUI3.X+Lombok","https://xdclass.net",49));
        map.put(4,new VideoDO(4,"玩转Nginx分布式架构实战教程 零基础到⾼级","https://xdclass.net",49));
        map.put(5,new VideoDO(5,"ssm新版SpringBoot2.3/spring5/mybatis3","https://xdclass.net",49));
        map.put(6,new VideoDO(6,"新⼀代微服务全家桶AlibabaCloud+SpringCloud实 战","https://xdclass.net",59));
    }

    /**
     * 模拟返回数据库资源
     * @param videoId
     * @return
     */
    public VideoDO findByVideoId(int videoId){
        return map.get(videoId);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

• JsonUtil工具类开发

public class JsonUtil {

    private static final ObjectMapper MAPPER = new ObjectMapper();

    /**
     * 对象转json字符串的方法
     * @param data
     * @return
     */
    public static String objectToJson(Object data){
        try{
            return MAPPER.writeValueAsString(data);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * json字符串转对象的方法
     * @param jsonData
     * @param beanType
     * @param <T>
     * @return
     */
    public static <T> T jsonToObject(String jsonData,Class<T> beanType){
        try {
            T t = MAPPER.readValue(jsonData, beanType);
            return t;
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return null;
    }

}
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

• 开发VideoCardController，购物车控制层

@RestController
@RequestMapping("/api/v1/cart")
@Slf4j
public class VideoCardController{
    @Autowired
    private VideoDao videoDao;
    
    @Autowired
    private RedisTemplate redisTemplate;
    
        /**
     * 添加到购物车
     * @param videoId
     * @param buyNum
     * @return
     */
    @RequestMapping("/add")
    public JsonData addCart(int videoId,int buyNum){

        /**
         * 获取购物车
         */
        BoundHashOperations<String, Object, Object> myCartOps = getMyCartOps();

        Object cacheObj = myCartOps.get(videoId + "");

        String result = "";

        //当购物车有这个商品,转化成字符串
        if(cacheObj != null){
            result = (String) cacheObj;
        }

        if(cacheObj == null){
            //购物车没这个商品，从数据库里拿出来，在放到缓存中
            CartItemVO cartItemVO = new CartItemVO();
            VideoDO videoDO = videoDao.findByVideoId(videoId);

            cartItemVO.setBuyNum(buyNum);
            cartItemVO.setPrice(videoDO.getPrice());
            cartItemVO.setProductId(videoDO.getId());
            cartItemVO.setProductImg(videoDO.getImg());
            cartItemVO.setProductTitle(videoDO.getTitle());
            cartItemVO.setTotalPrice(videoDO.getPrice()*buyNum);
            myCartOps.put(videoId+"", JsonUtil.objectToJson(cartItemVO));

        }else{
            //不为空就将字符串转成对象，增加商品购买数量，在转成字符串放到redis里
            CartItemVO cartItemVO = JsonUtil.jsonToObject(result, CartItemVO.class);
            cartItemVO.setBuyNum(cartItemVO.getBuyNum()+buyNum);
            myCartOps.put(videoId+"",JsonUtil.objectToJson(cartItemVO));
        }

        return JsonData.buildSuccess();
    }
    
   /**
     * 查看我的购物车
     * @return
     */
    @RequestMapping("/my-cart")
    public JsonData getMyCart(){
        //获取购物车
        BoundHashOperations<String, Object, Object> myCartOps = getMyCartOps();

        List<CartItemVO> cartItemVOS = new ArrayList<>();

        List<Object> itemList = myCartOps.values();

        for (Object item : itemList) {
            CartItemVO cartItemVO = JsonUtil.jsonToObject((String) item, CartItemVO.class);
            cartItemVOS.add(cartItemVO);
        }
        CartVO cartVO = new CartVO();

        cartVO.setCartItemVOS(cartItemVOS);

        return JsonData.buildSuccess(cartVO);
    }

    /**
     * 清空我的购物车
     * @return
     */
    @RequestMapping("/clear")
    public JsonData clear(){
        String cartKey = getCartKey();
        redisTemplate.delete(cartKey);

        return JsonData.buildSuccess();
    }
    
    /*******************通用的方法，获取购物购物车数据，获取当前key*****************/
    /**
     * 获取我的购物车通用方法
     * @return
     */
    private BoundHashOperations<String,Object,Object> getMyCartOps(){
        //获取定义在Hash里的key，指定方法拼接
        String key = getCartKey();
        //返回当前key的集合，没有则新建返回
        return redisTemplate.boundHashOps(key);
    }

    /**
     * 获取购物车的key，用前缀加上用户的id
     * @return
     */
    private String getCartKey(){
        //用户id,获取用户id，JWT解密后获取
        int userId = 88;
        String cartKey = String.format("video:cart:%s", userId);
        return cartKey;
    }
}
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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115

7.Set数据结构实战

7.1.大数据下的用户画像标签去重

（1）简介

• 用户画像 英文User Profile，是根据用户基本属性、社会属性、行为属性、心理属性等真实信息抽象出的一个标签化的、虚拟的用户模型。“用户画像”的实质是对“人”的数字化。
• 应用场景很多，比如个性化推荐、精准营销、金融风控、精细化运营等等，举个例子来理解用户画像的实际实用价值，我们经常用手机网购，淘宝里面的千人千面，通过“标签tag”来对用户的多维度特征进行提炼和标识，那灭个人的用户画像就需要存储，set集合就适合去重。
• 用户画像不止针对某个人，也可以某一人群或行业的画像。

（2）案例

    /**
     * 用户画像去重
     */
    @Test
    public void userProfile(){
        BoundSetOperations operations = redisTemplate.boundSetOps("user:tags:1");

        operations.add("car","student","rich","dog","guangdong","rich");

        Set<String> set1 = operations.members();
        System.out.println(set1);

        operations.remove("dog");

        Set<String> set2 = operations.members();
        System.out.println(set2);

    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

7.2.关注、粉丝、共同好友

（1）背景

• 社交应用里面的关注、粉丝、共同好友案例

（2）案例

public void testSet(){
	BoundSetOperations operationsLW = redisTemplate.boundSetOps("user:lw");
        operationsLW.add("A","B","C","D","E");
        System.out.println("LW的粉丝："+operationsLW.members());

        BoundSetOperations operationsLX = redisTemplate.boundSetOps("user:lx");
        operationsLX.add("A","B","F","Z","H");
        System.out.println("LX的粉丝："+operationsLX.members());

    	//差集
        Set lwSet = operationsLW.diff("user:lx");
		
        System.out.println("lw的专属用户："+lwSet);

        Set lxSet = operationsLX.diff("user:lw");

        System.out.println("lx的专属用户："+lxSet);
		
    	//交集
        Set intersectSet = operationsLW.intersect("user:lx");

        System.out.println("同时关注的用户："+intersectSet);

        Set union = operationsLW.union("user:lx");

    	//并集
        System.out.println("两个人的并集："+union);

        Boolean a = operationsLW.isMember("A");

        System.out.println("用户A是否为lw的粉丝："+a);
}
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

8.SortedSet数据结构实战

8.1.用户积分实时榜单

（1）背景

• 用户玩游戏-积分实时榜单
• IT视频热销实时榜单
• 电商商品热销实时榜单
• 一般的排行榜读多写少，可以对master进行写入操作做，然后多个slave进行读操作

（2）对象准备

public class UserPointVO {

    private String username;

    private String phone;

    public UserPointVO(String username, String phone) {
        this.username = username;
        this.phone = phone;
    }

    public UserPointVO() {
    }

    public String getUsername() {
        return username;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    public String getPhone() {
        return phone;
    }

    public void setPhone(String phone) {
        this.phone = phone;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        UserPointVO that = (UserPointVO) o;
        return Objects.equals(phone, that.phone);
    }

    @Override
    public int hashCode() {
        return Objects.hash(phone);
    }
}

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
41
42
43
44

@Test
    public void testData(){
        UserPointVO p1 = new UserPointVO("老王","13113");
        UserPointVO p2 = new UserPointVO("老A","324");
        UserPointVO p3 = new UserPointVO("老B","242");
        UserPointVO p4 = new UserPointVO("老C","542345");
        UserPointVO p5 = new UserPointVO("老D","235");
        UserPointVO p6 = new UserPointVO("老E","1245");
        UserPointVO p7 = new UserPointVO("老F","2356432");
        UserPointVO p8 = new UserPointVO("老G","532332");

        BoundZSetOperations boundZSetOperations = redisTemplate.boundZSetOps("point:rank:real");

        boundZSetOperations.add(p1,348);
        boundZSetOperations.add(p2,18);
        boundZSetOperations.add(p3,328);
        boundZSetOperations.add(p4,848);
        boundZSetOperations.add(p5,98);
        boundZSetOperations.add(p6,188);
        boundZSetOperations.add(p7,838);
        boundZSetOperations.add(p8,8828);

    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

（3）接口开发

• 返回榜单-从大到小排序

    /**
     * 返回全部榜单从大到小
     * @return
     */
    @RequestMapping("/real-rank2")
    public JsonData rankList2(){
        Set set = redisTemplate.boundZSetOps("point:rank:real").reverseRange(0, -1);
        return JsonData.buildSuccess(set);
    }
1
2
3
4
5
6
7
8
9

• 返回榜单-从小到大排序

    /**
     * 返回全部榜单从小到大
     * @return
     */
    @RequestMapping("/real-rank1")
    public JsonData rankList1(){
        Set range = redisTemplate.boundZSetOps("point:rank:real").range(0, -1);
        return JsonData.buildSuccess(range);
    }
1
2
3
4
5
6
7
8
9

• 查询个人用户排名

    /**
     * 查询个人用户排名
     * @param username
     * @param phone
     * @return
     */
    @RequestMapping("find_my_rank")
    public JsonData find(String username,String phone){

        UserPointVO userPointVO = new UserPointVO(username,phone);

        Long rank = redisTemplate.boundZSetOps("point:rank:real").reverseRank(userPointVO);

        return JsonData.buildSuccess(++rank);

    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

• 查看个人积分

    /**
     * 查看个人积分
     * @param username
     * @param phone
     * @return
     */
    @RequestMapping("find_my_score")
    public JsonData findMyScore(String username,String phone){

        UserPointVO userPointVO = new UserPointVO(username,phone);

        Double score = redisTemplate.boundZSetOps("point:rank:real").score(userPointVO);

        return JsonData.buildSuccess(score);

    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

• 个人加积分

	/**
     * 加积分
     * @param username
     * @param phone
     * @return
     */
    @RequestMapping("add_score")
    public JsonData addScore(String username,String phone){
        UserPointVO userPointVO = new UserPointVO(username,phone);
        redisTemplate.boundZSetOps("point:rank:real").incrementScore(userPointVO,1000000);

        return JsonData.buildSuccess(redisTemplate.boundZSetOps("point:rank:real").reverseRange(0,-1));
    }
1
2
3
4
5
6
7
8
9
10
11
12
13

9.SpringCache+MyBatisPlus整合

9.1.SpringCache+MyBatisPlus整合

（1）SpringCache简介

文档：https://spring.io/guides/gs/caching/

• 自Spring3.1起，提供了类似于@Transactional注解事务的注解Cache支持，且提供了Cache抽象
• 提供基本的Cache抽象，方便切换各种底层Cache
• 只需要更少的代码就可以完成业务数据的缓存
• 提供事务回滚时也自动回滚缓存，支持比较复杂的缓存逻辑
• 核心
  • 一个是Cache接口，缓存操作的API
  • 一个是CacheManager管理各类缓存，有多个缓存框架实现

（2）项目中引入cache的starter

        <!--springCache依赖包-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-cache</artifactId>
        </dependency>
1
2
3
4
5

（3）配置文件指定缓存类型

spring: 
  cache: 
    type: redis
1
2
3

（4）启动类开启缓存注解

@EnableCaching
1

（5）添加数据库依赖

        <!--mybatis plus和springboot整合-->
        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-boot-starter</artifactId>
            <version>3.4.0</version>
        </dependency>
        <!--数据库驱动-->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.15</version>
        </dependency>
1
2
3
4
5
6
7
8
9
10
11
12

（6）增加数据库配置以及mybatisplus日志打印的配置

#配置plus打印sql⽇志
mybatis-plus:
  configuration:
    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
spring: 
  #数据库连接配置
  datasource:
    url: jdbc:mysql://127.0.0.1:3306/redis_test?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai
    username: root
    password: 123456
    driver-class-name: com.mysql.cj.jdbc.Driver
1
2
3
4
5
6
7
8
9
10
11

（7）数据库表建立

CREATE TABLE `product` (
 `id` int(11) unsigned NOT NULL
AUTO_INCREMENT,
 `title` varchar(128) DEFAULT NULL COMMENT '标 题',
 `cover_img` varchar(128) DEFAULT NULL COMMENT
'封⾯图',
 `detail` varchar(256) DEFAULT '' COMMENT '详 情',
 `amount` int(10) DEFAULT NULL COMMENT '新价
格',
 `stock` int(11) DEFAULT NULL COMMENT '库存',
 `create_time` datetime DEFAULT NULL COMMENT
'创建时间',
 PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=5 DEFAULT
CHARSET=utf8mb4;


====================================================
INSERT INTO `product` (`id`, `title`,
`cover_img`, `detail`, `amount`, `stock`,
`create_time`)
VALUES
 (1, 'AlibabaCloud',
'https://file.xdclass.net/video/2020/alibabaclo
ud/zt-alibabacloud.png',
'https://file.xdclass.net/video/2021/60-
MLS/summary.jpeg', 213, 100, '2021-09-12
00:00:00'),
 (2, 'Linux',
'https://file.xdclass.net/video/2020/alibabaclo
ud/zt-alibabacloud.png',
'https://file.xdclass.net/video/2021/59-
Postman/summary.jpeg', 42, 100, '2021-03-12
00:00:00'),
 (3, 'Docker',
'https://file.xdclass.net/video/2020/alibabaclo
ud/zt-alibabacloud.png',
'https://file.xdclass.net/video/2021/60-
MLS/summary.jpeg', 12, 20, '2022-09-22
00:00:00'),
 (4, 'Nginx',
'https://file.xdclass.net/video/2020/alibabaclo
ud/zt-alibabacloud.png',
'https://file.xdclass.net/video/2021/60-
MLS/summary.jpeg', 14, 20, '2022-11-12
00:00:00');
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
41
42
43
44
45
46

（8）数据库表对应的实体类编写

@Data
@TableName("product")
public class ProductDO {

    @TableId(value = "id", type = IdType.AUTO)
    private Long id;
    /**
     * 标题
     */
    private String title;
    /**
     * 封⾯图
     */
    private String coverImg;
    /**
     * 详情
     */
    private String detail;
    /**
     * 新价格
     */
    private Integer amount;
    /**
     * 库存
     */
    private Integer stock;
    /**
     * 创建时间
     */
    private Date createTime;

}
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

（9）开发商品的CRUD和分页查询

• 主类开启对Mapper的支持

//mapper所在的路径
@MapperScan("xxx.xxx.xxx")
1
2

• 编写Mapper

public interface ProductMapper extends BaseMapper<ProductDO> {
}
1
2

• 编写service

@Service
public class ProductServiceImpl implements ProductService {

    @Autowired
    private ProductMapper productMapper;

    @Override
    public int save(ProductDO productDO) {

        int insert = productMapper.insert(productDO);

        return insert;
    }

    @Override
    public int del(int id) {

        int i = productMapper.deleteById(id);

        return i;
    }

    @Override
    public int update(ProductDO productDO) {

        int i = productMapper.updateById(productDO);

        return i;
    }

    @Override
    public ProductDO findById(int id) {

        return productMapper.selectById(id);

    }

    @Override
    public Map<String, Object> page(int page, int size) {

        Page pageInfo = new Page<>(page,size);

        IPage<ProductDO> iPage = productMapper.selectPage(pageInfo, null);

        Map<String,Object> pageMap = new HashMap<>(3);

        pageMap.put("total_record",iPage.getTotal());

        pageMap.put("total_page",iPage.getPages());

        pageMap.put("current_total",iPage.getRecords());

        return pageMap;
    }
}

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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

• 编写controller

@RestController
@RequestMapping("/api/v1/product")
public class ProductController {

    @Autowired
    private ProductService productService;

    @PostMapping("/add")
    public JsonData add(@RequestBody ProductDO productDO){

        int save = productService.save(productDO);

        return JsonData.buildSuccess(save);

    }

    @PostMapping("/update")
    public JsonData update(@RequestBody ProductDO productDO){

        int save = productService.update(productDO);

        return JsonData.buildSuccess(save);

    }

    @GetMapping("/findById")
        public JsonData findById(@RequestParam("product_id") int id){

        ProductDO productDO = productService.findById(id);

        return JsonData.buildSuccess(productDO);

    }

    @DeleteMapping("/del")
    public JsonData del(@RequestParam("product_id") int id){

        int i = productService.del(id);

        return JsonData.buildSuccess(i);

    }

    @GetMapping("/page")
    public JsonData page(@RequestParam("page") int page,@RequestParam("size") int size){

        Map<String, Object> map = productService.page(page, size);

        return JsonData.buildSuccess(map);

    }

}
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
41
42
43
44
45
46
47
48
49
50
51
52
53

9.2.Cacheable注解

（1）Cacheable注解

• 标记在一个方法上，也可以标记在一个类上
• 缓存标注对象的返回结果，标注咋i方法上缓存该方法的返回值，标注在类上缓存所有方法的返回值
• value缓存名称，可以有多个
• key缓存的key规则，可以用springEL表达式，默认是方法参数组合
• condition缓存条件，使用springEL编写，返回true才缓存

（2）用法案例

//对象
@Cacheable(value = {"product"},key="#root.methodName")
//分⻚
@Cacheable(value ={"product_page"},key="#root.methodName +#page+'_'+#size")
1
2
3
4

（3）spEL表达式

• methodName当前被调用的方法名
  • root.methodname
• args当前被调用的方法的参数列表
  • root.args[0]
• result方法执行后的返回值
  • result

9.3.自定义CacheManager

• 在springBoot的configuration类中添加修改redis缓存序列化器和配置manager过期时间

/**
     * 修改Redis缓存序列化器和配置manager过期时间
     */
    @Primary
    @Bean
    public RedisCacheManager cacheManager1Hour(RedisConnectionFactory connectionFactory){
        RedisCacheConfiguration config = instanceConfig(3600L);
        return RedisCacheManager.builder(connectionFactory).cacheDefaults(config).transactionAware().build();
    }

    @Bean
    public RedisCacheManager cacheManager1Day(RedisConnectionFactory connectionFactory){
        RedisCacheConfiguration config = instanceConfig(3600 * 24L);
        return RedisCacheManager.builder(connectionFactory).cacheDefaults(config).transactionAware().build();
    }

    @Bean
    public RedisCacheManager cacheManager10Min(RedisConnectionFactory connectionFactory){
        RedisCacheConfiguration config = instanceConfig(600L);
        return RedisCacheManager.builder(connectionFactory).cacheDefaults(config).transactionAware().build();
    }

    /**
     * 序列化机制
     * @param ttl
     * @return
     */
    private RedisCacheConfiguration instanceConfig(Long ttl) {
        Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class);
        ObjectMapper objectMapper = new ObjectMapper();

        objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
        objectMapper.registerModule(new JavaTimeModule());
        // 去掉各种@JsonSerialize注解的解析

        objectMapper.configure(MapperFeature.USE_ANNOTATIONS, false);
        // 只针对⾮空的值进⾏序列化

        objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
        // 将类型序列化到属性json字符串中

        objectMapper.activateDefaultTyping(LaissezFaireSubTypeValidator.instance,ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY);

        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);
        return
                RedisCacheConfiguration.defaultCacheConfig()

                        .entryTtl(Duration.ofSeconds(ttl))
                        .disableCachingNullValues()
                        .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(jackson2JsonRedisSerializer));
    }
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
41
42
43
44
45
46
47
48
49
50
51

9.4.自定义缓存KeyGenerator

• key规则定义麻烦，支持自定义的规则，同样在springBoot的configuration中进行配置

    /**
     * keyGenerator自定义key的规则
     */
    @Bean
    public KeyGenerator springCacheDefaultKeyGenerator(){
        return new KeyGenerator() {
            @Override
            public Object generate(Object o, Method method, Object... objects) {
                return o.getClass().getSimpleName()+":"+method.getName()+":"+ StringUtils.arrayToDelimitedString(objects,":");
            }
        };
    }
1
2
3
4
5
6
7
8
9
10
11
12

java代码实现

key 属性和keyGenerator属性只能⼆选⼀

@Cacheable(value = {"product"},keyGenerator ="springCacheCustomKeyGenerator", cacheManager ="cacheManager1Minute")
1

9.5.CachePut注解

• CachePut注解

• 根据方法的请求参数对其结果进行缓存，每次都会触发真实方法的调用

• value缓存名称，可以有多个

• key缓存的key规则，可以用springEL表达式，默认是方法参数组合

• condition缓存条件，使用springEL编写，返回true才缓存

@CachePut(value = {"product"},key = "#productDO.id")//常用于修改的方法上，修改数据库，然后修改对应的缓存
1

9.6.CacheEvict注解

• CacheEvict注解

• 从缓存中移除相应数据，触发缓存删除的操作

• value缓存名称，可以有多个

• key 缓存的key规则，可以⽤springEL表达式，默认是⽅法参数组合

• beforeInvocation = false

  • 缓存的清除是否在方法之前执行，默认代表缓存清除操作是在方法执行之后执行
  • 如果出现异常缓存就不会清除

• beforeInvocation = true

  • 代表清除缓存操作实在方法执行之前，无论方法是否出现异常，缓存都清除

@CacheEvict(value = {"product"},key = "#root.args[0]")
1

9.7.Caching注解

• Caching注解

• 组合多个Cache注解使用

• 允许在同一方法上使用多个@Cacheable、@CachePut、@CacheEvict注释

@Caching(
	cacheable = {
        @Cacheable = (value = "product",key = "#id"),
    },
    put = {
        @CachePut(value ="product",key = "#id"),
 		@CachePut(value ="product",key = "'stock:'+#id")
    }
)
1
2
3
4
5
6
7
8
9

10.Redis6持久化配置-RDB和AOF

10.1.Redis6.x持久化操作-RDB

（1）Redis持久化介绍

• Redis时一个内存数据库，如果没有配置持久化，redis重启后数据就会全部丢失。
• 因此开启redis的持久化功能，将数据保存到磁盘上，当redis重启后，可以从磁盘中恢复数据。

（2）两种持久化方式

• RDB（Redis DataBase）
• AOF（append only file）

（3）RDB持久化介绍

• 在指定的时间间隔内将内存中的数据集快照写入磁盘
• 默认的文件名为dump.rdb
• 产生快照的情况
  • save
    • 会阻塞当前Redis服务器，执行save命令期间，Redis不能处理其他命令，直到RDB过程完成为止
  • bgsave
    • fork创建子进程，RDB持久化过程由子进程负责，会在后台异步进行快照操作，快照同时还可以响应客户端请求
  • 自动化
    • 配置文件来完成，配置redis触发Redis的RDB持久化条件，比如“save m n”。表示m秒内数据集存在n次修改时，会自动触发bgsave
  • 主从架构
    • 从服务器同步数据的时候，会发送sync执行同步操作，master主服务器就会执行bgsave

（4）优点和缺点

优点

• RDB文件紧凑，全量备份，适合用于进行备份和灾难恢复
• 在恢复大数据集时的速度比AOF的恢复速度要快
• 生成的是一个紧凑的文件

缺点

• 每次快照时一次全量的备份，fork子进程进行后台操作，子进程存在开销
• 在快照持久化期间修改的数据不会被保存，可能丢失数据

（5）核心配置

• dir 持久化文件的路径
• dbfilename 文件名

#任何ip可以访问
bind 0.0.0.0

#守护进程
daemonize yes

#密码
requirepass 123456

#日志文件
logfile "/user/local/redis/log/redis.log"

#持久化文件名
dbfilename xdclass.rdb

#持久化文件路径
dir /usr/local/redis/data

#关闭rdb
#save ""

#持久化策略，10s内有1个key改动，执行快照
save 10 1

#导出rdb数据库文件压缩字符串和对象，默认时yes，会浪费CPU但是节省空间
rdbcompression yes

#导入时是否检查
rdbchecksum yes
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

（6）配置文件触发

#关闭RDB
save ""

#10秒2个key变动触发RDB
save 10 2

#100秒5个key变动触发RDB
save 100 5
1
2
3
4
5
6
7
8

（7）Linux内存分配策略

0 表示内核将检查是否有足够的可用内存供应用进程使用，如果有足够的可用内存，内存申请允许，否则，内存申请失败，并把错误返回给前台
1 表示内核允许分配所有的物理内存，而不管当前内存状态如何
2 表示内核允许分配超过所有物理内存和交换空间总和的内存

解决方式
echo > /proc/sys/vm/overcommit_memory

持久化配置
vim /etc/sysctl.conf

改为
vm.overcommit_memory=1

修改sysctl.conf后，需要执⾏ sysctl -p 以使⽣效
1
2
3
4
5
6
7
8
9
10
11
12
13
14

10.2.Redis6.x持久化操作-AOF

（1）AOF持久化介绍

• append only file ，追加文件的方式，文件容易被人读懂
• 以独立日志的方式记录每次写命令，重启时在重新执行AOF中的命令达到恢复数据的目的
• 写入过程中宕机，也不会影响之前的数据，可以通过redis-check-aof检查修复问题

（2）配置

• appendonly yes，默认不开启，开启aof持久化
• AOF文件名通过appendfilename配置设置，默认文件名是appendonly.aof
• 存储路径同RDB持久化方式一致，使用dir配置

（3）核心原理

• Redis每次写入命令会追加到aof_buf(缓冲区)
• AOF缓冲区根据对应的策略向磁盘做同步操作
• 高频的AOF会带来影响，特别是每次刷盘

（4）提供了3种同步方式，在性能和安全方面做出平衡

• appendfsync always：每次有数据修改发生时都会写入AOF文件，消耗性能多
• appendfsync everysec：每秒同步一次，该策略为AOF的缺省策略
• appendfsync no：不主从同步，有草祖宗系统自动调度刷磁盘，性能是最好的，但是最不安全

appendonly yes
appendfilename "xdclass.aof"
appendfsync everysec
1
2
3

（5）rewrite重写介绍

• AOF文件越来越大，需要定期对AOF文件进行重写达到压缩
• 旧的AOF文件含有无效命令会被忽略，保留最新的数据命令
• 多条写命令可以合并为一个
• AOF重写降低了文件占用空间
• 更小的AOF文件可以更快的被Redis加载

（6）重写触发配置

• 手动触发

  • 直接调用bgrewriteaof命令

• 自动触发

  • auto-aof-rewrite-min-size：表示运行AOF重写时文件最小体积，默认64mb
  • auto-aof-rewrite-percentage：代表当前AOF文件空间和上一次重写后AOF文件空间（aof_base_size）的比值。

（7）aof常用配置

# 是否开启aof
appendonly yes

# ⽂件名称
appendfilename "appendonly.aof"

# 同步⽅式
appendfsync everysec

# aof重写期间是否同步
no-appendfsync-on-rewrite no

# 重写触发配置
#AOF文件最小重写大小，只有AOF文件大小大于该值的时候才可以重写，默认64mb
auto-aof-rewrite-min-size 64mb
#当前AOF文件大小和最后一次重写后的大小之间的比率等于指定的增长率重写，100 表示当最后一次压缩为150 ，那么就300的时候进行压缩
auto-aof-rewrite-percentage 100


# 加载aof时如果有错如何处理
# yes表示如果aof尾部⽂件出问题，写log记录并继续执⾏。
#no表示提示写⼊等待修复后写⼊
aof-load-truncated yes
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

（8）重写前后对比

10.3.AOF和RDB的选择和混合模式

（1）Redis提供了不同的持久化选项

• RDB持久化以指定的时间间隔执行数据集的时间点快照。
• AOF持久化记录服务器接收的每个写入操作，将在服务器启动时再次读取，重建原始数据集。使与Redis本身相同的格式以仅追加的方式记录命令，当文件太大时，Redis能够重写。

（2）RDB的优缺点

• 优点：

  • RDB最⼤限度地提⾼了Redis的性能，⽗进程不需要参与磁盘I/O
  • RDB⽂件紧凑，全量备份，适合⽤于进⾏备份和灾难恢复
  • 在恢复⼤数据集时的速度⽐ AOF 的恢复速度要快
  • ⽣成的是⼀个紧凑压缩的⼆进制⽂件

• 缺点：

  • 如果您需要在Redis停⽌⼯作时（例如断电后）将数据丢失的可能性降⾄最低，则RDB并不好
  • RDB经常需要fork才能使⽤⼦进程持久存储在磁盘上。如果数据集很⼤，Fork可能会⾮常耗时

（3）AOF的优缺点

• 优点：

  • 数据更加安全
  • 当Redis AOF⽂件太⼤时，Redis能够在后台⾃动重写AOF
  • AOF以易于理解和解析的格式，⼀个接⼀个地包含所有操作的⽇志

• 缺点：

  • AOF⽂件通常⽐同⼀数据集的等效RDB⽂件⼤
  • 根据确切的fsync策略，恢复的时候AOF可能⽐RDB慢

（4）线上系统怎末处理

• RDB持久化与AOF持久化⼀起使⽤
• 如果Redis中的数据并不是特别敏感或者可以通过其它⽅式重写⽣成数据
• 集群中可以关闭AOF持久化，靠集群的备份⽅式保证可⽤性
• ⾃⼰制定策略定期检查Redis的情况，然后可以⼿动触发备份、重写数据
• 采⽤集群和主从同步

（5）Redis4.0后开始的rewrite支持混合模式

• 就是rdb和aof⼀起⽤

• 直接将rdb持久化的⽅式来操作将⼆进制内容覆盖到aof⽂件中,rdb是⼆进制，所以很⼩

• 有写⼊的话还是继续append追加到⽂件原始命令，等下次⽂件过⼤的时候再次rewrite

• 默认是开启状态

• 好处：

  • 混合持久化结合了RDB持久化和AOF持续化的优点，采取了rdb的文件小易于灾难恢复
  • 同时结合AOF，增量的数据以AOF⽅式保存了，数据更少的丢失

• 坏处：

  • 前部分是RDB格式，是⼆进制，所以阅读性较差

• 数据恢复：

  • 先看是否存在aof⽂件，若存在则先按照aof⽂件恢复，aof⽐rdb全，且aof⽂件也rewrite成rdb⼆进制格式
  • 若aof不存在，则才会查找rdb是否存在

11.Redis6服务端配置info+config命令

11.1.info命令介绍

• 服务器的各种信息和统计数值

Server:	#有关redis服务器的常规信息
redis_mode:standalone	#运行模式，单机或者集群
multiplexing_api:epoll	#redis所使用的事件处理机制
run_id:3abd26c33dfd059e87a0279defc4c96c13962e	#redis服务器的随机标识符（用于sentinel和集群）
config_file:/usr/local/redis/conf/redis/conf	#配置文件路径

Clinets: #客户端连接部分
connected_clients:10	#已连接客户端的数量（不包括slave连接的客户端）

Memory: #内存消耗相关信息
userd_memory:874152		#使用内存
used_memory_human:853.66K	#以⼈类可读的格式返回 Redis 分配的内存总量	
user_memory_rss:2834432		#系统给redis分配的内存即常驻内存，和top 、 ps 等命令的输出⼀致
used_memory_rss_human:2.70M	# 以⼈类可读的格式返回系统redis分配的常驻内存top、ps等命令的输出⼀致
used_memory_peak:934040		#系统使用的峰值大小
used_memory_peak_human:912.15K
total_system_memory:1039048704	#操作系统的总字节
total_system_memory_human:990.91M
used_memory_lua:37888 	# lua引擎使⽤的内存
used_memory_lua_human:37.00K
maxmemory:0 #最大内存的配置值，0表示不限制
maxmemory_human:0B
maxmemory_policy:noeviction	 #达到最⼤内存配置值后的策略

Persistence:	#rdb和aof相关信息
rdb_bgsave_in_progress:0	#标识rdb save是否进⾏中
rdb_last_bgsave_status:ok	# 上次的save操作状态
rdb_last_bgsave_time_sec:-1	# 上次rdb save操作使⽤的时间(单位s)
rdb_current_bgsave_time_sec:-1	#如果rdbsave操作正在进⾏，则是所使⽤的时间
aof_enabled:1	#是否开启aof，默认没开启
aof_rewrite_in_progress:0	# 标识aof的rewrite操作是否在进⾏中
aof_last_rewrite_time_sec:-1 	#上次rewrite操作使⽤的时间(单位s)
aof_current_rewrite_time_sec:-1	#如果rewrite操作正在进⾏，则记录所使⽤的时间
aof_last_bgrewrite_status:ok	#上次rewrite操作的状态
aof_current_size:0	# aof当前⼤⼩

Stats:	#一版统计
evicted_keys:0	#因为内存⼤⼩限制，⽽被驱逐出去的键的个数

Replication:	#主从同步信息
role:master		#角色
connected_slaves:1	#连接的从库数
master_sync_in_progress:0	#标识主redis正在同步到从redis

Cluster:	#集群部分
cluster_enabled:0 # 实例是否启⽤集群模式

Keyspace:	#数据库相关统计
db0:keys=4,expires=0,avg_ttl=0 # db0的key的数量,带有⽣存期的key的数,平均存活时间
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
41
42
43
44
45
46
47
48
49

11.2.config命令介绍

• 可以动态的调整Redis服务器的配置（configuration）而无需重启
• config get xxx ，config set key “xxxx”

timeout 	#客户端连接时的超时时间，单位为秒。当客户端在这段时间内没有发出任何指令，那么关闭该连接
databases 	#设置数据库的个数，可以使⽤ SELECT 命令来切换数据库。默认使⽤的数据库是 0
save 		#设置 Redis 进⾏rdb持久化数据库镜像的频率。
rdbcompression #在进⾏镜像备份时，是否进⾏压缩
slaveof 	#设置该数据库为其他数据库的从数据库
masterauth 	#当主数据库连接需要密码验证时，在这⾥配置
maxclients 	#限制同时连接的客户数量,当连接数超过这个值时,redis 将不再接收其他连接请求,返回error
maxmemory 	#设置 redis 能够使⽤的最⼤内存，
1
2
3
4
5
6
7
8

• maxmemory #设置redis能够使用的最大内存

• 备注

  • 防止所有内存超过服务器物理内存，maxmemory限制的时Redis实际使用的内存量，也就是used_memory统计项对应的内存
  • 由于内存碎片率的存在，实际消耗的内存可能会比maxmemory设置的更⼤，实际使⽤时要小心这部分内存溢出
  • 默认⽆限使⽤服务器内存， 为防⽌极端情况下导致系统内存耗尽， 建议所有的Redis进程都要配置maxmemory
  • 在64bit系统下，maxmemory设置为0表示不限制Redis内存使⽤，在32bit系统下，maxmemory不能超过3GB

• 注意：redis在占用的内存超过指定的maxmemory之后，通过maxmemory_policy确定redis是否释放内存以及如何释放内存

12.Redis6的key过期时间删除策略

（1）背景

• redis的key配置了过期时间，这个是怎么被删除的
• redis数据明明过期了，怎末还占用内存
• redis只能用10G，往里面写20G会发生什么

（2）Redis key过期策略

• 定期删除+惰性删除

（3）Redis如何淘汰过期的key

• 定期删除：
  • 隔一段时间，就会随机抽取一些设置了过期时间的key，检查是否过期，如果过期了就删除
  • 定期删除可能会导致很多过期的key到了时间但是并没有被删除，这块就用到惰性删除
• 惰性删除：
  • 当一些用户尝试访问它时，key会被发现并主动的过期，这会惰性删除算法会删除key
  • 放任键过期不管，但是每次从键空间中获取键时，都检查取得的键是否过期，如果过期了就删除
• Redis服务器实际使⽤的是惰性删除和定期删除两种策略：通过配合使⽤这两种删除策略，服务器可以很好地在合理使⽤CPU时间和避免浪费内存空间之间取得平衡。

问题：

如果定期删除漏掉了很多过期key，然后你也没有及时去查，也就没有走惰性删除，这回大量的过期key就会堆积在内存中，导致redis内存消耗尽了，就需要走内存淘汰机制。
注意：设计缓存中间件，可以参考redis的key过期淘汰方式和内存不足淘汰方式
1
2

（4）Redis key内存淘汰策略

redis在占⽤的内存超过指定的maxmemory之后，通过maxmemory_policy确定redis是否释放内存以及如何释放内存
1

• 策略

volatile-lru(least recently used)
最近最少使⽤算法，从设置了过期时间的键中选择空转时间最⻓的键值对清除掉；

volatile-lfu(least frequently used)
最近最不经常使⽤算法，从设置了过期时间的键中选择某段时间之内使⽤频次最⼩的键值对清除掉；

volatile-ttl
从设置了过期时间的键中选择过期时间最早的键值对清除 (删除即将过期的）

volatile-random
从设置了过期时间的键中，随机选择键进⾏清除；

allkeys-lru
最近最少使⽤算法，从所有的键中选择空转时间最⻓的键值对清除；

allkeys-lfu
最近最不经常使⽤算法，从所有的键中选择某段时间之内使⽤频次最少的键值对清除；

allkeys-random
所有的键中，随机选择键进⾏删除；

noeviction
不做任何的清理⼯作，在redis的内存超过限制之后，所有的写⼊操作都会返回错误；但是读操作都能正常的进⾏;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

注意：config配置的时候 下划线_的key需要⽤中横线-

127.0.0.1:6379> config set maxmemory_policy volatile-lru
(error) ERR Unsupported CONFIG parameter:maxmemory_policy

127.0.0.1:6379> config set maxmemory-policy volatile-lru
OK

1
2
3
4
5
6
7
8

8种
（1）不做处理（默认的）
（2）从所有key中随机删除
（3）从所有key中找出访问次数少的，不怎么使用的
（4）从所有key中找出最近空转时间最长的
（5）从设置过期的key中随机删除
（6）从设置过期key中最近空转时间最长的
（7）从过期key中，快要到期的key中删除
（8）从过期key中删除最近不常用的
1
2
3
4
5
6
7
8
9

13.Redis高可用之主从复制

13.1.Redis6主从复制+读写分离

• 架构：一主二从架构搭建

（1）背景

• 单机部署简单，但是可靠性低，其不能很好的利用CPU多核处理
• 生产环境-必须要保证高可用-一般不可能单机部署
• 读写分离时可用性要求不高、性能要求不高、数据规模小的情况

（2）目标

• 读写分离，扩展主节点的读能力，分担主节点读压力
• 容灾恢复，一旦主节点宕机，从节点作为主节点的备份可以随时顶上来

（3）主从复制架构环境搭建准备

• 配置

#创建三个存放redis.conf的配置文件
mkdir -p /data/redis/master/data
mkdir -p /data/redis/slave1/data
mkdir -p /data/redis/slave2/data

#从节点设置只读（默认）
replica-read-only yes

#从节点访问主节点的密码，和requirepass一样，注意主节点也要配下这个，因为主节点宕机后会重新选取主节点
masterauth 123456

#哪个主节点进行复制
replicaof 8.129.113.233 6379
1
2
3
4
5
6
7
8
9
10
11
12
13

• 创建配置文件主节点的redis.conf

bind 0.0.0.0
port 6379
daemonize yes
requirepass "123456"
logfile "/usr/local/redis/log/redis_master.log"
dbfilename "xdclass_master.rdb"
dir "/usr/local/redis/data"
appendonly yes
appendfilename "appendonly_master.aof"
masterauth "123456"
1
2
3
4
5
6
7
8
9
10

• 创建两个从节点配置⽂件redis.conf

bind 0.0.0.0
port 6380
daemonize yes
requirepass "123456"
logfile "/usr/local/redis/log/redis_slave1.log"
dbfilename "xdclass_slave1.rdb"
dir "/usr/local/redis/data"
appendonly yes
appendfilename "appendonly_slave1.aof"
replicaof 8.129.113.233 6379
masterauth "123456"
1
2
3
4
5
6
7
8
9
10
11

bind 0.0.0.0
port 6381
daemonize yes
requirepass "123456"
logfile "/usr/local/redis/log/redis_slave2.log"
dbfilename "xdclass_slave2.rdb"
dir "/usr/local/redis/data"
appendonly yes
appendfilename "appendonly_slave2.aof"
replicaof 8.129.113.233 6379
masterauth "123456"
1
2
3
4
5
6
7
8
9
10
11

（4）启动主节点和从节点

#启动主
./redis-server /data/redis/master/data/redis.conf

#启动从
./redis-server /data/redis/slave1/data/redis.conf

./redis-server /data/redis/slave2/data/redis.conf
1
2
3
4
5
6
7

13.2.主从复制读写分离原理解析

（1）主从复制分两种（主从刚连接时，进行全量同步，全量同步结束后，进行增量同步）

• 全量复制

  • 刚连接时，从节点会向主节点发送一个sync指令，master服务器会开启一个后台进程用于redis的数据生成一个rdb文件
  • 主服务器会缓存所有接受到的来自客户端的写命令，当后台保存进程处理完毕后，会将该rdb文件传递给slave
  • slave服务器会将rdb文件保存在磁盘并通过读取该文件将数据加载到内存
  • 在此之后master服务器会将此期间缓存的命令通过redis传输协议发送给slave服务器
  • 然后slave服务器将这些命令依次作用在自己的服务器上，保证主从数据的一致性

• 增量复制

  • slave初始化后开始正常工作时主服务器发生的写操作同步到从服务器的过程
  • 服务器每执行一个写命令就会向从服务器发送相同的写命令，从服务器接受并且执行

（2）特点

• 主从复制对于主/从redis服务器来说是非阻塞的，所以同步期间都可以正常处理外界请求
• 一个主redis可以包含多个从redis，每个从redis可以接受来其他从redis服务器的连接
• 从节点不会让key过期，而是主节点的key过期删除后，成为del命令传到从节点进行删除

（3）加速复制

• 完全重新同步需要在磁盘上创建⼀个RDB⽂件，然后加载这个⽂件以便为从服务器发送数据
• 在⽐较低速的磁盘，这种操作会给主服务器带来较⼤的压⼒
• 新版支持无磁盘的复制，子进程直接将RDB通过网络发送给从服务器，不使用磁盘作为中间存储
• repl-diskless-sync yes（默认是no）

（4）主从断开连接

• 如果遭遇连接断开，重新连接之后可以从中断处继续进⾏复制，⽽不必重新同步
• 2.8版本后 部分重新同步这个新特性内部使⽤PSYNC命令，旧的实现中使⽤SYNC命令

14.Redis6节点高可用监控之Sentinel

（1）背景

• Redis主从复制，当主机宕机后，需要手动将从服务器切换成主服务器，人工干预费事费力，还会造成一段时间内服务不可用

（2）哨兵模式介绍

• Redis提供了哨兵的命令，是一个独立的进程
• 原理：哨兵通过发送命令给多个节点，等待Redis服务器响应，从⽽监控运⾏的多个Redis实例的运⾏情况
• 当哨兵监测到master宕机，会⾃动将slave切换成master，通过通知其他的从服务器，修改配置⽂件切换主机

（3）Sentinel三大工作任务

• **监控：**Sentinel会不断的检查你的主服务器和从服务器是否运行正常

• **提醒：**当被监控的某个redis服务器出现问题时，Sentinel可以通过API向管理员或者其他应用程序发送通知

• **自动故障迁移：**当一个主服务器不能正常工作时，Sentinel会开始一次自动故障迁移操作，它会将失效主服务器的其中一个从服务器升级为新的主服务器，并让失效主服务器的其他从服务器改为连接新的主服务器，当客户端试图连接失效的主服务器时，集群也会向客户端返回新的服务器地址。

• 注意：一般使用多个哨兵进行监控，各个哨兵之间还会进行监控，形成哨兵模式

（4）多哨兵模式下线名称介绍

• 主观下线（Subjectively Down， 简称 SDOWN）

  • 是单个Sentinel 实例对服务器做出的下线判断，⽐如⽹络问题接收不到通知等
  • ⼀个服务器没有在 down-after-milliseconds 选项所指定的时间内， 对向它发送 PING 命令的 Sentinel返回⼀个有效回复（valid reply）， 那么 Sentinel就会将这个服务器标记为主观下线

• 客观下线（Objectively Down， 简称 ODOWN）

  • 指的是多个 Sentinel 实例在对同⼀个服务器做出SDOWN 判断， 并且通过 SENTINEL is-master-down-by-addr 命令互相交流之后， 得出的服务器下线判断
  • ⼀个 Sentinel 可以通过向另⼀个 Sentinel 发送SENTINEL is-master-down-by-addr 命令来询问对⽅是否认为给定的服务器已下线
  • 客观下线条件只适⽤于主服务器

• 仲裁 qurum

  • Sentinel 在给定的时间范围内， 从其他 Sentinel 那⾥接收到了【⾜够数量】的主服务器下线报告， 那么 Sentinel 就会将主服务器的状态从主观下线改变为客观下线
  • 这个【⾜够数量】就是配置⽂件⾥⾯的值，⼀般是Sentinel个数的⼀半加1，⽐如3个Sentinel则就设置为2
  • down-after-milliseconds 是⼀个哨兵在超过规定时间依旧没有得到响应后，会⾃⼰认为主机不可⽤
  • 当拥有认为主观下线的哨兵达到sentinel monitor所配置的数量时，就会发起⼀次投票，进⾏failover

（5）核心流程

• 每秒ping，超过时间不响应则任务主管下线

• 满足多个，则认为是客观下线

• 投票选择主节点

• 如果没有足够的节点同意master下线，则状态会被移除

（6）环境准备

• 配置三个哨兵，每个哨兵的配置都是一样的
• 启动顺序，先启动主节点在启动从节点，最后启动三个哨兵
• 哨兵端口是【26379】记得开发

#不限制ip
bind 0.0.0.0

# 让sentinel服务后台运⾏
daemonize yes

# 配置监听的主服务器，mymaster代表服务器的名称，⾃定义，172.18.172.109 代表监控的主服务器，6379代表端⼝，2代表只有两个或两个以上的哨兵认为主服务器不可⽤的时候，才会进⾏failover操作。
sentinel monitor mymaster 172.18.172.109 6379 2

# sentinel auth-pass定义服务的密码，mymaster是服务名称，123456是Redis服务器密码
sentinel auth-pass mymaster 123456

#超过5秒master还没有连接上，则认为master已经停⽌
sentinel down-after-milliseconds mymaster 5000

#如果该时间内没完成failover操作，则认为本次failover失败
sentinel failover-timeout mymaster 30000
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

• 在/usr/local/redis/conf创建三个文件sentinel-1.conf、sentinel-2.conf、sentinel-3.conf

port 26379
bind 0.0.0.0
daemonize yes
pidfile "/var/run/redis-sentinel-1.pid"
logfile "/var/log/redis/sentinel_26379.log"
dir "/tmp"
sentinel monitor mymaster 8.129.113.233 6379 2
sentinel down-after-milliseconds mymaster 5000
sentinel auth-pass mymaster 123456
sentinel failover-timeout mymaster 30000
1
2
3
4
5
6
7
8
9
10

port 26380
bind 0.0.0.0
daemonize yes
pidfile "/var/run/redis-sentinel-2.pid"
logfile "/var/log/redis/sentinel_26380.log"
dir "/tmp"
sentinel monitor mymaster 8.129.113.233 6379 2
sentinel down-after-milliseconds mymaster 5000
sentinel auth-pass mymaster 123456
sentinel failover-timeout mymaster 30000
1
2
3
4
5
6
7
8
9
10

port 26381
bind 0.0.0.0
daemonize yes
pidfile "/var/run/redis-sentinel-3.pid"
logfile "/var/log/redis/sentinel_26381.log"
dir "/tmp"
sentinel monitor mymaster 8.129.113.233 6379 2
sentinel down-after-milliseconds mymaster 5000
sentinel auth-pass mymaster 123456
sentinel failover-timeout mymaster 30000
1
2
3
4
5
6
7
8
9
10

（7）启动哨兵集群

./redis-server /usr/local/redis/conf/sentinel-1.conf --sentinel

./redis-server /usr/local/redis/conf/sentinel-2.conf --sentinel

./redis-server /usr/local/redis/conf/sentinel-3.conf --sentinel
1
2
3
4
5

（8）SpringBoot/微服务cloud整合Redis主从+Sentinel哨兵

• 注释掉host和port
• 新增配置

redis:
    #host: 8.140.116.67
    #port: 6379
    sentinel:
      master: mymaster
      nodes: 8.140.116.67:26379,8.140.116.67:26380,8.140.116.67:26381
    password: 123456
    client-type: jedis
1
2
3
4
5
6
7
8

15.Redis6节点高可用之Cluster集群

（1）背景

• Sentinel解决了主从架构故障自动迁移的问题
• 但是Master主节点的写能力和存储能力依旧受限
• 使用Redis的集群Cluster就是为了解决单机Redis容量有限的问题，将数据一定的规划分配到多台机器

（2）什么是集群Cluster

• 是一组项目独立的、通过高速网络互联的计算机，它们构成了⼀个组，并以单⼀系统的模式加以管理

（3）Redis集群模式介绍

• Cluster模式是Redis3.0开始推出
• 采用务无中心结构，每个节点保存数据和整个集群状态,每个节点都和其他所有节点连接
• 官⽅要求：⾄少6个节点才可以保证⾼可⽤，即3主3从；扩展性强、更好做到⾼可⽤
• 各个节点会互相通信，采⽤gossip协议交换节点元数据信息
• 数据分散存储到各个节点上

（4）Cluster数据分片和虚拟哈希槽介绍

• 常见的数据分区算法

  • 哈希取模
    • 对选择的partitioning key计算其哈希值，得到的哈希值就是对应的分区
  • 范围分片
    • 通过确定分区键是否在某个范围内来选择分区
  • 一致性Hash分区

• redis cluster集群没有采用一致性哈希方案，而是采用【数据分片】中的哈希槽来进行数据存储与读取的

（5）什么是Redis的哈希槽slot

• Redis集群预分好16384个槽，当需要在Redis集群中放置一个key-value时，根据CRC16(key)mod16384的值，决定key放在哪个桶中

（6）大体流程

假设主节点的数量为3，将16384个曹魏按照【用户自己的规则】取分配这三个节点，每个节点复制一部分槽位

• 节点1的槽位区间范围为0-5460
• 节点2的槽位区间范围为5461-10922
• 节点3的槽位区间范围为10923-16383

注意：从节点是没有槽位的，只有主节点才有

（7）存储查找

对要存储查找的键进行crc16哈希运算，得到一个值，并取模16384，判断这个值在哪个节点的范围区间，假设crc16(“test_key”)%16384=3000，就是节点1，crc16算法不是简单的hash算法，是一种校验算法

• 使⽤哈希槽的好处就在于可以⽅便的添加或移除节点。
• 当需要增加节点时，只需要把其他节点的某些哈希槽挪到新节点就可以了
• 当需要移除节点时，只需要把移除节点上的哈希槽挪到其他节点就⾏了

（8）Cluster集群环境准备

• 旧版本需要使用ruby语言进行构建，新版5之后直接redis-cli即可

• 6个节点，三主双从，主从节点会自动分配，不是人工指定

• 主节点故障后，从节点会替换主节点

• 节点：

6381、6382
6383、6384
6385、6386
1
2
3

• 配置

bind 0.0.0.0

port 6381

daemonize yes

requirepass "123456"

dbfilename "xdclass_6381.rdb"

logfile "/usr/local/redis/log/redis_6381.log"

dir "/usr/local/redis/data"

appendonly yes

appendfilename "appendonly_6381.aof"

masterauth "123456"

#是否开启集群
cluster-enabled yes

#生成node文件，记录集群节点信息，默认为nodes.conf，防止冲突，改为nodes-6381.conf
cluster-config-file nodes-6381.conf

#节点连接超时时间
cluster-node-timeout 20000

#集群节点的ip，当前节点ip
cluster-announce-ip 172.18.172.109

#集群节点映射端口
cluster-announce-bus-port 16381

#集群节点总线端口，节点之间互相通信，常规端口+1万
cluster-announce-bus-port 16381
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

注意：阿里云开放网络安全组

（9）Cluster集群三主三从搭建实战

• 启动六个节点

./redis-server ../conf/cluster/redis_6381.conf
./redis-server ../conf/cluster/redis_6382.conf
./redis-server ../conf/cluster/redis_6383.conf![在这里插入图片描述](https://img-blog.csdnimg.cn/ccf4aa46ba0b451a933bd2be22c3fab8.jpeg#pic_center)

./redis-server ../conf/cluster/redis_6384.conf
./redis-server ../conf/cluster/redis_6385.conf
./redis-server ../conf/cluster/redis_6386.conf
1
2
3
4
5
6
7

• 现在启动还没开启集群

• 加入集群

--cluster #构建集群中全部节点信息
--cluster-replicas 1 #主从节点的比例，1表示1主1从的方式

./redis-cli -a 123456 --cluster create 8.140.116.67:6381 8.140.116.67:6382 8.140.116.67:6383 8.140.116.67:6384 8.140.116.67:6385 8.140.116.67:6386 --cluster-replicas 1
1
2
3
4

• 检查状态信息（其中一个节点执行即可）

./redis-cli -a 123456 --cluster check 8.140.116.67:6381
1

（10）Cluster集群读写命令

./redis-cli -c -a 123456 -p 6379

#集群信息
cluster info

#节点信息
cluster nodes
1
2
3
4
5
6
7

• 测试集群读写命令set/get

key哈希运算计算槽位置
槽在当前节点的话直接插入/读取。否则自动转向对应的节点
操作都是主节点，从节点只是备份
1
2
3

• 流程解析

  • 主节点

  

  • 从节点

  

启动应用->加入集群->从节点请求复制主节点（主从复制一样）

（11）Cluster集群整合SpringBoot2.X

• 修改配置文件

redis:
    #host: 8.140.116.67
    #port: 6379
#    sentinel:
#      master: mymaster
#      nodes: 8.140.116.67:26379,8.140.116.67:26380,8.140.116.67:26381
    cluster:
      #命名的最多转发次数
      max-redirects: 3
      nodes: 8.140.116.67:6381,8.140.116.67:6382,8.140.116.67:6383,8.140.116.67:6384,8.140.116.67:6385,8.140.116.67:6386
1
2
3
4
5
6
7
8
9
10

注意：一定要在同一个网段

• 当kill掉主节点

• 从节点会尝试连接，知道连不上，把自己置成主节点

• 当原本的主节点挂掉之后，重新恢复，成为从节点，复制主节点的数据

16.新版Redis6核心特性

16.1.Redis6新特性-多线程

（1）支持多线程

• redis6多线程只是用来处理网络数据的读写和协议解析上，底层数据操作还是单线程
• 执行命令依旧是单线程，之所以这么设计是不想因为多线程而变的复杂，需要去控制key，lua，事务等等并发问题

（2）默认不开启

io-threads-do-reads yes
io-threads 线程数
1
2

4核的机器建议设置为 2 或 3 个线程

8核的建议设置为 4或6个线程

注意：开启多线程后，是否会存在线程并发安全问题？

不会有安全问题，Redis的多线程部分只是用来吃力网络数据的读写和协议解析，执行命令仍然是单线程顺序执行
1

16.2.Redis6新特性-acl权限控制

（1）引入ACL（Access Control List）

之前的redis没有⽤户的概念，redis6引⼊了acl
可以给每个⽤户分配不同的权限来控制权限
通过限制对命令和密钥的访问来提⾼安全性，以使不受信任的客户端⽆法访问
提⾼操作安全性，以防⽌由于软件错误或⼈为错误⽽导致进程或⼈员访问 Redis，从⽽损坏数据或配置
1
2
3
4

（2）常用命令

acl list	 #当前启用的ACL规则
acl cat		 #支持的权限分类列表
acl cat hash #返回指定类别中的命令
acl setuser	 #创建和修改用户命令
acl deluser	 #删除用户命令
1
2
3
4
5

+<command>:将命令添加到⽤户可以调⽤的命令列表中，如+@hash

-<command>:将命令从⽤户可以调⽤的命令列表中移除

#切换默认用户
auth default 123456

#例⼦ 密码 123 ，全部key，全部权限
acl setuser jack on >123 ~* +@all

#例⼦ 密码 123 ，全部key，get权限
acl setuser jack on >123 ~* +get
1
2
3
4
5
6
7
8
9
10
11
12

16.3.Redis6新特性-客户端缓存

client side caching客户端缓存