文章目录

初始Nosql
- 一、互联网大潮下,NoSQL 的机遇
- 二、什么是 NoSQL
Redis的安装配置
- 一、Redis介绍
- 2、Window 上安装 Redis
- 三、Linux 上安装 Redis
Redis客户端
- 一、客户端简介
- 二、Redis 命令行客户端
- 3.图形界面客户端
- 4.Redis 编程客户端
- 5.Redis 客户端连接超时解决
Redis基本操作命令
- 一、基本操作命令
- 2.Redis 的 Key 的操作命令
redis的数据类型
- 1.数据类型概述
- 2.Redis 数据类型操作命令
redis的发布订阅
- 1.发布和订阅
redis的事务
- 1.Redis事务
springboot整合SpringDataRedis
- 1. SpringDataRedis简介
- - 1.1.Jedis（过时）
  - 1.2.Spring Data Redis（推荐）
- 2.Spring Data Redis入门小Demo
- - 2.1.准备工作
  - 2.2.实体类
- 3. 字符串类型操作
- 4. SET类型操作
- 5. List类型操作
- 6. Hash类型操作
- 7. Zset类型操作
redis的主从复制
- 1.主从复制概念（master/slave）
- 2.主从复制（master/slave）实现
- 3.主从复制（master/slave）容灾处理
- 4.总结
redis的哨兵模式
- 1.高可用Sentinel哨兵介绍
- 2.高可用Sentinel哨兵环境搭建
Springboot连接哨兵模式
- 1.application.yml
- 2.配置文件
redis-cluster集群搭建
- 1.Redis-cluster架构图
- 2.Redis-cluster投票:容错
- 3.Redis-cluster集群搭建环境准备
- 4.Redis-cluster集群搭建过程
- 5.Redis-cluster集群连接
- 6.查看集群信息
- 7.查看集群信息
Springboot2.X集成redis集群(Lettuce)连接
- 1.前言
- 2.导入依赖
- 3. 配置application.yml
- 4.新建下面的两个配置类
- 5.执行测试
redis持久化详解
- 1.持久化
- - 1.1. 持久化简介
- 2. RDB持久化
- 3.AOF持久化
- 4.AOF重写
- 5.AOF持久化默认参数
- 6.持久化方式总结与抉择
- 7.如何选择使用哪种持久化方式？
过期删除策略和内存淘汰策略
- 1.引言
- 2.设置Redis键过期时间
- - 2.1.移除键的过期时间
  - 2.2.返回键的剩余生存时间
- 3.Redis过期时间的判定
- 4.过期删除策略
- 5.Redis过期删除策略
- 6.内存淘汰策略
- 7.总结
缓存穿透、缓存击穿、缓存雪崩
- 1.缓存穿透
- - 1.1.概念
  - 1.2.解决办法
- 2、缓存击穿
- - 2.1.概念
  - 2.2.解决办法
- 3、缓存雪崩
- - 3.1.概念
  - 3.2.解决办法
Redis布隆过滤器
- 1.布隆过滤器使用场景
- 2.布隆过滤器简介
- 3.Redis实现布隆过滤器
- 4.Redisson
数据库和缓存双写一致性
1.常见方案
2.先写缓存，再写数据库
3.先写数据库，再写缓存
- 3.1.写缓存失败了
- 3.2.高并发下的问题
- 3.2 浪费系统资源
4.先删缓存，再写数据库
- 4.1 高并发下的问题
- 4.2 缓存延迟双删
5. 先写数据库，再删缓存
6.删缓存失败怎么办？
7.定时任务
8.消息队列
9.binlog
Redis综述篇：与面试官彻夜长谈Redis缓存、持久化、淘汰机制、哨兵、集群底层原理！
- 一、Redis基本概念
- 二、Redis基本数据类型与常用指令
- 三、Redis缓存及一致性、雪崩、击穿与穿透问题
- 四、Redis八种淘汰策略与三种删除策略
- - 4.1. 八种键淘汰（过期）策略
  - 4.2. 三种键删除策略
- 五、Redis三种持久化机制
- 六、`Redis`的事务机制
- 七、`Redis`内存模型及内存划分
- 八、Redis虚拟内存
- 九、`Redis`客户端通信RESP协议
- 十、Redis高可用机制：主从复制、哨兵、代理式/分片式集群
- 十一、`Redis`版本新特性

初始Nosql

一、互联网大潮下,NoSQL 的机遇

2000 年后互联网逐渐发展起来，而且十分的迅猛。早期的关系型数据库已经不能适应互联网的快速变化。论坛、博客、sns、微博逐渐引领 web 领域的潮流。facebook , twitter等这样的网站，每天都会产生海量的数据，几亿，十几亿的数据量。淘宝，腾讯都拥有庞大的用户群。每天的访问量都是用亿来计算。用户账号也都以亿计算的。传统的 RDBMS 是难
以处理这样的数据的。数据量十分巨大，单个表存储的数据量是有上限的。不同的数据库上限是不同的。但一般达到 2-3 百万条数据后，查询和更新都会变慢，数据量的增加，处理速度回几何的增加。

在 90 年代初期，2000 年左右。那时的应用都比较小，用户量也比较小。计算机还没有普及。那时计算动辄上万元。网络速度也是出奇的慢。用户都是看看静态网页，看看新闻。和网站的交互很少。什么论坛都没有几个人。单机一个数据库足以满足用户的使用。

此时我们的系统这样的：

单机时代:

此方式的缺点：
1.一个机器的数据存储量是有大小限制的。
2.读写数据都在一个数据库实例完成，大数量是不能承受的。

随着用户的访问量逐渐的上升。数据库都不同的出现了性能问题。web 应用不仅考虑功能，还要追求性能。技术大牛们纷纷使用缓存技术缓解数据库的压力。开始的缓存数据是存在文件中。大量的文件又导致访问磁盘的 io 问题。此时 Memcached 横空出世，一个非常时尚，性能十分强悍。在内存中缓存数据。Memcached 作为一个独立的缓存服务器，为多个web 服务器提供了一个共享的高性能缓存。

现在的我们的系统演变为：

Memcached 只能缓解数据库的读取压力，读写数据集中在一个数据库，数据库不堪重负。技术大牛们开始使用读写分离技术。以提高数据库的读写能力。Mysql 的 master-slave 模式称为网站标配。

我们的系统演变为：缓存加主从数据库

当 Memcached 和 MySQL 的读写分离渐渐也不能满足日益增强的性能需求。在 MySQL的读写分离基础上，实现分表分库。数据库的集群。数据库的性能扩展是有极限的。关系数据库很强大，但是它并不能很好的应付所有的应用场景。MySQL 的扩展性差（需要复杂的技术来实现），大数据下 IO 压力大，表结构更改困难，正是当前使用 MySQL 以及关系型数据库的开发人员面临的问题。

二、什么是 NoSQL

为了解决高并发、高可扩展、高可用、大数据存储问题而产生的数据库解决方案，就是NoSql数据库。

NoSQL = Not Only SQL(不仅仅是 SQL) ，也解释为 non-relational(非关系型数据库)。在NoSQL 数据库中数据之间是无联系的，无关系的。数据的结构是松散的，可变的。它可以作为关系型数据库的良好补充。

常见的关系型数据库： relational db system 数据库存储的数据是以二维表格的方式展示的~、

Oracle、MySql、SQLServer…

常见的Nosql数据库? non-relational(非关系型数据库) 数据不是以二维表格方式存储的~

redis、MongoDB、Solr、ES

Redis的安装配置

一、Redis介绍

1.1 Redis历史发展

2008年，意大利的一家创业公司Merzia推出了一款基于MySQL的网站实时统计系统LLOOGG，然而没过多久该公司的创始人 Salvatore Sanfilippo便对MySQL的性能感到失望，于是他决定亲自为LLOOGG量身定做一个数据库，并于2009年开发完成，这个数据库就是Redis。不过Salvatore Sanfilippo并不满足只将Redis用于LLOOGG这一款产品，而是希望更多的人使用它，于是在同一年Salvatore Sanfilippo将Redis开源发布，并开始和Redis的另一名主要的代码贡献者Pieter Noordhuis一起继续着Redis的开发，直到今天。

Salvatore Sanfilippo自己也没有想到，短短的几年时间，Redis就拥有了庞大的用户群体。Hacker News在2012年发布了一份数据库的使用情况调查，结果显示有近12%的公司在使用Redis。国内如新浪微博、街旁网、知乎网，国外如GitHub、Stack Overflow、Flickr等都是Redis的用户。

VMware公司从2010年开始赞助Redis的开发， Salvatore Sanfilippo和Pieter Noordhuis也分别在3月和5月加入VMware，全职开发Redis。

Redis的代码托管在GitHub上https://github.com/antirez/redis，开发十分活跃，代码量只有3万多行。

1.2.Redis的简介

Remote Dictionary Server(Redis) 是一个开源的使用 ANSI C 语言编写、支持网络、可基于内存亦可持久化的 Key-Value 数据库. Key 字符类型，其值（value）可以是 字符串(String), 哈希(Map), 列表(list), 什么是 NoSQL 和 **有序集合(sorted sets)**等类型，每种数据类型有自己的专属命令。所以它通常也被称为数据结构服务器。

Redis 的作者是 Salvatore Sanfilippo，来自意大利的西西里岛，现在居住在卡塔尼亚。目前供职于 Pivotal 公司（Pivotal 是 Spring 框架的开发团队），Salvatore Sanfilippo 被称为 Redis之父。

官网：https://redis.io/
github: https://github.com/antirez/redis
中文：http://www.redis.cn/
http://www.runoob.com/redis/redis-tutorial.html

Redis 与其他 key – value 缓存产品有以下三个特点

i）Redis支持数据的持久化，可以将内存中的数据保持在磁盘中，重启的时候可以再次加载进行使用
ii）Redis不仅仅支持简单的key-value类型的数据，同时还提供list，set，zset，hash等数据结构的存储
iii）Redis支持数据的备份，即master-slave模式的数据备份(集群直连型集群、哨兵集群)

1.3.Redis优势

性能极高 – Redis能读的速度是110000次/s,写的速度是81000次/s 。
丰富的数据类型 – Redis支持二进制案例的 Strings, Lists, Hashes, Sets 及 Ordered Sets 数据类型操作。
原子 – Redis的所有操作都是原子性的，同时Redis还支持对几个操作全并后的原子性执行。
丰富的特性 – Redis还支持 publish/subscribe, 通知, key 过期等等特性

1.4.Redis的应用场景

缓存 Cache（数据查询、短连接、新闻内容、商品内容等等）（最多使用）
分布式集群架构中的session分离。解决：session一致性
聊天室的在线好友列表。
任务队列。（秒杀、抢购、12306等等）
应用排行榜。SortedSet
网站访问统计。
数据过期处理（可以精确到毫秒）
分布式锁(Redission)

2、Window 上安装 Redis

Windows 版本的 Redis 是 Microsoft 的开源部门提供的 Redis. 这个版本的 Redis 适合开发
人员学习使用，生产环境中使用 Linux 系统上的 Redis

2.1.Redis下载

官网地址：http://redis.io/
windows 版本：https://github.com/MSOpenTech/redis/releases

2.2.Redis安装

下载的 Redis-x64-3.2.100.zip 解压后，放到某个目录（例如 d:\tools\），即可使用。
目录结构：

2.3.Redis启动

A、Windows7 系统双击 redis-server.exe 启动 Redis

B、 Windows 10 系统
有的机器双击 redis-server.exe 执行失败，找不到配置文件，可以采用以下执行方式：
在命令行（cmd）中按如下方式执行：
D:\tools\Redis-x64-3.2.100>redis-server.exe redis.conf
如图：

2.4.Redis关闭

按 ctrl+c 退出 Redis 服务程序。

三、Linux 上安装 Redis

3.1 Linux版Redis下载

官网地址：http://redis.io/
下载地址：http://download.redis.io/releases/redis-3.0.0.tar.gz
在Linux中使用wget下载到linux或者下载到window在上传到linux

wget http://download.redis.io/releases/redis-3.0.0.tar.gz

3.2 Linux版Redis安装

Redis是C语言开发，建议在linux上运行，本教程使用Centos6.5作为安装环境。
第一步：在VMware中安装CentOS（参考Linux教程中的安装虚拟机）
第二步：在Linux下安装gcc环境（该步骤可以省略，CentOS中默认自带C环境）

yum install gcc-c++

可以通过rpm -qa | grep gcc 来查询是否已经安装了gcc

第三步：将下载的Redis源码包上传到Linux服务器中【如果是linux直接下载的，就省略这个步骤】
第四步：解压缩Redis源码包

# tar -zxf redis-3.0.0.tar.gz 【直接解压到当前文件夹】

第五步：编译redis源码

cd redis-3.0.0
make

第六步：安装redis

make install PREFIX=/usr/local/redis

/usr/local/redis 路径安装路径

make install PREFIX=/soft/redis/redis

3.3 Linux版Redis启动

3.3.1 Linux版Redis前端启动

直接运行bin/redis-server将以前端模式启动。【bin目录是在/usr/local/redis/bin】

# ./redis-server

ssh命令窗口关闭则redis-server程序结束，不推荐使用此方法

3.3.2 Linux版Redis前端关闭

前端启动的关闭：ctrl+c

3.3.3 Linux版Redis后端启动

第一步：将redis源码包中的redis.conf配置文件复制到/usr/local/redis/bin/下

# cd /root/redis-3.0.0
# cp redis.conf /usr/local/redis/bin/

第二步：修改redis.conf，将daemonize由no改为yes

# vi redis.conf

第三步：执行命令

# ./redis-server redis.conf

3.3.4 Linux版Redis后端关闭

方式1：
① 使用 redis 客户端关闭，向服务器发出关闭命令切换到 redis-3.2.9/src/ 目录，执行 ./redis-cli shutdown 推荐使用这种方式， redis 先完成数据操作，然后再关闭。

方式2：
② kill pid 或者 kill -9 pid
这种不会考虑当前应用是否有数据正在执行操作，直接就关闭应用。先使用 ps -ef | grep redis 查出进程号，在使用 kill pid

Redis客户端

一、客户端简介

Redis 客户端是一个程序，通过网络连接到 Redis 服务器，在客户端软件中使用 Redis可以识别的命令，向 Redis 服务器发送命令，告诉 Redis 想要做什么。Redis 把处理结果显示在客户端界面上。通过 Redis 客户端和 Redis 服务器交互。Redis 客户端发送命令，同时显示 Redis 服务器的处理结果。

二、Redis 命令行客户端

redis-cli （Redis Command Line Interface）是 Redis 自带的基于命令行的 Redis 客户端，用于与服务端交互，我们可以使用该客户端来执行 redis 的各种命令。
两种常用的连接方式：

A、直接连接 redis (默认 ip127.0.0.1，端口 6379)：./redis-cli
在 redis 安装目录\src, 执行 ./redis-cli
此命令是连接本机 127.0.0.1 ，端口 6379 的 redis

B、指定 IP 和端口连接 redis：./redis-cli -h 127.0.0.1 -p 6379
-h redis 主机 IP（可以指定任意的 redis 服务器）
-p 端口号（不同的端口表示不同的 redis 应用）
在 redis 安装目录\src, 执行 ./redis-cli -h 127.0.0.1 -p 6379 -a 密码

例 1：

3.图形界面客户端

Redis Desktop Manager：C++ 编写，响应迅速，性能好。
官网地址： https://redisdesktop.com/
github: https://github.com/uglide/RedisDesktopManager
使用文档：http://docs.redisdesktop.com/en/latest/

点击“DOWNLOAD”

A、安装客户端软件
在 Windows 系统使用此工具，连接 Linux 上或 Windows 上的 Redis , 双击此 exe 文件执
行安装

B、安装后启动界面

C、配置 Redis Desktop Manamager(RDM)，连接 Redis
在 RDM 的主窗口，点击左下的“Connect to Redis Server”

连接成功后：

4.Redis 编程客户端

A、Jedis (类似 JDBC)
redis 的 Java 编程客户端，Redis 官方首选推荐使用 Jedis，jedis 是一个很小但很健全的
redis 的 java 客户端。通过 Jedis 可以像使用 Redis 命令行一样使用 Redis。
jedis 完全兼容 redis 2.8.x and 3.x.x
Jedis 源码：https://github.com/xetorthio/jedis
api 文档：http://xetorthio.github.io/jedis/

B、Spring Data Redis (类似 JDBC)
Spring-data-redis是spring大家族的一部分，提供了在srping应用中通过简单的配置访问redis服务，对reids底层开发包(Jedis, JRedis, and RJC)进行了高度封装，RedisTemplate提供了redis各种操作、异常处理及序列化，支持发布订阅，并对spring 3.1 cache进行了实现。

C、 Lettuce:
Lettuce 是一个可伸缩线程安全的 Redis 客户端。多个线程可以共享同一个
RedisConnection。它能够高效地管理多个连接。
Lettuce 源码：https://github.com/lettuce-io/lettuce-core

D、 redis 的其他编程语言客户端:
C 、C++ 、C# 、Erlang、Lua 、Objective-C 、Perl 、PHP 、Python 、Ruby 、Scala 、
Go 等 40 多种语言都有连接 redis 的编程客户端

5.Redis 客户端连接超时解决

远程连接redis服务，需要关闭或者修改防火墙配置。

1.将修改后的端口添加到防火墙中.

/sbin/iptables -I INPUT -p tcp --dport 6379 -j ACCEPT
/etc/rc.d/init.d/iptables save

第一步：编辑iptables /etc/sysconfig/iptables

#vim

在命令模式下，选定要复制的那一行的末尾，然后点击键盘yyp，就完成复制，然后修改。

第二步：重启防火墙

# service iptables restart

注意：
默认一共是16个数据库，每个数据库之间是相互隔离。数据库的数量是在redis.conf中配置的

切换数据库使用命令：select 数据库编号
例如：select 1【相当于mysql 的use databasename】

Redis基本操作命令

一、基本操作命令

手册地址：
redis 英文版命令大全：https://redis.io/commands
redis 中文版命令大全：http://redisdoc.com/

redis命令参考: http://doc.redisfans.com/

redis 默认为 16 个库 (在 redis.conf 文件可配置，该文件很重要，后续很多操作都是这个配
置文件) redis 默认自动使用 0 号库。

A、沟通命令，查看状态
redis >ping 返回 PONG
解释：输入 ping，redis 给我们返回 PONG，表示 redis 服务运行正常

B、查看 redis 服务器的统计信息：info
语法：info [section]
作用：以一种易于解释且易于阅读的格式，返回关于 Redis 服务器的各种信息和统计数值。
section 用来返回指定部分的统计信息。 section 的值：server , clients ，
memory 等等。不加 section 返回全部统计信息
返回值：指定 section 的统计信息或全部信息

例 1：统计 server 的信息

例 2：统计全部信息

C、查看当前数据库中 key 的数目：dbsize
语法：dbsize
作用：返回当前数据库的 key 的数量。
返回值：数字，key 的数量

例：先查索引 5 的 key 个数，再查 0 库的 key 个数

D、redis 默认使用 16 个库
Redis 默认使用 16 个库，从 0 到 15。对数据库个数的修改，在 redis.conf 文件中 databases 16

E、切换库命令：select db
使用其他数据库，命令是 select index
例 1： select 5

F、config get parameter 获得 redis 的配置值
语法：config get parameter
作用：获取运行中 Redis 服务器的配置参数，获取全部配置可以使用 * 。参数信息来自
redis.conf 文件的内容。

例 1：获取数据库个数 config get databases

例 2：获取端口号 config get port

例 3：获取所有配置参数 config get *

F、删除所有库的数据：flushall

G、删除当前库的数据：flushdb

H、redis 自带的客户端退出当前 redis 连接: exit 或 quit

2.Redis 的 Key 的操作命令

A、keys
语法：keys pattern
作用：查找所有符合模式 pattern 的 key. pattern 可以使用通配符。
通配符：
● * ：表示 0-多个字符，例如：keys * 查询所有的 key。
● ？：表示单个字符，例如：wo?d , 匹配 word , wood
● [] ：表示选择[]内的字符，例如 wo[or]d, 匹配 word, wood, 不匹配 wold

例 1：显示所有的 key

例 2：使用 * 表示 0 或多个字符

例 3：使用？表示单个字符

B、 exists
语法：exists key [key…]
作用：判断 key 是否存在
返回值：整数，存在 key 返回 1，其他返回 0. 使用多个 key，返回存在的 key 的数量。

例 1：检查指定 key 是否存在

例 2：检查多个 key

C、 expire
语法：expire key seconds
作用：设置 key 的生存时间，超过时间，key 自动删除。单位是秒。
返回值：设置成功返回数字 1，其他情况是 0 。

例 1：设置红灯的倒计时是 5 秒

D、ttl
语法：ttl key
作用：以秒为单位，返回 key 的剩余生存时间（ttl: time to live）
返回值：
● -1 ：没有设置 key 的生存时间， key 永不过期。
● -2 ：key 不存在
● 数字：key 的剩余时间，秒为单位

例 1：设置 redlight 的过期时间是 10，查看剩余时间

E、 type
语法：type key
作用：查看 key 所存储值的数据类型
返回值：字符串表示的数据类型
● none (key 不存在)
● string (字符串)
● list (列表)
● set (集合)
● zset (有序集)
● hash (哈希表)

例 1：查看存储字符串的 key ：wood

例 2：查看不存在的 key

F、 del
语法：del key [key…]
作用：删除存在的 key ，不存在的 key 忽略。
返回值：数字，删除的 key 的数量。

例 1：删除指定的 key

E、 rename
语法：rename key
作用：修改key的名字
返回值：修改之后的状态

redis 127.0.0.1:6379[1]> keys *
1) "age"
redis 127.0.0.1:6379[1]> rename age age_new
OK
redis 127.0.0.1:6379[1]> keys *
1) "age_new"
redis 127.0.0.1:6379[1]>

redis的数据类型

1.数据类型概述

Redis中存储数据是通过key-value存储的，对于value的类型有以下几种：

字符串类型 string
字符串类型是 Redis 中最基本的数据类型，它能存储任何形式的字符串，包括二进制
数据，序列化后的数据，JSON 化的对象甚至是一张图片。最大 512M。
哈希类型 hash
Redis hash 是一个 string 类型的 field 和 value 的映射表，hash 特别适合用于存储对象。
列表类型 list
Redis 列表是简单的字符串列表，按照插入顺序排序。你可以添加一个元素到列表的头
部（左边）或者尾部（右边）
集合类型 set
Redis 的 Set 是 string 类型的无序集合，集合成员是唯一的，即集合中不能出现重复的数
据.
有序集合类型 zset （sorted set)
Redis 有序集合 zset 和集合 set 一样也是 string 类型元素的集合，且不允许重复的成员。
不同的是 zset 的每个元素都会关联一个分数（分数可以重复），redis 通过分数来为集合中的
成员进行从小到大的排序。

2.Redis 数据类型操作命令

2.1. 字符串类型（string）

字符串类型是 Redis 中最基本的数据类型，它能存储任何形式的字符串，包括二进制数
据，序列化后的数据，JSON 化的对象甚至是一张图片。

2.1.1 基本操作命令

A、set
将字符串值 value 设置到 key 中
语法：set key value

查看已经插入的 key

向已经存在的 key 设置新的 value，会覆盖原来的值

B、 get
获取 key 中设置的字符串值
语法： get key
例如：获取 username 这个 key 对应的 value

C、 incr
将 key 中储存的数字值加 1，如果 key 不存在，则 key 的值先被初始化为 0 再执行
incr 操作（只能对数字类型的数据操作）
语法：incr key
例 1：操作key,值增加 1

例 2：对非数字的值操作是不行的

D、 decr
将 key 中储存的数字值减1，如果 key 不存在，则么 key 的值先被初始化为 0 再执
行 decr 操作（只能对数字类型的数据操作）
语法：decr key
例1：不存在的key，初值为0，再减 1 。

例2：对存在的数字值的 key ，减 1 。
先执行 incr index ,增加到 3

incr ，decr 在实现关注人数上，文章的点击数上。

E、 append
语法：append key value
说明：如果 key 存在，则将 value 追加到 key 原来旧值的末尾
如果 key 不存在，则将 key 设置值为 value
返回值：追加字符串之后的总长度
例 1：追加内容到存在的 key

例 2：追加到不存在的 key，同 set key value

2.1.2 其他操作命令

A、 strlen
语法：strlen key
说明：返回 key 所储存的字符串值的长度
返回值：
①：如果key存在，返回字符串值的长度
②：key不存在，返回0
例 1：计算存在 key 的字符串长度

设置中文 set k4 中文长度，按字符个数计算

例 2：计算不存在的 key

B、 getrange
语法：getrange key start end
作用：获取 key 中字符串值从 start 开始到 end 结束的子字符串,包括 start 和 end, 负数
表示从字符串的末尾开始， -1 表示最后一个字符
返回值：截取的子字符串。
使用的字符串 key: school, value: bejingzhiyejishuxueyuan

例 1: 截取从 2 到 5 的字符

位置是从0开始计算

例 2：从字符串尾部截取，start ,end 是负数，最后一位是 -1

例 3：超出字符串范围的截取，获取合理的子串

C、 setrange
语法：setrange key offset value
说明：用 value 覆盖（替换）key 的存储的值从 offset 开始,不存在的 key 做空白字符串。
返回值：修改后的字符串的长度
例 1：替换给定的字符串

例 2：设置不存在的 key

D、mset
语法：mset key value [key value…]
说明：同时设置一个或多个 key-value 对
返回值： OK
例 1：一次设置多个 key， value

E、 mget
语法：mget key [key …]
作用：获取所有(一个或多个)给定 key 的值
返回值：包含所有 key 的列表

例 1：返回多个 key 的存储值

例 2：返回不存在的 key

2.1.3 应用-自增主键

商品编号、订单号采用string的递增数字特性生成。

定义商品编号key：items:id

192.168.101.3:7003> INCR items:id
(integer) 2
192.168.101.3:7003> INCR items:id
(integer) 3

2.2 哈希类型 hash

2.2.1 使用string的问题

假设有User对象以JSON序列化的形式存储到Redis中，User对象有id，username、password、age、name等属性，存储的过程如下：
保存、更新： User对象 →json(string) →redis

如果在业务上只是更新age属性，其他的属性并不做更新我应该怎么做呢？如果仍然采用上边的方法在传输、处理时会造成资源浪费，下边讲的hash可以很好的解决这个问题。
User “{“username”:”gyf”,”age”:”80”}”

2.2.2 hash介绍

hash叫散列类型，它提供了字段和字段值的映射。字段值只能是字符串类型，不支持散列类型、集合类型等其它类型。如下：

2.2.3 基本操作命令

A、hset
语法：hset hash 表的 key field value
作用：将哈希表 key 中的域 field 的值设为 value ，如果 key 不存在，则新建 hash 表，执
行赋值，如果有 field ,则覆盖值。
返回值：
①如果 field 是 hash 表中新 field，且设置值成功，返回 1
②如果 field 已经存在，旧值覆盖新值，返回 0

例 1：新的 field

例 2：覆盖旧的的 field

B、 hget
语法：hget key field
作用：获取哈希表 key 中给定域 field 的值
返回值：field 域的值，如果 key 不存在或者 field 不存在返回 nil
例 1：获取存在 key 值的某个域的值

例 2：获取不存在的 field

C、 hmset
语法：hmset key field value [field value…]
说明：同时将多个 field-value (域-值)设置到哈希表 key 中，此命令会覆盖已经存在的 field，
hash 表 key 不存在，创建空的 hash 表，执行 hmset.
返回值：设置成功返回 ok，如果失败返回一个错误

例 1：同时设置多个 field-value

使用 redis-desktop-manager 工具查看 hash 表 website 的数据结构

例 2：key 类型不是 hash,产生错误

D、hmget
语法：hmget key field [field…]
作用:获取哈希表 key 中一个或多个给定域的值
返回值：返回和 field 顺序对应的值，如果 field 不存在，返回 nil

例 1：获取多个 field 的值

E、 hgetall
语法：hgetall key
作用：获取哈希表 key 中所有的域和值
返回值：以列表形式返回 hash 中域和域的值，key 不存在，返回空 hash

例 1：返回 key 对应的所有域和值

例 2：不存在的 key，返回空列表

F、 hdel
语法：hdel key field [field…]
作用：删除哈希表 key 中的一个或多个指定域 field，不存在 field 直接忽略
返回值：成功删除的 field 的数量

例 1：删除指定的 field

2.2.4 其他操作命令

A、hlen
语法：hlen key
作用：获取哈希表 key 中域 field 的个数
返回值：数值，field 的个数。key 不存在返回 0.

例 1：获取指定 key 域的个数

例 2：不存在的 key，返回 0

B、 hkeys
语法：hkeys key
作用：查看哈希表 key 中的所有 field 域
返回值：包含所有 field 的列表，key 不存在返回空列表

例 1：查看 website 所有的域名称

C、 hvals
语法：hvals key
作用：返回哈希表中所有域的值
返回值：包含哈希表所有域值的列表，key 不存在返回空列表

例 1：显示 website 哈希表所有域的值

D、hexists
语法：hexists key field
作用：查看哈希表 key 中，给定域 field 是否存在
返回值：如果 field 存在，返回 1，其他返回 0

例 1：查看存在 key 中 field 域是否存在

2.2.5 应用-存储商品信息

商品字段
【商品id、商品名称、商品描述、商品库存、商品好评】

定义商品信息的key
商品1001的信息在 Redis中的key为：[items:1001]

存储商品信息

192.168.101.3:7003> HMSET items:1001 id 3 name apple price 999.9
OK

获取商品信息

  192.168.101.3:7003> HGET items:1001 id
  "3"
  192.168.101.3:7003> HGETALL items:1001
  1) "id"
  2) "3"
  3) "name"
  4) "apple"
  5) "price"
  6) "999.9"

2.3 列表类型（list）

2.3.1 ArrayList与LinkedList的区别

ArrayList使用数组方式存储数据，所以根据索引查询数据速度快，而新增或者删除元素时需要设计到位移操作，所以比较慢。

LinkedList使用双向链表方式存储数据，每个元素都记录前后元素的指针，所以插入、删除数据时只是更改前后元素的指针指向即可，速度非常快。然后通过下标查询元素时需要从头开始索引，所以比较慢，但是如果查询前几个元素或后几个元素速度比较快。

2.3.2 list类型介绍

列表类型（list）可以存储一个有序的字符串列表，常用的操作是向列表两端添加元素，或者获得列表的某一个片段。

列表类型内部是使用双向链表（double linked list）实现的，所以向列表两端添加元素的时间复杂度为0(1)，获取越接近两端的元素速度就越快。这意味着即使是一个有几千万个元素的列表，获取头部或尾部的10条记录也是极快的。

2.3.3 基本命令

A、lpush
语法：lpush key value [value…]
作用：将一个或多个值 value 插入到列表 key 的表头（最左边），从左边开始加入值，从左
到右的顺序依次插入到表头
返回值：数字，新列表的长度

例 1：将 a,b,c 插入到 mylist 列表类型

在 redis-desktop-manager 显示

插入图示：

例 2：插入重复值到 list 列表类型

在 redis-desktop-manager 显示

B、 rpush
语法：rpush key value [value…]
作用：将一个或多个值 value 插入到列表 key 的表尾（最右边），各个 value 值按从左到右
的顺序依次插入到表尾
返回值：数字，新列表的长度

例 1：插入多个值到列表

在 redis-desktop-manager 显示：

C、 lrange
语法：lrange key start stop
作用：获取列表 key 中指定区间内的元素，0 表示列表的第一个元素，以 1 表示列表的第
二个元素；start , stop 是列表的下标值，也可以负数的下标， -1 表示列表的最后一
个元素， -2 表示列表的倒数第二个元素，以此类推。 start ，stop 超出列表的范围
不会出现错误。
返回值：指定区间的列表

例 1：返回列表的全部内容

例 2：显示列表中第 2 个元素，下标从 0 开始

D、lpop
语法：lpop key
作用：移除并返回列表 key 头部第一个元素，即列表左侧的第一个下标值。相当于栈（stack）
返回值：列表左侧第一个下标值；列表 key 不存在，返回 nil

例 1：取出列表的左侧第一个下标值

E、 rpop
语法：rpop key
作用：移除并返回 key 的尾部元素
返回值：列表的尾部元素；key 不存在返回 nil

例 1：移除列表的尾部元素

F、 lindex
语法：lindex key index
作用：获取列表 key 中下标为指定 index 的元素，列表元素不删除，只是查询。0 表示列
表的第一个元素，以 1 表示列表的第二个元素；start , stop 是列表的下标值，也可以负数的下标， -1 表示列表的最后一个元素， -2 表示列表的倒数第二个元素，以此类推。
返回值：指定下标的元素；index 不在列表范围，返回 nil

例 1：返回下标是 1 的元素

例 2：不存在的下标

G、llen
语法：llen key
作用：获取列表 key 的长度
返回值：数值，列表的长度； key 不存在返回 0

例 1：显示存在 key 的列表元素的个数

2.3.4 其他命令

A、lrem
语法：lrem key count value
作用：根据参数 count 的值，移除列表中与参数 value 相等的元素， count >0 ，从列表的
左侧向右开始移除； count < 0 从列表的尾部开始移除；count = 0 移除表中所有
与 value 相等的值。
返回值：数值，移除的元素个数

例 1：删除 2 个相同的列表元素

例 2：删除列表中所有的指定元素，删除所有的 java

B、 ltrim
语法：ltrim key start stop
作用：删除指定区域外的元素，比如 LTRIM list 0 2 ，表示只保留列表 list 的前三个元素，
其余元素全部删除。0 表示列表的第一个元素，以 1 表示列表的第二个元素；start ,
stop 是列表的下标值，也可以负数的下标， -1 表示列表的最后一个元素， -2 表示
列表的倒数第二个元素，以此类推。
返回值：执行成功返回 ok

例 1：保留列表的前三个元素

C、 lset
语法：lset key index value
作用：将列表 key 下标为 index 的元素的值设置为 value。
返回值：设置成功返回 ok ; key 不存在或者 index 超出范围返回错误信息

例 1：设置下标 2 的 value 为“c”。

D、linsert
语法：linsert key BEFORE|ALFTER pivot value
作用：将值 value 插入到列表 key 当中位于值 pivot 之前或之后的位置。key 不存在，pivot
不在列表中，不执行任何操作。
返回值：命令执行成功，返回新列表的长度。没有找到 pivot 返回 -1， key 不存在返回 0。

例 1：修改列表 arch，在值 dao 之前加入 service

例 2：操作不存在的 pivot

2.3.5 应用-商品评论列表

思路：
在Redis中创建商品评论列表
用户发布商品评论，将评论信息转成json存储到list中。
用户在页面查询评论列表，从redis中取出json数据展示到页面。

定义商品评论列表key：
商品编号为1001的商品评论key【items: comment:1001】

192.168.101.3:7001> LPUSH items:comment:1001 '{"id":1,"name":"商品不错，很好！！","date":1430295077289}'

2.4 集合类型 set

2.4.1 集合类型Set介绍

集合中的数据是不重复且没有顺序。
集合类型和列表类型的对比：

集合类型的常用操作是向集合中加入或删除元素、判断某个元素是否存在等，由于集合类型的Redis内部是使用值为空的散列表实现，所有这些操作的时间复杂度都为0(1)。
Redis还提供了多个集合之间的交集、并集、差集的运算。

2.4.2 基本命令

A、sadd
语法：sadd key member [member…]
作用：将一个或多个 member 元素加入到集合 key 当中，已经存在于集合的 member 元
素将被忽略，不会再加入。
返回值：加入到集合的新元素的个数。不包括被忽略的元素。

例 1：添加单个元素

例 2：添加多个元素

B、 smembers
语法：smembers key
作用：获取集合 key 中的所有成员元素，不存在的 key 视为空集合

例 1：查看集合的所有元素

例 2：查看不存在的集合

C、 sismember
语法：sismember key member
作用：判断 member 元素是否是集合 key 的成员
返回值：member 是集合成员返回 1，其他返回 0 。

例 1：检查元素是否存在集合中

D、scard
语法：scard key
作用：获取集合里面的元素个数
返回值：数字，key 的元素个数。其他情况返回 0 。

例 1：统计集合的大小

例 2：统计不存在的 key

E、 srem
语法：srem key member [member…]
作用：删除集合 key 中的一个或多个 member 元素，不存在的元素被忽略。
返回值：数字，成功删除的元素个数，不包括被忽略的元素。

例 1：删除存在的一个元素，返回数字 1

例 2：删除不存在的元素

2.4.3 其他命令

A、srandmember
语法：srandmember key [count]
作用：只提供 key，随机返回集合中一个元素，元素不删除，依然在集合中；提供了 count
时，count 正数, 返回包含 count 个数元素的集合，集合元素各不相同。count 是负
数，返回一个 count 绝对值的长度的集合，集合中元素可能会重复多次。
返回值：一个元素；多个元素的集合

例 1：随机显示集合的一个元素

例 2：使用 count 参数， count 是正数

例 3：使用 count 参数，count 是负数

B、 spop
语法：spop key [count]
作用：随机从集合中删除一个元素, count 是删除的元素个数。
返回值：被删除的元素，key 不存在或空集合返回 nil

例如 1：随机从集合删除一个元素

例 2：随机删除指定个数的元素

2.4.4 set命令运算

集合的差集运算 A-B
属于A并且不属于B的元素构成的集合。

语法：SDIFF key [key …]

127.0.0.1:6379> sadd setA 1 2 3
(integer) 3
127.0.0.1:6379> sadd setB 2 3 4
(integer) 3
127.0.0.1:6379> sdiff setA setB 
1) "1"
127.0.0.1:6379> sdiff setB setA 
1) "4"

集合的交集运算 A ∩ B
属于A且属于B的元素构成的集合。

语法：SINTER key [key …]

127.0.0.1:6379> sinter setA setB 
1) "2"
2) "3"

集合的并集运算 A ∪ B
属于A或者属于B的元素构成的集合

语法：SUNION key [key …]

127.0.0.1:6379> sunion setA setB
1) "1"
2) "2"
3) "3"
4) "4"

2.5 有序集合类型 zset (sorted set)

2.5.1 sorted set介绍

在集合类型的基础上，有序集合类型为集合中的每个元素都关联一个分数，这使得我们不仅可以完成插入、删除和判断元素是否存在在集合中，还能够获得分数最高或最低的前N个元素、获取指定分数范围内的元素等与分数有关的操作。

在某些方面有序集合和列表类型有些相似。
1、二者都是有序的。
2、二者都可以获得某一范围的元素。
但是，二者有着很大区别：
1、列表类型是通过链表实现的，获取靠近两端的数据速度极快，而当元素增多后，访问中间数据的速度会变慢。
2、有序集合类型使用散列表实现，所有即使读取位于中间部分的数据也很快。
3、列表中不能简单的调整某个元素的位置，但是有序集合可以（通过更改分数实现）
4、有序集合要比列表类型更耗内存。

2.5.2 基本命令

A、zadd
语法：zadd key score member [score member…]
作用：将一个或多个 member 元素及其 score 值加入到有序集合 key 中，如果 member
存在集合中，则更新值；score 可以是整数或浮点数
返回值：数字，新添加的元素个数

例 1：创建保存学生成绩的集合

例 2：使用浮点数作为 score

B、 zrange
语法：zrange key start stop [WITHSCORES]
作用：查询有序集合，指定区间的内的元素。集合成员按 score 值从小到大来排序。 start，
stop 都是从 0 开始。0 是第一个元素，1 是第二个元素，依次类推。以 -1 表示最后一
个成员，-2 表示倒数第二个成员。WITHSCORES 选项让 score 和 value 一同返回。
返回值：自定区间的成员集合

例 1：显示集合的全部元素，不显示 score，不使用 WITHSCORES

例 2：显示集合全部元素，并使用 WITHSCORES

例 3：显示第 0,1 二个成员

例 4：排序显示浮点数的 score

C、 zrevrange
语法：zrevrange key start stop [WITHSCORES]
作用：返回有序集 key 中，指定区间内的成员。其中成员的位置按 score 值递减(从大到小)
来排列。其它同 zrange 命令。
返回值：自定区间的成员集合

例 1：成绩榜

D、zrem
语法：zrem key member [member…]
作用：删除有序集合 key 中的一个或多个成员，不存在的成员被忽略
返回值：被成功删除的成员数量，不包括被忽略的成员。

例 1：删除指定一个成员 wangwu

E、 zcard
语法：zcard key
作用：获取有序集 key 的元素成员的个数
返回值：key 存在返回集合元素的个数， key 不存在，返回 0

例 1：查询集合的元素个数

2.5.3 其他命令

A、zrangebyscore
语法：zrangebyscore key min max [WITHSCORES ] [LIMIT offset count]
作用：获取有序集 key 中，所有 score 值介于 min 和 max 之间（包括 min 和 max）的成
员，有序成员是按递增（从小到大）排序。
min ,max 是包括在内，使用符号 ( 表示不包括。 min ， max 可以使用 -inf ，
+inf 表示最小和最大 limit 用来限制返回结果的数量和区间。
withscores 显示 score 和 value
返回值：指定区间的集合数据
使用的准备数据

例 1：显示指定具体区间的数据

例 2：显示指定具体区间的集合数据，开区间（不包括 min，max）

例 3：显示整个集合的所有数据

例 4：使用 limit
增加新的数据：

显示从第一个位置开始，取一个元素。

B、 zrevrangebyscore
语法：zrevrangebyscore key max min [WITHSCORES ] [LIMIT offset count]
作用：返回有序集 key 中， score 值介于 max 和 min 之间(默认包括等于 max 或 min )的所有
的成员。有序集成员按 score 值递减(从大到小)的次序排列。其他同 zrangebyscore

例 1：查询工资最高到 3000 之间的员工

C、 zcount
语法：zcount key min max
作用：返回有序集 key 中， score 值在 min 和 max 之间(默认包括 score 值等于 min 或 max )
的成员的数量

例 1：求工资在 3000-5000 的员工数量

2.5.4 应用-商品销售排行榜

需求：根据商品销售量对商品进行排行显示
思路：定义商品销售排行榜（sorted set集合），Key为items:sellsort，分数为商品销售量。

写入商品销售量：
商品编号1001的销量是9，商品编号1002的销量是10

192.168.101.3:7007> ZADD items:sellsort 9 1001 10 1002

商品编号1001的销量加1

192.168.101.3:7001> ZINCRBY items:sellsort 1 1001

商品销量前10名：

192.168.101.3:7001> ZRANGE items:sellsort 0 9 withscores

redis的发布订阅

1.发布和订阅

1.1什么是发布和订阅

发布订阅是一种应用程序（系统）之间通讯，传递数据的技术手段。特别是在异构（不
同语言）系统之间作用非常明显。发布订阅可以是实现应用（系统）之间的解耦合。

● 发布订阅：类似微信中关注公众号/订阅号，公众号/订阅号发布的文章，信息。订阅
者能及时获取到最新的内容。微博的订阅也是类似的。日常生活中听广播，看电视。都
需要有信息的发布者，收听的人需要订阅（广播、电视需要调动某个频道）。
发布订阅是一对多的关系。

● 订阅：对某个内容感兴趣，需要实时获取新的内容。只要关注的内容有变化就能立即得
到通知。多的一方。

● 发布：提供某个内容，把内容信息发送给多个对此内容感兴趣的订阅者。是有主动权，
是一的一方。

发布订阅应用在即时通信应用中较多，比如网络聊天室，实时广播、实时提醒等。滴滴
打车软件的抢单；外卖的抢单；在微信群发红包，抢红包都可以使用发布订阅实现。

1.2 Redis的发布和订阅

Redis发布订阅(pub/sub)是一种消息通信模式：发送者(publish)发送消息，订阅者
(subscribe)接收消息。发布订阅也叫生产者消费者模式,是实现消息队列的一种方式。

1.3 如何实现

发布订阅的相关命令：

A、publish发布消息

语法：publish chanel message
作用：将message消息发送到channel频道。message是要发送的消息，channel是自定
义的频道名称（例如cctv1，cctv5），唯一标识发布者。
返回值：数字。接收到消息订阅者的数量

B、subscribe订阅频道
语法：subscribe channel[channel…]
作用：订阅一个或多个频道的信息
返回值：订阅的消息

C、unsubscribe退订频道
语法：unsubscribe channel [channel]
作用：退出指定的频道，不订阅。
返回值：退订的告知消息

1.3.1 命令行实现

注意要启动订阅者，等待接收发布者的消息，否则订阅者接收不到消息
A、开启4个redis客户端，3个客户端作为消息订阅者，1个为消息发布者：./redis-cli

在发布者或其他窗口启动redis

B、让3个消息订阅者订阅某个频道主题：subscribechannel
在订阅者的三个窗口中分别启动redis客户端，redis安装目录/src下执行./redis-cli

C、让1个消息发布者向频道主题上发布消息：publish channel message在发布者窗口：

D、然后观察消息的发布和订阅情况，在任意一个订阅窗口：

1.3.2 编程实现

Jedis是Java操作Redis的API工具类，类似于JDBC

JedisPubSub类：Jedis中的JedisPubSub类是Jedis的一个抽象类，此类定义了publish
/subscribe的回调方法，通过继承JedisPubSub类，重写回调方法。实现java中Redis
的发布订阅。当Reids发生发布或订阅的相关事件时会调用这些回调方法。只在回调方法中
实现自己的业务逻辑。

onMessage()：发布者发布消息时，会执行订阅者的回调方法onMessage(),接收发布的
消息。在此方法实现消息接收后的，自定义业务逻辑处理，比如访问数据库，更新库存等。

A、订阅者SUB工程
①：新建 Java Project，名称：MyRedisSubScribe
②：导入 jar : jedis-2.9.0.jar 加入 Build Path

③：定义订阅者者 RedisSubScribe ,继承 JedisPubSub

public class RedisSubScribe extends JedisPubSub {

	/**
	 * 当订阅者接收到消息时回自动调用改方法 String channel--->频道的名称 String message--->发布的消息
	 */
	@Override
	public void onMessage(String channel, String message) {
		SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		System.out.println("我是订阅者:订阅频道[" + channel + "],收到的消息是:[" + message + "],时间为:[" + df.format(new Date()) + "]");
	}

	public static void main(String[] args) {
		// 创建Jedis
		Jedis jedis = new Jedis("192.168.6.129", 6379);
		// 创建redisSubScribe对象
		RedisSubScribe redisSubScribe = new RedisSubScribe();
		// 从Redis订阅
		jedis.subscribe(redisSubScribe, "cctv6");
	}
}

B、发布者工程
①：新建 Java Project 名称 MyRedisPublish
②：导入 jar : jedis-2.9.0.jar 加入 Build Path

③：定义发布者类。发布消息

public class MyRedisPublish {

	public static void main(String[] args) {
		// 创建Jedis
		Jedis jedis = new Jedis("192.168.6.129", 6379);
		jedis.publish("cctv6", "战狼2");
		System.out.println("发布消息完毕....");
	}
}

C、执行程序
先启动订阅者，在运行发布者。

redis的事务

1.Redis事务

1.1.什么是事务

事务是指一系列操作步骤，这一系列的操作步骤，要么完全地执行，要么完全地不执行。
比如微博中：A用户关注了B用户，那么A的关注人列表里面就会有B用户，B的粉丝列表
里面就会有A用户。

这个关注与被关注的过程是由一系列操作步骤构成：
（1）A用户添加到B的粉丝列表里面
（2）B用户添加到A的关注列表里面；
这两个步骤必须全部执行成功，整个逻辑才是正确的，否则就会产生数据的错误，比如A
用户的关注列表有B用户，但B的粉丝列表里没有A用户；
要保证一系列的操作都完全成功，提出了事务控制的概念。

Redis中的事务（transaction）是一组命令的集合，至少是两个或两个以上的命令，redis事
务保证这些命令被执行时中间不会被任何其他操作打断。

2.2.事务操作的命令

multi ———–> beginTransaction 开启事务
语法： multi
作用：标记一个事务的开始。事务内的多条命令会按照先后顺序被放进一个队列当中。
返回值：总是返回 ok

exec ——-> commit提交
语法：exec
作用：执行所有事务块内的命令
返回值：事务内的所有执行语句内容，事务被打断（影响）返回 nil

discard——> Rollback 回滚
语法：discard
作用：取消事务，放弃执行事务块内的所有命令
返回值：总是返回 ok

watch
语法：watch key [key …]
作用：监视一个(或多个) key ，如果在事务执行之前这个(或这些) key 被其他命令所改动，
那么事务将被打断。
返回值：总是返回 ok

unwatch
语法：unwatch
作用：取消 WATCH 命令对所有 key 的监视。如果在执行 WATCH 命令之后， EXEC 命令
或 DISCARD 命令先被执行了的话，那么就不需要再执行 UNWATCH 了
返回值：总是返回 ok

2.3.事务的实现

● 正常执行事务
事务的执行步骤：首先开启事务，其次向事务队列中加入命令，最后执行事务提交

例 1：事务的执行:
1）multi ：用 multi 命令告诉 Redis，接下来要执行的命令你先不要执行，而是把它们暂
时存起来（开启事务）
2）sadd works john 第一条命令进入等待队列（命令入队）
3）sadd works rose 第二条命令进入等待队列（命令入队）
4）exce 告知 redis 执行前面发送的两条命令

查看works集合

● 事务执行exec之前，入队命令错误（语法错误；严重错误导致服务器不能正常工作（例如内存不足）），放弃事务。

执行事务步骤：
1）MULTI正常命令
2）SETkeyvalue正常命令
3）INCR命令语法错误
4）EXEC无法执行事务，那么第一条正确的命令也不会执行，所以key的值不会设置成
功

结论：事务执行exec之前，入队命令错误，事务终止，取消，不执行。

● 事务执行exec命令后，执行队列命令，命令执行错误，事务提交
执行步骤：
1）MULTI正常命令
2）SET username zhangsan正常命令
3）lpop username正常命令，语法没有错误，执行命令时才会有错误。
4）EXEC正常执行,发现错误可以在事务提交前放弃事务，执行discard.

结论：在 exec 执行后的所产生的错误，即使事务中有某个/某些命令在执行时产生了错误，
事务中的其他命令仍然会继续执行。Redis 在事务失败时不进行回滚，而是继续执行余下的命令。

Redis 这种设计原则是：Redis 命令只会因为错误的语法而失败（这些问题不能在入队时发
现），或是命令用在了错误类型的键上面，失败的命令并不是 Redis 导致，而是由编程错误
造成的，这样错误应该在开发的过程中被发现，生产环境中不应出现语法的错误。就是在程
序的运行环境中不应该出现语法的错误。而 Redis 能够保证正确的命令一定会被执行。再者不需要对回滚进行支持，所以 Redis 的内部可以保持简单且快速。

● 放弃事务
执行步骤：

MULTI 开启事务
SET age 25 命令入队
SET age 30 命令入队
DISCARD 放弃事务，则命令队列不会被执行
例 1：

● Redis的watch机制
A、Redis的WATCH机制
WATCH机制原理：
WATCH机制：使用WATCH监视一个或多个key,跟踪key的value修改情况，如果有
key的value值在事务EXEC执行之前被修改了，整个事务被取消。EXEC返回提示信息，表
示事务已经失败。
WATCH机制使的事务EXEC变的有条件，事务只有在被WATCH的key没有修改的前提下
才能执行。不满足条件，事务被取消。使用WATCH监视了一个带过期时间的键，那么即使
这个键过期了，事务仍然可以正常执行

大多数情况下，不同的客户端会访问不同的键，相互同时竞争同一key的情况一般都
很少，乐观锁能够以很好的性能解决数据冲突的问题。

B、何时取消key的监视（WATCH）？
①WATCH命令可以被调用多次。对键的监视从WATCH执行之后开始生效，直到调
用EXEC为止。不管事务是否成功执行，对所有键的监视都会被取消。
②当客户端断开连接时，该客户端对键的监视也会被取消。
③UNWATCH命令可以手动取消对所有键的监视

C、 WATCH 的事例
执行步骤：
首先启动 redis-server , 在开启两个客户端连接。分别叫 A 客户端和 B 客户端。
启动 Redis 服务器

A 客户端：WATCH 某个 key, 同时执行事务
B 客户端：对 A 客户端 WATCH 的 key 修改其 value 值

1）在 A 客户端设置 key : str.lp 登录人数为 10
2）在 A 客户端监视 key : str.lp
3）在 A 客户端开启事务 multi
4）在 A 客户端修改 str.lp 的值为 11
5）在 B 客户端修改 str.lp 的值为 15
6）在 A 客户端执行事务 exec
7）在 A 客户端查看 str.lp 值，A 客户端执行的事务没有提交，因为 WATCH 的 str.lp 的值已
经被修改了，所有放弃事务。

例 1：乐观锁

springboot整合SpringDataRedis

1. SpringDataRedis简介

1.1.Jedis（过时）

Jedis是Redis官方推出的一款面向Java的客户端，提供了很多接口供Java语言调用。可以在Redis官网下载，当然还有一些开源爱好者提供的客户端，如Jredis、SRP等等，推荐使用Jedis。

Jedis和SpringBoot整合繁琐

1.2.Spring Data Redis（推荐）

Spring-data-redis是spring大家族的一部分，提供了在srping应用中通过简单的配置访问redis服务，对reids底层开发包(Jedis, JRedis, and RJC)进行了高度封装，RedisTemplate提供了redis各种操作、异常处理及序列化，支持发布订阅，并对spring 3.1 cache进行了实现。
spring-data-redis针对jedis提供了如下功能：
1.连接池自动管理，提供了一个高度封装的“RedisTemplate”类
2.针对jedis客户端中大量api进行了归类封装,将同一类型操作封装为operation接口

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

2.Spring Data Redis入门小Demo

2.1.准备工作

（1）构建Maven工程SpringDataRedisDemo
（2）引入Springboot和SpringDataRedis依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <!-- lookup parent from repository -->
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.4.RELEASE</version>
        <relativePath/>
    </parent>

    <groupId>com.bruceliu.redis.pubsub</groupId>
    <artifactId>springboot-redis</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <!-- springBoot 的启动器 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!-- Spring Data Redis 的启动器 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>2.9.0</version>
        </dependency>
    </dependencies>

</project>

（4）在src/main/resources下创建properties文件夹，建立application.properties

spring.redis.jedis.pool.max-idle=10
spring.redis.jedis.pool.min-idle=5
spring.redis.pool.max-total=20
spring.redis.hostName=122.51.50.249
spring.redis.port=6379

(5) 添加Redis的配置类
添加Redis的java配置类，设置相关的信息。

package com.bruceliu.config;

import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import redis.clients.jedis.JedisPoolConfig;

@Configuration
public class RedisConfig {

    /**
     * 1.创建JedisPoolConfig对象。在该对象中完成一些链接池配置
     * @ConfigurationProperties:会将前缀相同的内容创建一个实体。
     */
    @Bean
    @ConfigurationProperties(prefix="spring.redis.pool")
    public JedisPoolConfig jedisPoolConfig(){
        JedisPoolConfig config = new JedisPoolConfig();
		/*//最大空闲数
		config.setMaxIdle(10);
		//最小空闲数
		config.setMinIdle(5);
		//最大链接数
		config.setMaxTotal(20);*/
        System.out.println("默认值："+config.getMaxIdle());
        System.out.println("默认值："+config.getMinIdle());
        System.out.println("默认值："+config.getMaxTotal());
        return config;
    }

    /**
     * 2.创建JedisConnectionFactory：配置redis链接信息
     */
    @Bean
    @ConfigurationProperties(prefix="spring.redis")
    public JedisConnectionFactory jedisConnectionFactory(JedisPoolConfig config){
        System.out.println("配置完毕："+config.getMaxIdle());
        System.out.println("配置完毕："+config.getMinIdle());
        System.out.println("配置完毕："+config.getMaxTotal());

        JedisConnectionFactory factory = new JedisConnectionFactory();
        //关联链接池的配置对象
        factory.setPoolConfig(config);
        //配置链接Redis的信息
        //主机地址
		/*factory.setHostName("192.168.70.128");
		//端口
		factory.setPort(6379);*/
        return factory;
    }

    /**
     * 3.创建RedisTemplate:用于执行Redis操作的方法
     */
    @Bean
    public RedisTemplate<String,Object> redisTemplate(JedisConnectionFactory factory){
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        //关联
        template.setConnectionFactory(factory);

        //为key设置序列化器
        template.setKeySerializer(new StringRedisSerializer());
        //为value设置序列化器
        template.setValueSerializer(new StringRedisSerializer());

        return template;
    }
}

2.2.实体类

package com.bruceliu.bean;

import java.io.Serializable;

public class Users implements Serializable {

    private static final long serialVersionUID = 6206472024994638411L;

    private Integer id;
    private String name;
    private Integer age;
    public Integer getId() {
        return id;
    }
    public void setId(Integer id) {
        this.id = id;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public Integer getAge() {
        return age;
    }
    public void setAge(Integer age) {
        this.age = age;
    }
    @Override
    public String toString() {
        return "Users [id=" + id + ", name=" + name + ", age=" + age + "]";
    }

}

3. 字符串类型操作

package com.bruceliu.test;

import com.bruceliu.APP;
import com.bruceliu.bean.Users;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest(classes = APP.class)
public class SpringbootRedisDemoApplicationTests {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    /**
     * 添加一个字符串
     */
    @Test
    public void testSet(){
        this.redisTemplate.opsForValue().set("key", "bruceliu...");
    }

    /**
     * 获取一个字符串
     */
    @Test
    public void testGet(){
        String value = (String)this.redisTemplate.opsForValue().get("key");
        System.out.println(value);
    }

    /**
     * 添加Users对象
     */
    @Test
    public void testSetUesrs(){
        Users users = new Users();
        users.setAge(20);
        users.setName("张三丰");
        users.setId(1);
        //重新设置序列化器
        this.redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
        this.redisTemplate.opsForValue().set("users", users);
    }

    /**
     * 取Users对象
     */
    @Test
    public void testGetUsers(){
        //重新设置序列化器
        this.redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
        Users users = (Users)this.redisTemplate.opsForValue().get("users");
        System.out.println(users);
    }

    /**
     * 基于JSON格式存Users对象
     */
    @Test
    public void testSetUsersUseJSON(){
        Users users = new Users();
        users.setAge(20);
        users.setName("李四丰");
        users.setId(1);
        this.redisTemplate.setValueSerializer(new Jackson2JsonRedisSerializer<>(Users.class));
        this.redisTemplate.opsForValue().set("users_json", users);
    }

    /**
     * 基于JSON格式取Users对象
     */
    @Test
    public void testGetUseJSON(){
        this.redisTemplate.setValueSerializer(new Jackson2JsonRedisSerializer<>(Users.class));
        Users users = (Users)this.redisTemplate.opsForValue().get("users_json");
        System.out.println(users);
    }

    
      /**
     * 把一个对象序列化存入Redis中，String类型
     */
    @Test
    public void testSetUsersUseJSON_1(){
        Users users1 = new Users(111,"张三丰",123);
        Users users2 = new Users(111,"张三丰",123);
        List<Users> usersList=new ArrayList<>();
        usersList.add(users1);
        usersList.add(users2);
        //重新设置序列化器
        this.redisTemplate.setValueSerializer(new Jackson2JsonRedisSerializer<>(List.class));
        this.redisTemplate.opsForValue().set("usersList_json", usersList);
    }

    /**
     * 基于JSON格式取Users对象
     */
    @Test
    public void testGetUseJSON_1(){
        this.redisTemplate.setValueSerializer(new Jackson2JsonRedisSerializer<>(List.class));
        List<Users> usersList = (List<Users>)this.redisTemplate.opsForValue().get("usersList_json");
        System.out.println(usersList);
    }
}

4. SET类型操作

package com.bruceliu.test;

import com.bruceliu.APP;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.Set;
@RunWith(SpringRunner.class)
@SpringBootTest(classes = APP.class)
public class TestSet {

    @Autowired
    private RedisTemplate redisTemplate;

    @Test
    public void setValue() {
        redisTemplate.boundSetOps("nameset").add("曹操");
        redisTemplate.boundSetOps("nameset").add("刘备");
        redisTemplate.boundSetOps("nameset").add("孙权");
    }

    @Test
    public void getValue() {
        Set set = redisTemplate.boundSetOps("nameset").members();
        System.out.println(set);
    }

    @Test
    public void removeValue() {
        redisTemplate.boundSetOps("nameset").remove("孙权");
    }

    @Test
    public void delete() {
        redisTemplate.delete("nameset");
    }
}

5. List类型操作

package com.bruceliu.test;

import com.bruceliu.APP;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.List;
import java.util.Set;

@RunWith(SpringRunner.class)
@SpringBootTest(classes = APP.class)
public class TestList {

    @Autowired
    private RedisTemplate redisTemplate;

    /*
     * 右压栈 : 后添加的元素排在后边
     */
    @Test
    public void testSetValue1() {
        redisTemplate.boundListOps("namelist1").rightPush("刘备");
        redisTemplate.boundListOps("namelist1").rightPush("关羽");
        redisTemplate.boundListOps("namelist1").rightPush("张飞");
    }

    /**
     * 显示右压栈的值
     */
    @Test
    public void testGetValue1() {
        List list = redisTemplate.boundListOps("namelist1").range(0, 10);
        System.out.println(list);
    }

    @Test
    public void delete() {
        redisTemplate.delete("namelist1");
    }

    /**
     * 左压栈
     */
    @Test
    public void testSetValue2() {
        redisTemplate.boundListOps("namelist2").leftPush("刘备");
        redisTemplate.boundListOps("namelist2").leftPush("关羽");
        redisTemplate.boundListOps("namelist2").leftPush("张飞");
    }

    /**
     * 显示左压栈的值
     */
    @Test
    public void testGetValue2() {
        List list = redisTemplate.boundListOps("namelist2").range(0, 10);
        System.out.println(list);
    }

    /**
     * 删除值
     */
    @Test
    public void removeValue() {
        redisTemplate.boundListOps("namelist1").remove(0, "刘备");
    }
}

6. Hash类型操作

package com.bruceliu.test;

import com.bruceliu.APP;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.List;
import java.util.Set;


@RunWith(SpringRunner.class)
@SpringBootTest(classes = APP.class)
public class TestHash {

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 存值
     */
    @Test
    public void testSetValue() {
        redisTemplate.boundHashOps("namehash").put("a", "唐僧");
        redisTemplate.boundHashOps("namehash").put("b", "悟空");
        redisTemplate.boundHashOps("namehash").put("c", "八戒");
        redisTemplate.boundHashOps("namehash").put("d", "沙僧");
    }

    /**
     * 获取所有的key
     */
    @Test
    public void testGetKes() {
        Set keys = redisTemplate.boundHashOps("namehash").keys();
        System.out.println(keys);
    }

    /**
     * 获取所有的值
     */
    @Test
    public void testGetValues() {
        List list = redisTemplate.boundHashOps("namehash").values();
        System.out.println(list);
    }

    /**
     * 根据KEY取值
     */
    @Test
    public void searchValueByKey() {
        String str = (String) redisTemplate.boundHashOps("namehash").get("b");
        System.out.println(str);
    }

    /**
     * 移除某个小key的值
     */
    @Test
    public void removeValue() {
        redisTemplate.boundHashOps("namehash").delete("c");
    }
}

7. Zset类型操作

package com.bruceliu.test;

import com.bruceliu.APP;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.redis.core.DefaultTypedTuple;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

@RunWith(SpringRunner.class)
@SpringBootTest(classes = APP.class)
public class TestZset {

    @Autowired
    RedisTemplate redisTemplate;

    /**
     * Boolean add(V value, double score);
     * 添加元素到变量中同时指定元素的分值。
     */
    @Test
    public void testAdd() {
        redisTemplate.boundZSetOps("zSetValue").add("A", 1);
        redisTemplate.boundZSetOps("zSetValue").add("B", 3);
        redisTemplate.boundZSetOps("zSetValue").add("C", 2);
        redisTemplate.boundZSetOps("zSetValue").add("D", 5);
        redisTemplate.boundZSetOps("zSetValue").add("E", 4);
    }

    /**
     * Set<V> range(long start, long end);
     * 获取变量指定区间的元素。
     * 通过索引区间返回有序集合成指定区间内的成员，其中有序集成员按分数值递增(从小到大)顺序排列
     */
    @Test
    public void testRange(){
        Set zSetValue = redisTemplate.boundZSetOps("zSetValue").range(0,-1);
        System.out.println("获取指定区间的元素:" + zSetValue);
    }

    /**
     * 新增一个有序集合
     */
    @Test
    public void testAdd1(){
        ZSetOperations.TypedTuple<Object> objectTypedTuple1 = new DefaultTypedTuple<Object>("zset-5",9.6);
        ZSetOperations.TypedTuple<Object> objectTypedTuple2 = new DefaultTypedTuple<Object>("zset-6",9.9);
        Set<ZSetOperations.TypedTuple<Object>> tuples = new HashSet<ZSetOperations.TypedTuple<Object>>();
        tuples.add(objectTypedTuple1);
        tuples.add(objectTypedTuple2);

        System.out.println(redisTemplate.boundZSetOps("zset1").add(tuples));

        System.out.println(redisTemplate.boundZSetOps("zset1").range(0,-1));
    }


    /**
     * 从有序集合中移除一个或者多个元素
     */
    @Test
    public void testRemove(){
        System.out.println(redisTemplate.boundZSetOps("zset1").range(0,-1));
        System.out.println(redisTemplate.boundZSetOps("zset1").remove("zset1","zset-6"));
        System.out.println(redisTemplate.boundZSetOps("zset1").range(0,-1));
    }

    /**
     * 增加元素的score值，并返回增加后的值
     */
    @Test
    public void testIncrementScore(){
        System.out.println(redisTemplate.boundZSetOps("zset1").incrementScore("zset-1",1.1));  //原为1.1
    }

    /**
     * 返回有序集中指定成员的排名，其中有序集成员按分数值递增(从小到大)顺序排列
     */
    @Test
    public void testRank(){
        System.out.println(redisTemplate.boundZSetOps("zset1").range(0,-1));
        System.out.println(redisTemplate.boundZSetOps("zset1").rank("zset-5"));
    }

    /**
     * 返回有序集中指定成员的排名，其中有序集成员按分数值递减(从大到小)顺序排列
     */
    @Test
    public void testReverseRank(){
        System.out.println(redisTemplate.boundZSetOps("zset1").range(0,-1));
        System.out.println(redisTemplate.boundZSetOps("zset1").reverseRank("zset-5"));
    }

    /**
     * 通过索引区间返回有序集合成指定区间内的成员对象，其中有序集成员按分数值递增(从小到大)顺序排列
     */
    @Test
    public void testRangeWithScore(){
        Set<ZSetOperations.TypedTuple<Object>> tuples = redisTemplate.boundZSetOps("zset1").rangeWithScores(0,-1);
        Iterator<ZSetOperations.TypedTuple<Object>> iterator = tuples.iterator();
        while (iterator.hasNext()) {
            ZSetOperations.TypedTuple<Object> typedTuple = iterator.next();
            System.out.println("value:" + typedTuple.getValue() + "score:" + typedTuple.getScore());
        }
    }

    /**
     * 获取有序集合的成员数
     */
    @Test
    public void testzCard(){
        System.out.println(redisTemplate.boundZSetOps("zset1").zCard());
    }

    /**
     * 移除指定索引位置的成员，其中有序集成员按分数值递增(从小到大)顺序排列
     */
    @Test
    public void testRemoveRange(){
        System.out.println(redisTemplate.boundZSetOps("zset1").range(0,-1));
        redisTemplate.boundZSetOps("zset1").removeRange(1,2);
        System.out.println(redisTemplate.boundZSetOps("zset1").range(0,-1));
    }
}

redis的主从复制

1.主从复制概念（master/slave）

持久化保证了即使redis服务重启也不会丢失数据，因为redis服务重启后会将硬盘上持久化的数据恢复到内存中，但是当redis服务器的硬盘损坏了可能会导致数据丢失，如果通过redis的主从复制机制就可以避免这种单点故障，如下图：

说明：

主redis中的数据有两个副本（replication）即从redis1和从redis2，即使一台redis服务器宕机其它两台redis服务也可以继续提供服务。
主redis中的数据和从redis上的数据保持实时同步，当主redis写入数据时通过主从复制机制会复制到两个从redis服务上。
只有一个主redis，可以有多个从redis。
主从复制不会阻塞master，在同步数据时，master 可以继续处理client 请求

2.主从复制（master/slave）实现

修改配置文件，启动时，服务器读取配置文件，并自动成为指定服务器的从服
务器，从而构成主从复制的关系

A、新建三个Redis的配置

B、修改三个Redis的配置端口

C、修改其中2个从Redis的配置

D、启动服务器 Master/Slave 都启动
启动方式 ./redis-server 配置文件
启动 Redis,并查看启动进程

E、查看配置后的服务信息
命令：
①： Redis 客户端使用指定端口连接 Redis 服务器
./bin/redis-cli -h 192.168.6.129 -p 6379
②：查看服务器信息
info replication

在新的窗口分别登录到 6380 ，6381 查看信息

F、向 Master 写入数据
在 6379 执行 flushall 清除数据，避免干扰的测试数据。生产环境避免使用

G、在从 Slave 读数据
6380,6380都可以读主 Master 的数据，不能写

Slave 写数据失败

3.主从复制（master/slave）容灾处理

当Master 服务出现故障，需手动将 slave 中的一个提升为 master，剩下的 slave 挂至新的
master 上（冷处理：机器挂掉了，再处理）
命令：
①：slaveof no one，将一台 slave 服务器提升为 Master （提升某 slave 为 master）
②：slaveof 127.0.0.1 6381 （将 slave 挂至新的 master 上）

执行步骤：
A、将 Master:6379停止（模拟挂掉）

B、选择一个 Slave 升到 Master，其它的 Slave 挂到新提升的 Master

C、将其他 Slave 挂到新的 Master
在 Slave 6380 上执行

现在的主从（Master/Slave）关系：Master 是 6381 ， Slave 是 6380
查看 6381 ：

D、原来的服务器重新添加到主从结构中
6379的服务器修改后，从新工作，需要把它添加到现有的 Master/Slave 中
先启动 6379 的 Redis 服务

连接到 6379 端口

新增加的默认是master
当前服务挂到 Master上

E、查看新的 Master 信息
在 6381 执行

现在的 Master/Slaver 关系是：
Master: 6381
Slave: 6380
6379

4.总结

1、一个 master 可以有多个 slave
2、slave 下线，读请求的处理性能下降
3、master 下线，写请求无法执行
4、当 master 发生故障，需手动将其中一台 slave 使用 slaveof no one 命令提升为 master，其
它 slave 执行 slaveof 命令指向这个新的 master，从新的 master 处同步数据
5、主从复制模式的故障转移需要手动操作，要实现自动化处理，这就需要 Sentinel 哨兵，
实现故障自动转移

redis的哨兵模式

1.高可用Sentinel哨兵介绍

Sentinel哨兵是redis官方提供的高可用方案，可以用它来监控多个Redis服务实例的运
行情况。RedisSentinel是一个运行在特殊模式下的Redis服务器。RedisSentinel是在多个
Sentinel进程环境下互相协作工作的。
Sentinel系统有三个主要任务：

监控：Sentinel不断的检查主服务和从服务器是否按照预期正常工作。
提醒：被监控的Redis出现问题时，Sentinel会通知管理员或其他应用程序。
自动故障转移：监控的主Redis不能正常工作，Sentinel会开始进行故障迁移操作。将
一个从服务器升级新的主服务器。让其他从服务器挂到新的主服务器。同时向客户端
提供新的主服务器地址。

2.高可用Sentinel哨兵环境搭建

（1）Sentinel 配置文件
复制三份sentinel.conf文件

（2）三份 sentinel 配置文件修改：
1、修改 port 26379、 port 26380、 port 26381

2、修改 sentinel monitor mymaster 127.0.0.1 6380 2
格式：Sentinel monitor <Quorum 投票数>
Sentinel监控主(Master)Redis, Sentinel根据Master的配置自动发现Master的Slave,Sentinel
默认端口号为26379

Sentinel监控主(Master)Redis, Sentinel根据Master的配置自动发现Master的Slave,Sentinel
默认端口号为26379 。

sentinel26380.conf
1)修改 port

2）修改监控的 master 地址

sentinel26379.conf 修改port 26379 , master的port 6381
sentinel26381.conf 修改port 26381 , master的port 6381

（3）启动主从（Master/Slave）Redis
启动 Reids

（4）启动 Sentinel
redis安装时make编译后就产生了redis-sentinel程序文件，可以在一个redis中运行多个
sentinel进程。
启动一个运行在Sentinel模式下的Redis服务实例
./redis-sentinel sentinel 配置文件
执行以下三条命令，将创建三个监视主服务器的Sentinel实例：

./redis-sentinel   ../sentinel26379.conf

./redis-sentinel ../sentinel26380.conf

./redis-sentinel ../sentinel26381.conf

（5）主 Redis 不能工作
让 Master 的 Redis 停止服务，执行 shutdown
先执行 info replication 确认 Master 的 Redis ，再执行 shutdown

（6） Sentinel 的起作用
在 Master 执行 shutdown 后，稍微等一会 Sentinel 要进行投票计算，从可用的 Slave
选举新的 Master。
查看 Sentinel 日志，三个 Sentinel 窗口的日志是一样的。

查看新的 Master

查看原 Slave 的变化

（7）新的 Redis 加入 Sentinel 系统，自动加入 Master
重新启动 6381
查看 6380 的信息

测试数据：在 Master 写入数据

在 6381 上读取数据，不能写入

（8）监控
1）Sentinel会不断检查Master和Slave是否正常
2）如果Sentinel挂了，就无法监控，所以需要多个哨兵，组成Sentinel网络，一个健康的
Sentinel至少有3个Sentinel应用。彼此在独立的物理机器或虚拟机。
3）监控同一个Master的Sentinel会自动连接，组成一个分布式的Sentinel网络，互相通信
并交换彼此关于被监控服务器的信息
4）当一个Sentinel认为被监控的服务器已经下线时，它会向网络中的其它Sentinel进行确
认，判断该服务器是否真的已经下线
5）如果下线的服务器为主服务器，那么Sentinel网络将对下线主服务器进行自动故障转移，
通过将下线主服务器的某个从服务器提升为新的主服务器，并让其从服务器转移到新的主服
务器下，以此来让系统重新回到正常状态
6）下线的旧主服务器重新上线，Sentinel会让它成为从，挂到新的主服务器下

（9）总结
主从复制，解决了读请求的分担，从节点下线，会使得读请求能力有所下降，Master下
线，写请求无法执行
Sentinel会在Master下线后自动执行故障转移操作，提升一台Slave为Master，并让其它
Slave成为新Master的Slave

Springboot连接哨兵模式

1.application.yml

###################以下为Redis增加的配置###########################
spring:
  redis:
    #单机配置
    # host: 122.51.50.249
    # port: 6380
    timeout: 6000
    # password: 123456
    ###################以下为redis哨兵增加的配置###########################
    sentinel:
      nodes: 122.51.50.249:26379,122.51.50.249:26380,122.51.50.249:26381
      master: mymaster
    ###################以下为lettuce连接池增加的配置###########################
    lettuce:
      pool:
        max-active:  100 # 连接池最大连接数（使用负值表示没有限制）
        max-idle: 100 # 连接池中的最大空闲连接
        min-idle: 50 # 连接池中的最小空闲连接
        max-wait: 6000 # 连接池最大阻塞等待时间（使用负值表示没有限制
  ###################以下为springcache增加的配置###########################
  cache:
    redis:
      use-key-prefix: true
      key-prefix: dev
      cache-null-values: false
      time-to-live: 20s

2.配置文件

RedisSentinelConfig.java

package com.bruceliu.config;


import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.*;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.connection.lettuce.LettucePoolingClientConfiguration;
import java.util.*;

@Configuration
@ConfigurationProperties(prefix = "spring.redis.sentinel")
public class RedisSentinelConfig {
    private Set<String> nodes;
    private String master;

    @Value("${spring.redis.timeout}")
    private long timeout;
    //@Value("${spring.redis.password}")
    //private String password;
    @Value("${spring.redis.lettuce.pool.max-idle}")
    private int maxIdle;
    @Value("${spring.redis.lettuce.pool.min-idle}")
    private int minIdle;
    @Value("${spring.redis.lettuce.pool.max-wait}")
    private long maxWait;
    @Value("${spring.redis.lettuce.pool.max-active}")
    private int maxActive;


    @Bean
    public RedisConnectionFactory lettuceConnectionFactory() {
        RedisSentinelConfiguration redisSentinelConfiguration = new RedisSentinelConfiguration(master, nodes);
        //redisSentinelConfiguration.setPassword(RedisPassword.of(password.toCharArray()));
        GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();
        genericObjectPoolConfig.setMaxIdle(maxIdle);
        genericObjectPoolConfig.setMinIdle(minIdle);
        genericObjectPoolConfig.setMaxTotal(maxActive);
        genericObjectPoolConfig.setMaxWaitMillis(maxWait);
        LettucePoolingClientConfiguration lettuceClientConfiguration = LettucePoolingClientConfiguration.builder()
                .poolConfig(genericObjectPoolConfig)
                .build();
        return new LettuceConnectionFactory(redisSentinelConfiguration, lettuceClientConfiguration);
    }

    public void setNodes(Set<String> nodes) {
        this.nodes = nodes;
    }

    public void setMaster(String master) {
        this.master = master;
    }
}

RedisTemplateConfig.java

package com.bruceliu.config;


import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;

/*
 * 将默认序列化改为Jackson2JsonRedisSerializer序列化，
 * */
@Configuration
public class RedisTemplateConfig {
    @Bean
    public RedisTemplate<String, Object> redisTemplate(LettuceConnectionFactory factory) {
        //创建Json序列化对象
        Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(Object.class);

        //解决查询缓存转换异常的问题
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);

        // 将默认序列化改为Jackson2JsonRedisSerializer序列化
        RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
        template.setKeySerializer(jackson2JsonRedisSerializer);// key序列化
        template.setValueSerializer(jackson2JsonRedisSerializer);// value序列化
        template.setHashKeySerializer(jackson2JsonRedisSerializer);// Hash key序列化
        template.setHashValueSerializer(jackson2JsonRedisSerializer);// Hash value序列化
        template.setConnectionFactory(factory);
        template.afterPropertiesSet();
        return template;
    }
}

SpringCacheRedisConfig.java

package com.bruceliu.config;


import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cache.CacheManager;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.cache.RedisCacheConfiguration;
import org.springframework.data.redis.cache.RedisCacheManager;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import java.time.Duration;

/*
 * 修改 CacheManager 接口，将默认序列化改为JSON序列化，并使用redis作为缓存
 * */
@Configuration
@EnableCaching//开启缓存
@ConfigurationProperties(prefix = "spring.cache.redis")
public class SpringCacheRedisConfig {

    private Duration timeToLive = Duration.ZERO;
    public void setTimeToLive(Duration timeToLive) {
        this.timeToLive = timeToLive;
    }

    @Bean
    public CacheManager cacheManager(LettuceConnectionFactory factory) {
        RedisSerializer<String> redisSerializer = new StringRedisSerializer();
        //创建Json序列化对象
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);

        //解决转换异常的问题
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);

        // 将默认序列化改为Jackson2JsonRedisSerializer序列化
        RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
                .entryTtl(timeToLive)
                .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(redisSerializer))
                .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(jackson2JsonRedisSerializer))
                .disableCachingNullValues();

        // 使用redis作为缓存，进入 CacheManager 接口按Control+H查看 CacheManager 的实现类，其中 RedisCacheManager 与redis有关
        RedisCacheManager cacheManager = RedisCacheManager.builder(factory)
                .cacheDefaults(config)
                .build();
        return cacheManager;
    }
}

redis-cluster集群搭建

1.Redis-cluster架构图

架构细节:

(1)所有的redis节点彼此互联(PING-PONG机制),内部使用二进制协议优化传输速度和带宽.

(2)节点的fail是通过集群中超过半数的节点检测失效时才生效.

(3)客户端与redis节点直连,不需要中间proxy层.客户端不需要连接集群所有节点,连接集群中任何一个可用节点即可

(4) redis-cluster把所有的物理节点映射到[0-16383]slot槽上,cluster 负责维护node<->slot<->value
Redis 集群中内置了 16384 个哈希槽，当需要在 Redis 集群中放置一个 key-value 时，redis 先对 key 使用 crc16 算法算出一个结果，然后把结果对 16384 求余数，这样每个 key 都会对应一个编号在 0-16383 之间的哈希槽，redis 会根据节点数量大致均等的将哈希槽映射到不同的节点！

示例如下：

2.Redis-cluster投票:容错

(1)集群中所有master参与投票,如果半数以上master节点与其中一个master节点通信超过(cluster-node-timeout),认为该master节点挂掉.

(2):什么时候整个集群不可用(cluster_state:fail)?
 如果集群任意master挂掉,且当前master没有slave，则集群进入fail状态。也可以理解成集群的[0-16383]slot映射不完全时进入fail状态。
 如果集群超过半数以上master挂掉，无论是否有slave，集群进入fail状态。

3.Redis-cluster集群搭建环境准备

集群管理工具（redis-trib.rb）是使用ruby脚本语言编写的。

第一步: 安装ruby

[root@redis1 bin]# yum install ruby
[root@redis1 bin]# yum install rubygems

第二步：将以下文件上传到linux系统

第三步：安装ruby和redis接口

-rw-r--r--. 1 root root 22301 8月  18 2016 dangdang.sql
drwxr-xr-x. 3 root root  4096 7月   1 10:00 redis
drwxr-xr-x. 3 root root  4096 7月   1 09:54 redis1
drwxrwxr-x. 6 root root  4096 4月   1 2015 redis-3.0.0
-rw-r--r--. 1 root root 57856 7月   1 10:03 redis-3.0.0.gem
drwxr-xr-x. 9 root root  4096 6月  28 23:26 tomcat
[root@bruce1 myapps]# gem install redis-3.0.0.gem

第四步：将redis-3.0.0包下src目录中的以下文件拷贝到redis-cluster/

[root@bruce1 myapps]# mkdir redis-cluster
[root@bruce1 myapps]# cd redis-3.0.0/src/
[root@bruce1 src]# cp redis-trib.rb ~/myapps/redis-cluster/
[root@bruce1 src]# ll ~/myapps/redis-cluster/
总用量 48
-rwxr-xr-x. 1 root root 48141 7月   1 10:20 redis-trib.rb

4.Redis-cluster集群搭建过程

搭建集群最少也得需要3台主机，如果每台主机再配置一台从机的话，则最少需要6台机器。
设计端口如下：创建6个redis实例，需要端口号7001~7006 伪集群

第一步：复制出一个7001实例

[root@bruce1 myapps]# cp redis/ redis-cluster/7001 -r 
[root@bruce1 myapps]# cd redis-cluster/7001
[root@bruce1 7001]# ll
总用量 52
drwxr-xr-x. 2 root root  4096 7月   1 10:22 bin
-rw-r--r--. 1 root root  3446 7月   1 10:22 dump.rdb
-rw-r--r--. 1 root root 41404 7月   1 10:22 redis.conf

第二步：如果存在持久化文件，则删除

[root@bruce1 bin]# rm -rf appendonly.aof dump.rdb

第三步：修改redis.conf配置文件，打开cluster-enable yes

第四步：修改端口

第五步：复制出7002-7006机器

[root@bruce1 redis-cluster]# cp 7001/ 7002 -r
[root@bruce1 redis-cluster]# cp 7001/ 7003 -r
[root@bruce1 redis-cluster]# cp 7001/ 7004 -r
[root@bruce1 redis-cluster]# cp 7001/ 7005 -r
[root@bruce1 redis-cluster]# cp 7001/ 7006 –r

第六步：修改7002-7006机器的端口
第七步：启动7001-7006这六台机器，写一个启动脚本

cd redis_cluster7001
./bin/redis-server ./redis.conf
cd ..
cd redis_cluster7002
./bin/redis-server ./redis.conf 
cd ..
cd redis_cluster7003
./bin/redis-server ./redis.conf
cd ..
cd redis_cluster7004
./bin/redis-server ./redis.conf
cd ..
cd redis_cluster7005
./bin/redis-server ./redis.conf
cd ..
cd redis_cluster7006
./bin/redis-server ./redis.conf
cd ..

第八步：修改start-all.sh文件的权限

[root@bruce1 redis-cluster]# chmod u+x start-all.sh

第九步：启动所有的实例

[root@bruce1 redis-cluster]# ./start-all.sh

第十步：创建集群

[root@bruce1 redis-cluster]# ./redis-trib.rb create --replicas 1 192.168.227.200:7001 192.168.227.200:7002 192.168.227.200:7003 192.168.227.200:7004 192.168.227.200:7005  192.168.227.200:7006
>>> Creating cluster
Connecting to node 192.168.221.128:7001: OK
Connecting to node 192.168.221.128:7002: OK
Connecting to node 192.168.221.128:7003: OK
Connecting to node 192.168.221.128:7004: OK
Connecting to node 192.168.221.128:7005: OK
Connecting to node 192.168.221.128:7006: OK
>>> Performing hash slots allocation on 6 nodes...
Using 3 masters:
192.168.221.128:7001
192.168.221.128:7002
192.168.221.128:7003
Adding replica 192.168.221.128:7004 to 192.168.221.128:7001
Adding replica 192.168.221.128:7005 to 192.168.221.128:7002
Adding replica 192.168.221.128:7006 to 192.168.221.128:7003
M: e7fb45e74f828b53ccd8b335f3ed587aa115b903 192.168.221.128:7001
   slots:0-5460 (5461 slots) master
M: 4a312b6fc90bfee187d43588ead99d83b407c892 192.168.221.128:7002
   slots:5461-10922 (5462 slots) master
M: 713218b88321e5067fd8ad25c3bf7db88c878ccf 192.168.221.128:7003
   slots:10923-16383 (5461 slots) master
S: 4f8c7455574e2f0aab1e2bb341eae319ac065039 192.168.221.128:7004
   replicates e7fb45e74f828b53ccd8b335f3ed587aa115b903
S: 8879c2ed9c141de70cb7d5fcb7d690ed8a200792 192.168.221.128:7005
   replicates 4a312b6fc90bfee187d43588ead99d83b407c892
S: b1183545245b3a710a95d669d7bbcbb5e09896a0 192.168.221.128:7006
   replicates 713218b88321e5067fd8ad25c3bf7db88c878ccf
Can I set the above configuration? (type 'yes' to accept): yes
>>> Nodes configuration updated
>>> Assign a different config epoch to each node
>>> Sending CLUSTER MEET messages to join the cluster
Waiting for the cluster to join....
>>> Performing Cluster Check (using node 192.168.221.128:7001)
M: e7fb45e74f828b53ccd8b335f3ed587aa115b903 192.168.221.128:7001
   slots:0-5460 (5461 slots) master
M: 4a312b6fc90bfee187d43588ead99d83b407c892 192.168.221.128:7002
   slots:5461-10922 (5462 slots) master
M: 713218b88321e5067fd8ad25c3bf7db88c878ccf 192.168.221.128:7003
   slots:10923-16383 (5461 slots) master
M: 4f8c7455574e2f0aab1e2bb341eae319ac065039 192.168.221.128:7004
   slots: (0 slots) master
   replicates e7fb45e74f828b53ccd8b335f3ed587aa115b903
M: 8879c2ed9c141de70cb7d5fcb7d690ed8a200792 192.168.221.128:7005
   slots: (0 slots) master
   replicates 4a312b6fc90bfee187d43588ead99d83b407c892
M: b1183545245b3a710a95d669d7bbcbb5e09896a0 192.168.221.128:7006
   slots: (0 slots) master
   replicates 713218b88321e5067fd8ad25c3bf7db88c878ccf
[OK] All nodes agree about slots configuration.
>>> Check for open slots...
>>> Check slots coverage...
[OK] All 16384 slots covered.

5.Redis-cluster集群连接

命令: [root@bruce1 7001]# ./bin/redis-cli -h 192.168.221.128 -p 7001 -c
-c：指定是集群连接

[root@bruce1 7001]# ./bin/redis-cli -h 192.168.221.128 -p 7001 -c
192.168.221.128:7001> set username bruceliu123
-> Redirected to slot [14315] located at 192.168.221.128:7003
OK

6.查看集群信息

192.168.221.128:7003> cluster info
cluster_state:ok
cluster_slots_assigned:16384
cluster_slots_ok:16384
cluster_slots_pfail:0
cluster_slots_fail:0
cluster_known_nodes:6
cluster_size:3
cluster_current_epoch:6
cluster_my_epoch:3
cluster_stats_messages_sent:1186
cluster_stats_messages_received:1186

7.查看集群信息

192.168.221.128:7003> cluster nodes
713218b88321e5067fd8ad25c3bf7db88c878ccf 192.168.221.128:7003 myself,master - 0 0 3 connected 10923-16383
e7fb45e74f828b53ccd8b335f3ed587aa115b903 192.168.221.128:7001 master - 0 1498877677276 1 connected 0-5460
b1183545245b3a710a95d669d7bbcbb5e09896a0 192.168.221.128:7006 slave 713218b88321e5067fd8ad25c3bf7db88c878ccf 0 1498877679294 3 connected
8879c2ed9c141de70cb7d5fcb7d690ed8a200792 192.168.221.128:7005 slave 4a312b6fc90bfee187d43588ead99d83b407c892 0 1498877678285 5 connected
4a312b6fc90bfee187d43588ead99d83b407c892 192.168.221.128:7002 master - 0 1498877674248 2 connected 5461-10922
4f8c7455574e2f0aab1e2bb341eae319ac065039 192.168.221.128:7004 slave e7fb45e74f828b53ccd8b335f3ed587aa115b903 0 1498877680308 4 connected

Springboot2.X集成redis集群(Lettuce)连接

1.前言

搭建好redis集群环境，如图:

2.导入依赖

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

3. 配置application.yml

spring:
  # Redis 配置信息
  redis:
    timeout: 6000
    cluster:
      nodes: 172.81.235.217:7001,172.81.235.217:7002,172.81.235.217:7003,172.81.235.217:7004,172.81.235.217:7005,172.81.235.217:7006
    ###################以下为lettuce连接池增加的配置###########################
    lettuce:
      pool:
        max-active:  100 # 连接池最大连接数（使用负值表示没有限制）
        max-idle: 100 # 连接池中的最大空闲连接
        min-idle: 50 # 连接池中的最小空闲连接
        max-wait: 10000 # 连接池最大阻塞等待时间（使用负值表示没有限制

4.新建下面的两个配置类

package com.micro.config;

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.MapPropertySource;
import org.springframework.data.redis.connection.RedisClusterConfiguration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisSentinelConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.connection.lettuce.LettucePoolingClientConfiguration;

import java.util.Set;

/**
 * @BelongsProject: MicroService
 * @BelongsPackage: com.micro.config
 * @CreateTime: 2020-11-25 09:28
 * @Description: TODO
 */
@Configuration
@ConfigurationProperties(prefix = "spring.redis.cluster")
public class RedisClusterConfig {

    private Set<String> nodes;
    private String master;

    @Value("${spring.redis.timeout}")
    private long timeout;
    @Value("${spring.redis.lettuce.pool.max-idle}")
    private int maxIdle;
    @Value("${spring.redis.lettuce.pool.min-idle}")
    private int minIdle;
    @Value("${spring.redis.lettuce.pool.max-wait}")
    private long maxWait;
    @Value("${spring.redis.lettuce.pool.max-active}")
    private int maxActive;

    /**
     * 创建一个Redis连接工厂
     * @return
     */
    @Bean
    public RedisConnectionFactory lettuceConnectionFactory() {

        RedisClusterConfiguration redisClusterConfiguration= new RedisClusterConfiguration(nodes);

        GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();
        genericObjectPoolConfig.setMaxIdle(maxIdle);
        genericObjectPoolConfig.setMinIdle(minIdle);
        genericObjectPoolConfig.setMaxTotal(maxActive);
        genericObjectPoolConfig.setMaxWaitMillis(maxWait);

        LettucePoolingClientConfiguration lettuceClientConfiguration = LettucePoolingClientConfiguration.builder()
                .poolConfig(genericObjectPoolConfig)
                .build();
        return new LettuceConnectionFactory(redisClusterConfiguration, lettuceClientConfiguration);
    }



    public void setNodes(Set<String> nodes) {
        this.nodes = nodes;
    }

    public void setMaster(String master) {
        this.master = master;
    }
}

package com.micro.config;

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;

/**
 * @BelongsProject: MicroService
 * @BelongsPackage: com.micro.config
 * @CreateTime: 2020-11-25 09:33
 * @Description: TODO
 */
@Configuration
public class RedisTemplateConfig {

    @Bean
    public RedisTemplate<String, Object> redisTemplate(LettuceConnectionFactory factory) {
        //创建Json序列化对象
        Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(Object.class);

        //解决查询缓存转换异常的问题
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);

        // 将默认序列化改为Jackson2JsonRedisSerializer序列化
        RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
        template.setKeySerializer(jackson2JsonRedisSerializer);// key序列化
        template.setValueSerializer(jackson2JsonRedisSerializer);// value序列化
        template.setConnectionFactory(factory);
        template.afterPropertiesSet();
        return template;
    }
}

5.执行测试

@SpringBootTest
@RunWith(SpringRunner.class)
public class RedisConfigurationTest {

@Autowired
private RedisTemplate redisTemplate;

	@Test
	public void redisTemplate() throws Exception {
	
	        redisTemplate.opsForValue().set("author", "Damein_xym");
	}

}

redis持久化详解

1.持久化

1.1. 持久化简介

持久化（Persistence），持久化是将程序数据在持久状态和瞬时状态间转换的机制，即把数据（如内存中的对象）保存到可永久保存的存储设备中（如磁盘）。

RDB持久化方式(默认开启)：可以在指定的时间间隔能对数据进行快照存储.

AOF持久化方式(需要开启，优先级高)：记录每次对服务器写的操作,AOF命令以redis协议追加保存每次写的操作到文件末尾.当服务器重启的时候会重新执行这些命令来恢复原始的数据，Redis还能对AOF文件进行后台重写,使得AOF文件的体积不至于过大.

如果服务器开启了AOF持久化功能。服务器会优先使用AOF文件还原数据。只有关闭了AOF持久化功能，服务器才会使用RDB文件还原数据.

2. RDB持久化

RDB文件是一个经过压缩的二进制文件（默认的文件名：dump.rdb）

2.1. RDB文件持久化创建与载入

在 Redis持久化时， RDB 程序将当前内存中的数据库状态保存到磁盘文件中，在 Redis 重启动时， RDB 程序可以通过载入 RDB 文件来还原数据库的状态。

2.2. 触发条件

RDB持久化的触发分为手动触发和自动触发两种。

2.2.1.手动触发

save命令和bgsave命令都可以生成RDB文件。

SAVE
同步操作，在执行该命令时，服务器会被阻塞，拒绝客户端发送的命令请求

redis> save

save命令会阻塞Redis服务器进程，直到RDB文件创建完毕为止，在Redis服务器阻塞期间，服务器不能处理任何命令请求。

BGSAVE

异步操作，在执行该命令时，子进程执行保存工作，服务器还可以继续让主线程处理客户端发送的命令请求

redis>bgsave

而bgsave命令会创建一个子进程，由子进程来负责创建RDB文件，父进程(即Redis主进程则继续处理请求。

bgsave命令执行过程中，只有fork子进程时会阻塞服务器，而对于save命令，整个过程都会阻塞服务器，因此save已基本被废弃，线上环境要杜绝save的使用。

2.2.2.自动触发

在自动触发RDB持久化时，Redis也会选择bgsave而不是save来进行持久化。

自动触发最常见的情况是在配置文件中通过save m n，指定当m秒内发生n次变化时，会触发bgsave。

比如：
/*服务器在900秒之内，对数据库进行了至少1次修改*/
Save  900   1
/*服务器在300秒之内，对数据库进行了至少10次修改*/
Save  300   10
/*服务器在60秒之内，对数据库进行了至少10000次修改*/
Save  60     10000

只要满足其中一个条件就会执行BGSAVE命令

2.2.3.其他自动触发机制

除了save m n 以外，还有一些其他情况会触发bgsave：

●在主从复制场景下，如果从节点执行全量复制操作，则主节点会执行bgsave命令，并将rdb文件发送给从节点。
●执行shutdown命令时，自动执行rdb持久化。

2.3.RDB 默认配置

################################ SNAPSHOTTING  ################################
#
# Save the DB on disk:
#在给定的秒数和给定的对数据库的写操作数下，自动持久化操作。
#   save <seconds> <changes>
# 
save 900 1
save 300 10
save 60 10000

#bgsave发生错误时是否停止写入，一般为yes
stop-writes-on-bgsave-error yes

#持久化时是否使用LZF压缩字符串对象?
rdbcompression yes

#是否对rdb文件进行校验和检验，通常为yes
rdbchecksum yes

# RDB持久化文件名
dbfilename dump.rdb

#持久化文件存储目录
dir ./

3.AOF持久化

3.1.AOF持久化简介

AOF持久化是通过保存Redis服务器所执行的写命令来记录数据库状态.

RDB持久化是将进程数据写入文件，而AOF持久化，则是将Redis执行的每次写、删除命令记录到单独的日志文件中，查询操作不会记录；当Redis重启时再次执行AOF文件中的命令来恢复数据。

与RDB相比，AOF的实时性更好，因此已成为主流的持久化方案。

3.2.AOF持久化功能实现

append命令追加：当AOF持久化功能处于打开状态时，服务器执行完一个写命令会协议格式被执行的命令追加服务器状态的aof_buf缓冲区的末尾。

 reids>SET KET VAULE
 //协议格式
 \r\n$3\r\nSET\r\n$3\r\nKEY\r\n$5\r\nVAULE\r\n

文件写入和同步sync：当执行了append命令追加后，服务器会调用flushAppendOnlyFile函数是否需要将AOF缓冲区的内容写入和保存到AOF文件

3.3.AOF持久化策略

AOF持久化策略(即缓冲区内容写入和同步sync到AOF中)，可以通过配置appendfsync属性来选择AOF持久化策略：

always：将aof_buf缓冲区中的所有内容写入并同步到AOF文件，每次有新命令追加到 AOF 文件时就执行一次 fsync。

everysec（默认）：如果上次同步AOF的时间距离现在超过一秒，先将aof_buf缓冲区中的所有内容写入到AOF文件，再次对AOF文件进行同步，且同步操作由一个专门线程负责执行。

no：将aof_buf缓冲区中的所有内容写入到AOF文件，但并不对AOF文件进行同步，何时同步由操作系统(OS)决定。

AOF持久化策略的效率与安全性：

Always:效率最慢的，但安全性是最安全的，即使出现故障宕机，持久化也只会丢失一个事件循环的命令数据
everysec：兼顾速度和安全性，出现宕机也只是丢失一秒钟的命令数据. 每秒钟强制写入磁盘一次，在性能和持久化方面做了很好的折中
No:写入最快，但综合起来单次同步是时间是最长的，且出现宕机时会丢失上传同步AOF文件之后的所有命令数据。完全依赖os，性能最好，持久化没保证。

4.AOF重写

4.1.Why

AOF（append only file）对Redis进行持久化是通过保存被执行的写命令来记录数据库状态的，随着服务器运行，AOF文件内容越来越多，载入AOF文件的时间会越来越长，影响Redis服务。

所以有必要对’冗余‘的AOF文件进行优化，即AOF文件重写。

那为什么要进行AOF后台重写？因为Redis单线程特性，AOF重写操作会引入大量写操作，引起stop the world，所以fork出子进程执行AOF后台重写，这样父进程可以继续处理命令请求。

4.2.AOF重写原理

怎么重写？是对旧的AOF文件进行读取、分析后进行优化吗？Redis没有采用，Redis是通过读取服务器当前数据库状态实现该功能的。
举个例子：
对一个animals键执行以下命令：

SADD animals "Cat"
SADD animals "Dog" "Tiger"
SREM animals "Tiger"

为了保存上述状态，AOF需要进行3次命令保存。
AOF重写后只需要读取animals键的所有值，再用SADD animals “Cat” "Dog"一条命令即可。
从3条减少为1条，减少了文件体积。

4.3.AOF后台重写原理（BGREWRITEAOF命令）

使用子进程（而不是开启一个线程）进行AOF重写虽然可以避免使用锁的情况下，保证数据安全性，但是会带来子进程和父进程一致性问题。

例如在开始重写之后父进程又接收了新的键值对此时子进程是无法知晓的，当子进程重写完成后的数据库和父进程的数据库状态是不一致的。

见下表：

在T7时刻服务器进程有了4个键，而子进程只有1个键，即所谓的漏追加。

为了解决这种不一致性，redis设置了一个AOF重写缓冲区。

在子进程执行AOF重写期间。服务器进程需要执行以下3个动作：

1.执行客户端命令
2.执行后追加到AOF缓冲区
3.执行后追加到AOF重写缓冲区

执行完上述3个操作后可以保证：

- AOF缓冲区内容会定期被写入和同步到AOF文件，对现有AOF文件处理正常进行
- 子进程开始，服务器执行的所有写命令都会进入AOF重写缓冲区

子进程完成AOF重写后，它向父进程发送一个信号，父进程收到信号后会调用一个信号处理函数，该函数把AOF重写缓冲区的命令追加到新AOF文件中然后替换掉现有AOF文件。父进程处理完毕后可以继续接受客户端命令调用，可以看出在AOF后台重写过程中只有这个信号处理函数会阻塞服务器进程。
下表是完整的AOF后台重写过程：

T8 T9执行的任务会阻塞服务器处理命令。

5.AOF持久化默认参数

############################## APPEND ONLY MODE ###############################

#开启AOF持久化方式
appendonly no

#AOF持久化文件名
appendfilename "appendonly.aof"
    
#每秒把缓冲区的数据fsync到磁盘
appendfsync everysec
# appendfsync no

#是否在执行重写时不同步数据到AOF文件
no-appendfsync-on-rewrite no

# 触发AOF文件执行重写的增长率  aof文件大小比起上次重写时的大小,增长率100%时,重写.
auto-aof-rewrite-percentage 100
    
#触发AOF文件执行重写的最小size
auto-aof-rewrite-min-size 64mb

#redis在恢复时，会忽略最后一条可能存在问题的指令
aof-load-truncated yes

#正在到处rdb快照的过程中,要不要停止同步aof
no-appendfsync-on-rewrite yes: 

#是否打开混合开关  Redis 4.0.0 或以上
aof-use-rdb-preamble yes

手动执行重写的命令是:

执行重写可以在登录状态下执行,直接输入bgrewriteaof.

6.持久化方式总结与抉择

6.1.RDB的优点

RDB是一个非常紧凑的文件,它保存了某个时间点得数据集,非常适用于数据集的备份,比如你可以在每个小时报保存一下过去24小时内的数据,同时每天保存过去30天的数据,这样即使出了问题你也可以根据需求恢复到不同版本的数据集.
基于RDB文件紧凑性,便于复制数据到一个远端数据中心，非常适用于灾难恢复.

RDB在保存RDB文件时父进程唯一需要做的就是fork出一个子进程,接下来的工作全部由子进程来做，父进程不需要再做其他IO操作，所以RDB持久化方式可以最大化redis的性能.
与AOF相比,在恢复大的数据集的时候，RDB方式会更快一些.

6.2.RDB的缺点

如果你希望在redis意外停止工作（例如电源中断）的情况下丢失的数据最少的话，那么RDB不适合你.虽然你可以配置不同的save时间点(例如每隔5分钟并且对数据集有100个写的操作),是Redis要完整的保存整个数据集是一个比较繁重的工作,你通常会每隔5分钟或者更久做一次完整的保存,万一在Redis意外宕机,你可能会丢失几分钟的数据.

RDB 需要经常fork子进程来保存数据集到硬盘上,当数据集比较大的时候,fork的过程是非常耗时的,可能会导致Redis在一些毫秒级内不能响应客户端的请求.如果数据集巨大并且CPU性能不是很好的情况下,这种情况会持续1秒,AOF也需要fork,但是你可以调节重写日志文件的频率来提高数据集的耐久度.

6.3.AOF优点

（1）AOF可以更好的保护数据不丢失，一般AOF会每隔1秒，通过一个后台线程执行一次fsync操作，最多丢失1秒钟的数据。

（2）AOF日志文件没有任何磁盘寻址的开销，写入性能非常高，文件不容易破损。

（3）AOF日志文件即使过大的时候，出现后台重写操作，也不会影响客户端的读写。

（4）AOF日志文件的命令通过非常可读的方式进行记录，这个特性非常适合做灾难性的误删除的紧急恢复。比如某人不小心用flushall命令清空了所有数据，只要这个时候后台rewrite还没有发生，那么就可以立即拷贝AOF文件，将最后一条flushall命令给删了，然后再将该AOF文件放回去，就可以通过恢复机制，自动恢复所有数据

6.4.AOF缺点

（1）对于同一份数据来说，AOF日志文件通常比RDB数据快照文件更大

（2）AOF开启后，支持的写QPS会比RDB支持的写QPS低，因为AOF一般会配置成每秒fsync一次日志文件，当然，每秒一次fsync，性能也还是很高的

7.如何选择使用哪种持久化方式？

一般来说，选择的话，两者加一起才更好.

其他问题说明
注:在dump.rdb过程中,aof如果停止同步,会不会丢失?
答:不会,所有的操作缓存在内存的队列里,dump完成后统一操作.

注:aof重写指得是什么
答:aof重写是指把内存中的数据逆化成命令,写入到.aof日志里,已解决aof日志过大的问题.

注:如果rdb文件,和aof文件都存在,优先用谁来恢复数据?
答:aof

注:两种是否可以同时用?
答:可以,而且推荐这样做.

注:恢复时 rdb和aof哪个恢复的快?
答:rdb快,因为其是数据的内存映射,直接载入到内存,二aof是命令需要逐条执行.

注意点:如果两种持久化方式都开启,则以aof为准,虽然快照方式恢复速度快,但是最终被aof给覆盖,所以两种方式都开启时,以aof为准.

过期删除策略和内存淘汰策略

1.引言

我们先来看如下几个问题：
①、如何设置Redis键的过期时间？
②、设置完一个键的过期时间后，到了这个时间，这个键还能获取到么？假如获取不到那这个键还占据着内存吗？
③、如何设置Redis的内存大小？当内存满了之后，Redis有哪些内存淘汰策略？我们又该如何选择？

如果上面的几个问题你都懂，那么下面的内容你就不用看了；如果你不是很懂，那就带着这些问题往下看。

2.设置Redis键过期时间

Redis提供了四个命令来设置过期时间（生存时间）。

　①、EXPIRE <key> <ttl> ：表示将键 key 的生存时间设置为 ttl 秒。
　②、PEXPIRE <key> <ttl> ：表示将键 key 的生存时间设置为 ttl 毫秒。
　
　③、EXPIREAT <key> <timestamp> ：表示将键 key 的生存时间设置为 timestamp 所指定的秒数时间戳。
　④、PEXPIREAT <key> <timestamp> ：表示将键 key 的生存时间设置为 timestamp 所指定的毫秒数时间戳。

在Redis内部实现中，前面三个设置过期时间的命令最后都会转换成最后一个PEXPIREAT 命令来完成。

另外补充两个知识点：

2.1.移除键的过期时间

　　PERSIST <key> ：表示将key的过期时间移除。

2.2.返回键的剩余生存时间

　　TTL <key> ：以秒的单位返回键 key 的剩余生存时间。

　　PTTL <key> ：以毫秒的单位返回键 key 的剩余生存时间。

3.Redis过期时间的判定

在Redis内部，每当我们设置一个键的过期时间时，Redis就会将该键带上过期时间存放到一个过期字典中。当我们查询一个键时，Redis便首先检查该键是否存在过期字典中，如果存在，那就获取其过期时间。然后将过期时间和当前系统时间进行比对，比系统时间大，那就没有过期；反之判定该键过期。

4.过期删除策略

通常删除某个key，我们有如下三种方式进行处理。

①、定时删除

在设置某个key 的过期时间同时，我们创建一个定时器，让定时器在该过期时间到来时，立即执行对其进行删除的操作。

优点：定时删除对内存是最友好的，能够保存内存的key一旦过期就能立即从内存中删除。

缺点：对CPU最不友好，在过期键比较多的时候，删除过期键会占用一部分 CPU 时间，对服务器的响应时间和吞吐量造成影响。

②、惰性删除
设置该key 过期时间后，我们不去管它，当需要该key时，我们在检查其是否过期，如果过期，我们就删掉它，反之返回该key。

优点：对 CPU友好，我们只会在使用该键时才会进行过期检查，对于很多用不到的key不用浪费时间进行过期检查。

缺点：对内存不友好，如果一个键已经过期，但是一直没有使用，那么该键就会一直存在内存中，如果数据库中有很多这种使用不到的过期键，这些键便永远不会被删除，内存永远不会释放。从而造成内存泄漏。

③、定期删除
每隔一段时间，我们就对一些key进行检查，删除里面过期的key。

优点：可以通过限制删除操作执行的时长和频率来减少删除操作对 CPU 的影响。另外定期删除，也能有效释放过期键占用的内存。

缺点：难以确定删除操作执行的时长和频率。

如果执行的太频繁，定期删除策略变得和定时删除策略一样，对CPU不友好。

如果执行的太少，那又和惰性删除一样了，过期键占用的内存不会及时得到释放。

另外最重要的是，在获取某个键时，如果某个键的过期时间已经到了，但是还没执行定期删除，那么就会返回这个键的值，这是业务不能忍受的错误。

5.Redis过期删除策略

前面讨论了删除过期键的三种策略，发现单一使用某一策略都不能满足实际需求，聪明的你可能想到了，既然单一策略不能满足，那就组合来使用吧。

没错，Redis的过期删除策略就是：惰性删除和定期删除两种策略配合使用。

惰性删除：Redis的惰性删除策略由 db.c/expireIfNeeded 函数实现，所有键读写命令执行之前都会调用 expireIfNeeded 函数对其进行检查，如果过期，则删除该键，然后执行键不存在的操作；未过期则不作操作，继续执行原有的命令。

定期删除：由redis.c/activeExpireCycle 函数实现，函数以一定的频率运行，每次运行时，都从一定数量的数据库中取出一定数量的随机键进行检查，并删除其中的过期键。

注意：并不是一次运行就检查所有的库，所有的键，而是随机检查一定数量的键。

定期删除函数的运行频率，在Redis2.6版本中，规定每秒运行10次。在Redis2.8版本后，可以通过修改配置文件redis.conf 的 hz 选项来调整这个次数。

看上面对这个参数的解释，建议不要将这个值设置超过 100，否则会对CPU造成比较大的压力。

我们看到，通过过期删除策略，对于某些永远使用不到的键，并且多次定期删除也没选定到并删除，那么这些键同样会一直驻留在内存中，又或者在Redis中存入了大量的键，这些操作可能会导致Redis内存不够用，这时候就需要Redis的内存淘汰策略了。

6.内存淘汰策略

①、设置Redis最大内存
　　在配置文件redis.conf 中，可以通过参数 maxmemory <bytes> 来设定最大内存：

不设定该参数默认是无限制的，但是通常会设定其为物理内存的四分之三。

②、设置内存淘汰方式
　　当现有内存大于 maxmemory 时，便会触发redis主动淘汰内存方式，通过设置 maxmemory-policy ，有如下几种淘汰方式：

　　1）volatile-lru   利用LRU算法移除设置过过期时间的key (LRU:最近使用 Least Recently Used ) 。
　　2）allkeys-lru   利用LRU算法移除任何key （和上一个相比，删除的key包括设置过期时间和不设置过期时间的）。通常使用该方式。
　　3）volatile-random 移除设置过过期时间的随机key 。
　　4）allkeys-random  无差别的随机移除。
　　5）volatile-ttl   移除即将过期的key(minor TTL) 
　　6）noeviction 不移除任何key，只是返回一个写错误 ，默认选项，一般不会选用。

在redis.conf 配置文件中，可以设置淘汰方式：

7.总结

通过上面的介绍，相信大家对Redis的过期数据删除策略和内存淘汰策略有一定的了解了。这里总结一下：

Redis过期删除策略是采用惰性删除和定期删除这两种方式组合进行的，惰性删除能够保证过期的数据我们在获取时一定获取不到，而定期删除设置合适的频率，则可以保证无效的数据及时得到释放，而不会一直占用内存数据。

但是我们说Redis是部署在物理机上的，内存不可能无限扩充的，当内存达到我们设定的界限后，便自动触发Redis内存淘汰策略，而具体的策略方式要根据实际业务情况进行选取。

缓存穿透、缓存击穿、缓存雪崩

我们来介绍Redis使用过程中需要注意的三种问题：缓存穿透、缓存击穿、缓存雪崩。

1.缓存穿透

1.1.概念

缓存穿透：缓存和数据库中都没有的数据，可用户还是源源不断的发起请求，导致每次请求都会到数据库，从而压垮数据库。

如下图的流程：

比如客户查询一个根本不存在的东西，首先从Redis中查不到，然后会去数据库中查询，数据库中也查询不到，那么就不会将数据放入到缓存中，后面如果还有类似源源不断的请求，最后都会压到数据库来处理，从而给数据库造成巨大的压力。

1.2.解决办法

①、业务层校验

用户发过来的请求，根据请求参数进行校验，对于明显错误的参数，直接拦截返回。

比如，请求参数为主键自增id，那么对于请求小于0的id参数，明显不符合，可以直接返回错误请求。

②、不存在数据设置短过期时间

对于某个查询为空的数据，可以将这个空结果进行Redis缓存，但是设置很短的过期时间，比如30s，可以根据实际业务设定。注意一定不要影响正常业务。

③、布隆过滤器

关于布隆过滤器，后面会详细介绍。布隆过滤器是一种数据结构，利用极小的内存，可以判断大量的数据“一定不存在或者可能存在”。

对于缓存穿透，我们可以将查询的数据条件都哈希到一个足够大的布隆过滤器中，用户发送的请求会先被布隆过滤器拦截，一定不存在的数据就直接拦截返回了，从而避免下一步对数据库的压力。

2、缓存击穿

2.1.概念

缓存击穿：Redis中一个热点key在失效的同时，大量的请求过来，从而会全部到达数据库，压垮数据库。

这里要注意的是这是某一个热点key过期失效，和后面介绍缓存雪崩是有区别的。比如淘宝双十一，对于某个特价热门的商品信息，缓存在Redis中，刚好0点，这个商品信息在Redis中过期查不到了，这时候大量的用户又同时正好访问这个商品，就会造成大量的请求同时到达数据库。

2.2.解决办法

①、设置热点数据永不过期

对于某个需要频繁获取的信息，缓存在Redis中，并设置其永不过期。当然这种方式比较粗暴，对于某些业务场景是不适合的。

②、定时更新

比如这个热点数据的过期时间是1h，那么每到59minutes时，通过定时任务去更新这个热点key，并重新设置其过期时间。

③、互斥锁

互斥锁简单来说就是在Redis中根据key获得的value值为空时，先锁上，然后从数据库加载，加载完毕，释放锁。若其他线程也在请求该key时，发现获取锁失败，则睡眠一段时间（比如100ms）后重试。

3、缓存雪崩

3.1.概念

缓存雪崩：Redis中缓存的数据大面积同时失效，或者Redis宕机，从而会导致大量请求直接到数据库，压垮数据库。

对于一个业务系统，如果Redis宕机或大面积的key同时过期，会导致大量请求同时打到数据库，这是灾难性的问题。

3.2.解决办法

①、设置有效期均匀分布

避免缓存设置相近的有效期，我们可以在设置有效期时增加随机值；

或者统一规划有效期，使得过期时间均匀分布。

②、数据预热

对于即将来临的大量请求，我们可以提前走一遍系统，将数据提前缓存在Redis中，并设置不同的过期时间。

③、保证Redis服务高可用

前面我们介绍过Redis的哨兵模式和集群模式，为防止Redis集群单节点故障，可以通过这两种模式实现高可用。+

Redis布隆过滤器

1.布隆过滤器使用场景

比如有如下几个需求：

①、原本有10亿个号码，现在又来了10万个号码，要快速准确判断这10万个号码是否在10亿个号码库中？

解决办法一：将10亿个号码存入数据库中，进行数据库查询，准确性有了，但是速度会比较慢。

解决办法二：将10亿号码放入内存中，比如Redis缓存中，这里我们算一下占用内存大小：10亿*8字节=8GB，通过内存查询，准确性和速度都有了，但是大约8gb的内存空间，挺浪费内存空间的。

②、接触过爬虫的，应该有这么一个需求，需要爬虫的网站千千万万，对于一个新的网站url，我们如何判断这个url我们是否已经爬过了？

解决办法还是上面的两种，很显然，都不太好。

③、同理还有垃圾邮箱的过滤。

那么对于类似这种，大数据量集合，如何准确快速的判断某个数据是否在大数据量集合中，并且不占用内存，布隆过滤器应运而生了。

2.布隆过滤器简介

带着上面的几个疑问，我们来看看到底什么是布隆过滤器。

布隆过滤器：一种数据结构，是由一串很长的二进制向量组成，可以将其看成一个二进制数组。既然是二进制，那么里面存放的不是0，就是1，但是初始默认值都是0。

如下所示：

①、添加数据

介绍概念的时候，我们说可以将布隆过滤器看成一个容器，那么如何向布隆过滤器中添加一个数据呢？

如下图所示：当要向布隆过滤器中添加一个元素key时，我们通过多个hash函数，算出一个值，然后将这个值所在的方格置为1。

比如，下图hash1(key)=1，那么在第2个格子将0变为1（数组是从0开始计数的），hash2(key)=7，那么将第8个格子置位1，依次类推。

②、判断数据是否存在？

知道了如何向布隆过滤器中添加一个数据，那么新来一个数据，我们如何判断其是否存在于这个布隆过滤器中呢？

很简单，我们只需要将这个新的数据通过上面自定义的几个哈希函数，分别算出各个值，然后看其对应的地方是否都是1，如果存在一个不是1的情况，那么我们可以说，该新数据一定不存在于这个布隆过滤器中。

反过来说，如果通过哈希函数算出来的值，对应的地方都是1，那么我们能够肯定的得出：这个数据一定存在于这个布隆过滤器中吗？

答案是否定的，因为多个不同的数据通过hash函数算出来的结果是会有重复的，所以会存在某个位置是别的数据通过hash函数置为的1。

我们可以得到一个结论：布隆过滤器可以判断某个数据一定不存在，但是无法判断一定存在。

③、布隆过滤器优缺点

优点：优点很明显，二进制组成的数组，占用内存极少，并且插入和查询速度都足够快。

缺点：随着数据的增加，误判率会增加；还有无法判断数据一定存在；另外还有一个重要缺点，无法删除数据。

3.Redis实现布隆过滤器

①、bitmaps
　　我们知道计算机是以二进制位作为底层存储的基础单位，一个字节等于8位。

比如“big”字符串是由三个字符组成的，这三个字符对应的ASCII码分为是98、105、103，对应的二进制存储如下：

在Redis中，Bitmaps 提供了一套命令用来操作类似上面字符串中的每一个位。

3.1.设置值

setbit key offset value

　我们知道”b”的二进制表示为0110 0010，我们将第7位（从0开始）设置为1，那0110 0011 表示的就是字符“c”，所以最后的字符 “big”变成了“cig”。

3.2.获取值

gitbit key offset

3.3. 获取位图指定范围值为1的个数

bitcount key [start end]

如果不指定，那就是获取全部值为1的个数。

注意：start和end指定的是字节的个数，而不是位数组下标。

4.Redisson

Redis 实现布隆过滤器的底层就是通过 bitmap 这种数据结构，至于如何实现，这里就不重复造轮子了，介绍业界比较好用的一个客户端工具——Redisson。

Redisson 是用于在 Java 程序中操作 Redis 的库，利用Redisson 我们可以在程序中轻松地使用 Redis。

下面我们就通过 Redisson 来构造布隆过滤器。

下面使用Redisson对布隆过滤器进行的封装。
1、引入maven依赖

 <!-- https://mvnrepository.com/artifact/org.redisson/redisson -->
    <dependency>
        <groupId>org.redisson</groupId>
        <artifactId>redisson</artifactId>
        <version>3.11.2</version>
    </dependency>

import org.junit.Test;
import org.redisson.Redisson;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;


public class TestRedissonBloomFilter {

    @Test
    public void test1(){

        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        //config.useSingleServer().setPassword("123");
        //构造Redisson
        RedissonClient redisson = Redisson.create(config);

        RBloomFilter<String> bloomFilter = redisson.getBloomFilter("phoneList");
        //初始化布隆过滤器：预计元素为100000000L,误差率为3%
        bloomFilter.tryInit(100000000L,0.03);
        
        //将号码10086插入到布隆过滤器中
        bloomFilter.add("10086");
        bloomFilter.add("10083");
        bloomFilter.add("10081");
        bloomFilter.add("10089");

        //判断下面号码是否在布隆过滤器中
        System.out.println(bloomFilter.contains("123456"));//false
        System.out.println(bloomFilter.contains("10086"));//true
    }

}

这是单节点的Redis实现方式，如果数据量比较大，期望的误差率又很低，那单节点所提供的内存是无法满足的，这时候可以使用分布式布隆过滤器，同样也可以用 Redisson 来实现。

数据库和缓存双写一致性

数据库(Mysql )和缓存（比如：Redis）双写数据一致性问题，是一个跟开发语言无关的公共问题。尤其在高并发的场景下，这个问题变得更加严重。

该问题无论在面试，还是工作中遇到的概率非常大！！！！

1.常见方案

通常情况下，我们使用缓存的主要目的是为了提升查询的性能。大多数情况下，我们是这样使用缓存的：

用户请求过来之后，先查缓存有没有数据，如果有则直接返回。
如果缓存没数据，再继续查数据库。
如果数据库有数据，则将查询出来的数据，放入缓存中，然后返回该数据。
如果数据库也没数据，则直接返回空

这是缓存非常常见的用法。一眼看上去，好像没有啥问题。

但你忽略了一个非常重要的细节：如果数据库中的某条数据，放入缓存之后，数据库又被更新了，那么该如何更新缓存(缓存如何同步)呢？

不更新缓存行不行？

答：当然不行，如果不更新缓存，在很长的一段时间内（决定于缓存的过期时间），用户请求从缓存中获取到的都可能是旧值，而非数据库的最新值。这不是有数据不一致的问题？

那么，我们该如何更新缓存呢？

目前有以下4种方案：

先写缓存，再写数据库
先写数据库，再写缓存
先删缓存，再写数据库
先写数据库，再删缓存

接下来，我们详细说说这4种方案。

2.先写缓存，再写数据库

对于更新缓存的方案，很多人第一个想到的可能是在写操作中直接更新缓存（写缓存），更直接明了。

那么，问题来了：在写操作中，到底是先写缓存，还是先写数据库呢？

我们在这里先聊聊先写缓存，再写数据库的情况，因为它的问题最严重。

某一个用户的每一次写操作，如果刚写完缓存，突然网络出现了异常，导致写数据库失败了。

其结果是缓存更新成了最新数据，但数据库没有，这样缓存中的数据不就变成脏数据了？如果此时该用户的查询请求，正好读取到该数据，就会出现问题，因为该数据在数据库中根本不存在，这个问题非常严重。

我们都知道，缓存的主要目的是把数据库的数据临时保存在内存，便于后续的查询，提升查询速度。

但如果某条数据，在数据库中都不存在，你缓存这种“假数据”又有啥意义呢？

因此，先写缓存，再写数据库的方案是不可取的，在实际工作中用得不多。

3.先写数据库，再写缓存

既然上面的方案行不通，接下来，聊聊先写数据库，再写缓存的方案，该方案在低并发编程中有人在用。

用户的写操作，先写数据库，再写缓存，可以避免之前“假数据”的问题。但它却带来了新的问题。

什么问题呢？

3.1.写缓存失败了

如果把写数据库和写缓存操作，放在同一个事务当中，当写缓存失败了，我们可以把写入数据库的数据进行回滚。

如果是并发量比较小，对接口性能要求不太高的系统，可以这么玩。

但如果在高并发的业务场景中，写数据库和写缓存，都属于远程操作。为了防止出现大事务，造成的死锁问题，通常建议写数据库和写缓存不要放在同一个事务中。

也就是说在该方案中，如果写数据库成功了，但写缓存失败了，数据库中已写入的数据不会回滚。

这就会出现：数据库是新数据，而缓存是旧数据，两边数据不一致的情况。

3.2.高并发下的问题

假设在高并发的场景中，针对同一个用户的同一条数据，有两个写数据请求：a和b，它们同时请求到业务系统。

其中请求a获取的是旧数据，而请求b获取的是新数据，如下图所示：

请求a先过来，刚写完了数据库。但由于网络原因，卡顿了一下，还没来得及写缓存。
这时候请求b过来了，先写了数据库。
接下来，请求b顺利写了缓存。
此时，请求a卡顿结束，也写了缓存。
很显然，在这个过程当中，请求b在缓存中的新数据，被请求a的旧数据覆盖了。

也就是说：在高并发场景中，如果多个线程同时执行先写数据库，再写缓存的操作，可能会出现数据库是新值，而缓存中是旧值，两边数据不一致的情况。

3.2 浪费系统资源

该方案还有一个比较大的问题就是：每个写操作，写完数据库，会马上写缓存，比较浪费系统资源。

为什么这么说呢？

你可以试想一下，如果写的缓存，并不是简单的数据内容，而是要经过非常复杂的计算得出的最终结果。这样每写一次缓存，都需要经过一次非常复杂的计算，不是非常浪费系统资源吗？

尤其是cpu和内存资源。

还有些业务场景比较特殊：写多读少。

如果在这类业务场景中，每个用的写操作，都需要写一次缓存，有点得不偿失。

由此可见，在高并发的场景中，先写数据库，再写缓存，这套方案问题挺多的，也不太建议使用。

如果你已经用了，赶紧看看踩坑了没？

4.先删缓存，再写数据库

通过上面的内容我们得知，如果直接更新缓存的问题很多。

那么，为何我们不能换一种思路：不去直接更新缓存，而改为删除缓存呢？

删除缓存方案，同样有两种：

先删缓存，再写数据库
先写数据库，再删缓存

我们一起先看看：先删缓存，再写数据库的情况。

说白了，在用户的写操作中，先执行删除缓存操作，再去写数据库。这套方案，可以是可以，但也会有一样问题。

4.1 高并发下的问题

假设在高并发的场景中，同一个用户的同一条数据，有一个读数据请求c，还有另一个写数据请求d（一个更新操作），同时请求到业务系统。如下图所示：

请求d先过来，把缓存删除了。但由于网络原因，卡顿了一下，还没来得及写数据库。
这时请求c过来了，先查缓存发现没数据，再查数据库，有数据，但是旧值。
请求c将数据库中的旧值，更新到缓存中。
此时，请求d卡顿结束，把新值写入数据库。

在这个过程当中，请求d的新值并没有被请求c写入缓存，同样会导致缓存和数据库的数据不一致的情况。更正：图中步骤7写入旧值，步骤9要删掉。

那么，这种场景的数据不一致问题，能否解决呢？

4.2 缓存延迟双删

在上面的业务场景中，一个读数据请求，一个写数据请求。当写数据请求把缓存删了之后，读数据请求，可能把当时从数据库查询出来的旧值，写入缓存当中。

有人说还不好办，请求d在写完数据库之后，把缓存重新删一次不就行了？

这就是我们所说的缓存双删，即在写数据库之前删除一次，写完数据库后，再删除一次。

该方案有个非常关键的地方是：

第二次删除缓存，并非立马就删，而是要在一定的时间间隔之后。

我们再重新回顾一下，高并发下一个读数据请求，一个写数据请求导致数据不一致的产生过程：

请求d先过来，把缓存删除了。但由于网络原因，卡顿了一下，还没来得及写数据库。
这时请求c过来了，先查缓存发现没数据，再查数据库，有数据，但是旧值。
请求c将数据库中的旧值，更新到缓存中。
此时，请求d卡顿结束，把新值写入数据库。

一段时间之后，比如：500ms，请求d将缓存删除。这样来看确实可以解决缓存不一致问题。

那么，为什么一定要间隔一段时间之后，才能删除缓存呢？

请求d卡顿结束，把新值写入数据库后，请求c将数据库中的旧值，更新到缓存中。

此时，如果请求d删除太快，在请求c将数据库中的旧值更新到缓存之前，就已经把缓存删除了，这次删除就没任何意义。必须要在请求c更新缓存之后，再删除缓存，才能把旧值及时删除了。

所以需要在请求d中加一个时间间隔，确保请求c，或者类似于请求c的其他请求，如果在缓存中设置了旧值，最终都能够被请求d删除掉。

接下来，还有一个问题：如果第二次删除缓存时，删除失败了该怎么办？

5. 先写数据库，再删缓存

从前面得知，先删缓存，再写数据库，在并发的情况下，也可能会出现缓存和数据库的数据不一致的情况。

那么，我们只能寄希望于最后的方案了。

接下来，我们重点看看先写数据库，再删缓存的方案。

在高并发的场景中，有一个读数据请求，有一个写数据请求，更新过程如下：

请求e先写数据库，由于网络原因卡顿了一下，没有来得及删除缓存。
请求f查询缓存，发现缓存中有数据，直接返回该数据。
请求e删除缓存。

在这个过程中，只有请求f读了一次旧数据，后来旧数据被请求e及时删除了，看起来问题不大。

但如果是读数据请求先过来呢？

请求f查询缓存，发现缓存中有数据，直接返回该数据。
请求e先写数据库。
请求e删除缓存。

这种情况看起来也没问题呀？

答：对的。

但就怕出现下面这种情况，即缓存自己失效了。如下图所示：

缓存过期时间到了，自动失效。
请求f查询缓存，发缓存中没有数据，查询数据库的旧值，但由于网络原因卡顿了，没有来得及更新缓存。
请求e先写数据库，接着删除了缓存。
请求f更新旧值到缓存中。

这时，缓存和数据库的数据同样出现不一致的情况了。

但这种情况还是比较少的，需要同时满足以下条件才可以：

缓存刚好自动失效。
请求f从数据库查出旧值，更新缓存的耗时，比请求e写数据库，并且删除缓存的还长。

我们都知道查询数据库的速度，一般比写数据库要快，更何况写完数据库，还要删除缓存。所以绝大多数情况下，写数据请求比读数据情况耗时更长。

由此可见，系统同时满足上述两个条件的概率非常小。

推荐大家使用先写数据库，再删缓存的方案，虽说不能100%避免数据不一致问题，但出现该问题的概率，相对于其他方案来说是最小的。

但在该方案中，如果删除缓存失败了该怎么办呢？

6.删缓存失败怎么办？

其实先写数据库，再删缓存的方案，跟缓存双删的方案一样，有一个共同的风险点，即：如果缓存删除失败了，也会导致缓存和数据库的数据不一致。

那么，删除缓存失败怎么办呢？

答：需要加重试机制。

在接口中如果更新了数据库成功了，但更新缓存失败了，可以立刻重试3次。如果其中有任何一次成功，则直接返回成功。如果3次都失败了，则写入数据库，准备后续再处理。

当然，如果你在接口中直接同步重试，该接口并发量比较高的时候，可能有点影响接口性能。

这时，就需要改成异步重试了。

异步重试方式有很多种，比如：

每次都单独起一个线程，该线程专门做重试的工作。但如果在高并发的场景下，可能会创建太多的线程，导致系统OOM问题，不太建议使用。
将重试的任务交给线程池处理，但如果服务器重启，部分数据可能会丢失。
将重试数据写表，然后使用elastic-job、xxl-Job、Quartz等定时任务进行重试。
将重试的请求写入mq等消息中间件中，在mq的consumer中处理。
订阅mysql的binlog，在订阅者中，如果发现了更新数据请求，则删除相应的缓存。

7.定时任务

使用定时任务重试的具体方案如下：

当用户操作写完数据库，但删除缓存失败了，需要将用户数据写入重试表中。如下图所示：
在定时任务中，异步读取重试表中的用户数据。重试表需要记录一个重试次数字段，初始值为0。然后重试5次，不断删除缓存，每重试一次该字段值+1。如果其中有任意一次成功了，则返回成功。如果重试了5次，还是失败，则我们需要在重试表中记录一个失败的状态，等待后续进一步处理。

在高并发场景中，定时任务推荐使用elastic-job。相对于xxl-job等定时任务，它可以分片处理，提升处理速度。同时每片的间隔可以设置成：1,2,3,5,7秒等。

使用定时任务重试的话，有个缺点就是实时性没那么高，对于实时性要求特别高的业务场景，该方案不太适用。但是对于一般场景，还是可以用一用的。

但它有一个很大的优点，即数据是落库的，不会丢数据。

8.消息队列

在高并发的业务场景中，mq（消息队列）是必不可少的技术之一。它不仅可以异步解耦，还能削峰填谷。对保证系统的稳定性是非常有意义的。

mq的生产者，生产了消息之后，通过指定的topic发送到mq服务器。然后mq的消费者，订阅该topic的消息，读取消息数据之后，做业务逻辑处理。

使用mq重试的具体方案如下：

当用户操作写完数据库，但删除缓存失败了，产生一条mq消息，发送给mq服务器。
mq消费者读取mq消息，重试5次删除缓存。如果其中有任意一次成功了，则返回成功。如果重试了5次，还是失败，则写入死信队列中。
推荐mq使用rocketmq，重试机制和死信队列默认是支持的。使用起来非常方便，而且还支持顺序消息，延迟消息和事务消息等多种业务场景。

当然在该方案中，删除缓存可以完全走异步。即用户的写操作，在写完数据库之后，不用立刻删除一次缓存。而直接发送mq消息，到mq服务器，然后有mq消费者全权负责删除缓存的任务。

因为mq的实时性还是比较高的，因此改良后的方案也是一种不错的选择。

9.binlog

前面我们聊过的，无论是定时任务，还是mq（消息队列），做重试机制，对业务都有一定的侵入性。

在使用定时任务的方案中，需要在业务代码中增加额外逻辑，如果删除缓存失败，需要将数据写入重试表。

而使用mq的方案中，如果删除缓存失败了，需要在业务代码中发送mq消息到mq服务器。

其实，还有一种更优雅的实现，即监听binlog，比如使用：canal等中间件。

具体方案如下：

在业务接口中写数据库之后，就不管了，直接返回成功。
mysql服务器会自动把变更的数据写入binlog中。
binlog订阅者获取变更的数据，然后删除缓存。
这套方案中业务接口确实简化了一些流程，只用关心数据库操作即可，而在binlog订阅者中做缓存删除工作。

但如果只是按照图中的方案进行删除缓存，只删除了一次，也可能会失败。

如何解决这个问题呢？

答：这就需要加上前面聊过的重试机制了。如果删除缓存失败，写入重试表，使用定时任务重试。或者写入mq，让mq自动重试。

在这里推荐使用mq自动重试机制。

在binlog订阅者中如果删除缓存失败，则发送一条mq消息到mq服务器，在mq消费者中自动重试5次。如果有任意一次成功，则直接返回成功。如果重试5次后还是失败，则该消息自动被放入死信队列，后面可能需要人工介入。

Redis综述篇：与面试官彻夜长谈Redis缓存、持久化、淘汰机制、哨兵、集群底层原理！

一、Redis基本概念

面试官心理： 靠！手上活都没干完又叫我过来面试，这不耽误我事么，今儿又得加班补活了…咦，这小伙子简历不错啊，先考考它Redis…
面试官： 谈谈你对Redis的理解？
我： Redis是ANSI C语言编写的一个基于内存的高性能键值对（key-value）的NoSQL数据库，一般用于架设在Java程序与数据库之间用作缓存层，为了防止DB磁盘IO效率过低造成的请求阻塞、响应缓慢等问题，用来弥补DB与Java程序之间的性能差距，同时，也可以在DB吞吐跟不上系统并发量时，避免请求直接落入DB从而起到保护DB的作用。
而Redis一般除了缓存DB数据之外还可以利用它丰富的数据类型及指令来实现一些其他功能，比如：计数器、用户在线状态、排行榜、session存储等，同时Redis的性能也非常可观，通过官方给出的数据显示能够达到10w/s的QPS处理，但是在生产环境的实测结果大概读取QPS在7-9w/s，写入QPS在6-8w/s左右（注：与机器性能也有关），同时Redis也提供事务、持久化、高可用等一些机制的支持。

二、Redis基本数据类型与常用指令

面试官： 刚刚听你提到了可以利用它丰富的数据类型及指令来实现一些其他功能，那你跟我讲讲Redis的一些常用指令。
我： Redis常用的一些命令的话一般是都是对于基本数据类型的操作指令以及一些全局指令…叭啦叭啦叭…，如下：

命令	作用
`keys *`	返回所有键（`keys`还能用来搜索，比如`keys h*`：搜索所有以h开头的键）
`dbsize`	返回键数量，如果存在大量键，线上禁止使用此指令
`exists key`	检查键是否存在，存在返回 1，不存在返回 0
`del key`	删除键，返回删除键个数，删除不存在键返回 0
`ttl key`	查看键存活时间，返回键剩余过期时间，不存在返回-1
`expire key seconds`	设置过期时间（单位：s），成功返回1，失败返回0
`expireat key timestamp`	设置`key`在某个时间戳（精确到秒）之后过期
`pexpire key milliseconds`	设置过期时间（单位：ms），成功返回1，失败返回0
`persist key`	去掉过期时间
`monitor`	实时监听并返回`Redis`服务器接收到的所有请求信息
`shutdown`	把数据同步保存到磁盘上，并关闭`Redis`服务
`info`	查看当前`Redis`节点信息
…	…
当然了，一般也是记得一些常用的命令，但是更多命令参考：`Redis`命令大全，因为`Redis`命令和JVM参数一样，只要记得可以这样做就行了，但是具体的可以去参考相关文档资料。

面试官： 嗯嗯，不错，那再接着讲讲Redis的基本数据类型以及你是在项目中怎么使用它们的吧！
我： Redis数据类型在之前是五种，但是现在的版本中存在九种，分别为：字符串（strings/string）、散列（hashes/hash）、列表（lists/list）、集合（sets/set）、有序集合（sorted sets/zset）以及后续的四种数据类型：bitmaps、hyperloglogs、地理空间（geospatial）、消息（Streams），不过无论是哪种数据类型Redis都不会直接将它放在内存中存储，而是转而内部使用RedisObject来存储以及表示所有类型的key-value（说着说着我拿出了纸和笔，给面试官画了一张图）：
Redis内部使用一个RedisObject对象来表示所有的key和value，RedisObject最主要的信息如上图所示：type表示一个value对象具体是何种数据类型，encoding是不同数据类型在Redis内部的存储方式。比如：type=string表示value存储的是一个普通字符串，那么encoding可以是raw或者int，而关于其他数据类型的内部编码实现我顿时再拿起笔chua~ chua~ chua：
我接着回答： 下面我再简单讲讲Redis的基本数据类型以及它们的应用场景：

类型	描述	特性	场景
`string`	二进制安全	可以存储任何元素（数字、字符、音视频、图片、对象…）	计数器、分布式锁、字符缓存、分布式ID生成、`session`共享、秒杀`token`、IP限流等
`hash`	键值对存储，类似于Map集合	适合存储对象，可以将对象属性一个个存储，更新时也可以更新单个属性，操作某一个字段	对象缓存、购物车等
`list`	双向链表	增删快	栈、队列、有限集合、消息队列、消息推送、阻塞队列等
`set`	元素不能重复，每次获取无序	添加、删除、查找的复杂度都是O(1)，提供了求交集、并集、差集的操作	抽奖活动、朋友圈点赞、用户（微博好友）关注、相关关注、共同关注、好友推荐（可能认识的人）等
`sorted set`	有序集合，每个元素有一个对应的分数，不允许元素重复	基于分数进行排序，如果分数相等，以key值的 ascii 值进行排序	商品评价标签（好评、中评、差评等）、排行榜等
`bitmaps`	`Bitmaps`是一个字节由 8 个二进制位组成	在字符串类型上面定义的位操作	在线用户统计、用户访问统计、用户点击统计等
`hyperloglog`	`Redis2.8.9`版本添加了 `HyperLogLog`结构。`Redis HyperLogLog`是用来做基数统计的算法。	用于进行基数统计，不是集合，不保存数据，只记录数量而不是具体数据	统计独立UV等
`geospatial`	`Redis3.2`版本新增的数据类型：`GEO`对地理位置的支持	以将用户给定的地理位置信息储存起来，并对这些信息进行操作	地理位置计算
`stream`	`Redis5.0`之后新增的数据类型	支持发布订阅，一对多消费	消息队列

PS：HyperLogLog的优点是，在输入元素的数量或者体积非常非常大时，计算基数所需的空间总是固定的、并且是很小的。在 Redis 里面，每个HyperLogLog键只需要花费12 KB内存，就可以计算接近2^64个不同元素的基数。这和计算基数时，元素越多耗费内存就越多的集合形成鲜明对比。但是，因为HyperLogLog只会根据输入元素来计算基数，而不会储存输入元素本身，所以HyperLogLog不能像集合那样，返回输入的各个元素（核心是基数估算算法，最终数值存在一定误差误差范围：基数估计的结果是一个带有0.81%标准错误的近似值，耗空间极小，每个hyperloglog key占用了12K的内存用于标记基数，pfadd命令不是一次性分配12K内存使用，会随着基数的增加内存逐渐增大，Pfmerge命令合并后占用的存储空间为12K，无论合并之前数据量多少）

三、Redis缓存及一致性、雪崩、击穿与穿透问题

面试官提问： 那么你们在使用Redis做为缓存层的时候是怎么通过Java操作Redis的呢？
我的心理： 这问题不是送命题吗…
我： Java操作Redis的客户端有很多，比如springData中的RedisTemplate，也有SpringCache集成Redis后的注解形式，当然也会有一些Jedis、Lettuce、Redisson等等，而我们使用的是Lettuce以及Redisson........
面试官提问： 那你们在使用Redis作为缓存的时候有没有遇到什么问题呢？
我：咳咳，是的，确实遇到了以及考虑到了一些问题，比如缓存一致性、雪崩、穿透与击穿，关于Redis与MySQL之间的数据一致性问题其实也考虑过很多方案，比如先删后改，延时双删等等很多方案，但是在高并发情况下还是会造成数据的不一致性，所以关于DB与缓存之间的强一致性一定要保证的话那么就对于这部分数据不要做缓存，操作直接走DB，但是如果这个数据比较热点的话那么还是会给DB造成很大的压力，所以在我们的项目中还是采用先删再改+过期的方案来做的，虽然也会存在数据的不一致，但是勉强也能接受，因为毕竟使用缓存访问快的同时也能减轻DB压力，而且本身采用缓存就需要接受一定的数据延迟性和短暂的不一致性，我们只能采取合适的策略来降低缓存和数据库间数据不一致的概率，而无法保证两者间的强一致性。合适的策略包括合适的缓存更新策略，合适的缓存淘汰策略，更新数据库后及时更新缓存、缓存失败时增加重试机制等。
面试官话锋一转： 打断一下，你刚刚提到了使用缓存能让访问变快，那么你能不能讲讲Redis为什么快呢？
我的心理： 好家伙，这一手来的我猝不及防…
硬着头发回答： Redis快的原因嘛其实可以从多个维度来看待：
- 一、Redis完全基于内存
- 二、Redis整个结构类似于HashMap，查找和操作复杂度为O(1)，不需要和MySQL查找数据一样需要产生随机磁盘IO或者全表
- 三、Redis对于客户端的处理是单线程的，采用单线程处理所有客户端请求，避免了多线程的上下文切换和线程竞争造成的开销
- 四、底层采用select/epoll多路复用的高效非阻塞IO模型
- 五、客户端通信协议采用RESP，简单易读，避免了复杂请求的解析开销
面试官露出姨父般的慈笑： 嗯嗯，还不错，那你继续谈谈刚刚的缓存雪崩、穿透与击穿的问题吧
我：好的，先说缓存雪崩吧，缓存雪崩造成的原因是因为我们在做缓存时为了保证内存利用率，一般在写入数据时都会给定一个过期时间，而就是因为过期时间的设置有可能导致大量的热点key在同一时间内全部失效，此时来了大量请求访问这些key，而Redis中却没有这些数据，从而导致所有请求直接落入DB查询，造成DB出现瓶颈或者直接被打宕导致雪崩情况的发生。关于解决方案的的话也可以从多个维度来考虑：
- 一、设置热点数据永不过期，避免热点数据的失效导致大量的相同请求落入DB
- 二、错开过期时间的设置，根据业务以及线上情况合理的设置失效时间
- 三、使用分布式锁或者MQ队列使得请求串行化，从而避免同一时间请求大量落入DB（性能会受到很大的影响）
面试官： 那缓存穿透呢？指的是什么？又该怎么解决？
我喝了口水接着回答： 缓存穿透这个问题是由于请求参数不合理导致的，比如对外暴露了一个接口getUser?userID=xxx，而数据库中的userID是从1开始的，当有黑客通过这个接口携带不存在的ID请求时，比如：getUser?userID=-1，请求会先来到Redis中查询缓存，但是发现没有对应的数据从而转向DB查询，但是DB中也无此值, 所以也无法写入数据到缓存，而黑客就通过这一点利用“肉鸡”等手段疯狂请求这个接口，导致出现大量Redis不存在数据的请求落入DB，从而导致DB出现瓶颈或者直接被打宕机，整个系统陷入瘫痪。
面试官： 嗯，那又该如果避免这种情况呢？
我：解决方案也有好几种呢：
- 一、做IP限流与黑名单，避免同一IP一瞬间发送大量请求
- 二、对于请求做非法校验，对于携带非法参数的请求直接过滤
- 三、对于DB中查询不存在的数据写入Redis中“Not Data”并设置短暂的过期时间，下次请求能够直接被拦截在Redis而不会落入DB
- 四、布隆过滤器
面试官： 那接下来的缓存击穿呢？又是怎么回事？怎么解决？
我：这个简单，缓存击穿和缓存雪崩有点类似，都是由于请求的key过期导致的问题，但是不同点在于失效key的数量，对于雪崩而言指的是大量的key失效导致大量请求落入DB，而对于击穿而言，指的是某一个热点key突然过期，而这个时候又突然又大量的请求来查询它，但是在Redis中却并没有查询到结果从而导致所有请求全部打向DB，导致在这个时刻DB直接被打穿。解决方案的话也是有多种：
- 一、设置热点key永不过期
- 二、做好Redis监控，请求串行化访问（性能较差）
- 使用mutex锁机制：就是在缓存失效的时候（判断拿出来的值为空），不是立即去load db，而是先使用缓存工具的某些带成功操作返回值的操作（比如Redis的SETNX或者Memcache的ADD）去set一个mutex key，当操作返回成功时，再进行load db的操作并回设缓存；否则，就重试整个get缓存的方法，代码实现如下：

public Result get(int ID){
    RedisResult = Redis.get(ID);
     if(RedisResult != null){
         return RedisResult;
     }
     if(Redis.setnx("update:" + ID) != "0"){
         DBResult = DB.selectByID(ID);
         if(DBResult != null){ // 避免缓存穿透
             Redis.set(ID,DBResult);
             Redis.del("update:" + ID);
             return DBResult;
         }
         Redis.set(ID,"Not Data");
         return "抱歉，当前查询暂时没有找到数据......";
     }
     Thread.sleep(2);
     return get(ID);
}
复制代码

四、Redis八种淘汰策略与三种删除策略

4.1. 八种键淘汰（过期）策略

面试官： 你前面提到过，Redis的数据是全部放在内存中的，那么有些数据我也没有设置过期时间，导致了大量的内存浪费，当我有新的数据需要写入内存不够用了怎么办？
我的内心： 好家伙，问个Redis淘汰策略这么拐弯抹角…
我：我想你是想问内存淘汰策略吧，Redis在5.0之前为我们提供了六种淘汰策略，而5.0为我们提供了八种，但是大体上来说这些lru、lfu、random、ttl四种类型，如下：

策略	概述
`volatile-lru`	从已设置过期时间的数据集中挑选最近最少使用的数据淘汰，没有设置过期时间的key不会被淘汰，这样就可以在增加内存空间的同时保证需要持久化的数据不会丢失。
`volatile-ttl`	从已设置过期时间的数据集中挑选将要过期的数据淘汰，ttl值越大越优先被淘汰。
`volatile-random`	从已设置过期时间的数据集中任意选择数据淘汰
`volatile-lfu`	从已设置过期时间的数据集挑选使用频率最低的数据淘汰
`allkeys-lru`	从数据集中挑选最近最少使用的数据淘汰，该策略要淘汰的key面向的是全体key集合，而非过期的key集合（应用最广泛的策略）。
`allkeys-lfu`	从数据集中挑选使用频率最低的数据淘汰
`allkeys-random`	从数据集（`server.db[i].dict`）中任意选择数据淘汰
`no-enviction`（驱逐）	禁止驱逐数据，这也是默认策略。意思是当内存不足以容纳新入数据时，新写入操作就会报错，请求可以继续进行，线上任务也不能持续进行，采用`no-enviction`策略可以保证数据不被丢失。

我喘了口气接着说：
- 一、在Redis中，数据有一部分访问频率较高，其余部分访问频率较低，或者无法预测数据的使用频率时，设置allkeys-lru是比较合适的。
- 二、如果所有数据访问概率大致相等时，可以选择allkeys-random。
- 三、如果研发者需要通过设置不同的ttl来判断数据过期的先后顺序，此时可以选择volatile-ttl策略。
- 四、如果希望一些数据能长期被保存，而一些数据可以被淘汰掉时，选择volatile-lru或volatile-random都是比较不错的。
- 五、由于设置expire会消耗额外的内存，如果计划避免Redis内存在此项上的浪费，可以选用allkeys-lru策略，这样就可以不再设置过期时间，高效利用内存了。
- maxmemory-policy：参数配置淘汰策略。maxmemory：限制内存大小。

4.2. 三种键删除策略

面试官： 那Redis的Key删除策略有了解过吗？
我：
```
Redis
```
删除
```
Key
```
的策略策略有三种：
- 定时删除：在设置键的过期时间的同时，设置一个定时器，当键过期了，定时器马上把该键删除。（定时删除对内存来说是友好的，因为它可以及时清理过期键；但对CPU是不友好的，如果过期键太多，删除操作会消耗过多的资源。）
- 惰性删除：key过期后任然留在内存中不做处理，当有请求操作这个key的时候，会检查这个key是否过期，如果过期则删除，否则返回key对应的数据信息。（惰性删除对CPU是友好的，因为只有在读取的时候检测到过期了才会将其删除。但对内存是不友好，如果过期键后续不被访问，那么这些过期键将积累在缓存中，对内存消耗是比较大的。）
- 定期删除：Redis数据库默认每隔100ms就会进行随机抽取一些设置过期时间的key进行检测，过期则删除。（定期删除是定时删除和惰性删除的一个折中方案。可以根据实际场景自定义这个间隔时间，在CPU资源和内存资源上作出权衡。）
- Redis默认采用定期+惰性删除策略。

五、Redis三种持久化机制

5.1. RDB持久化

面试官： 那么你刚刚提到的Redis为了保证性能会将所有数据放在内存，那么机器突然断电或宕机需要重启，内存中的数据岂不是没有了？
我： Redis的确是将数据存储在内存的，但是也会有相关的持久化机制将内存持久化备份到磁盘，以便于重启时数据能够重新恢复到内存中，避免数据丢失的风险。而Redis持久化机制由三种，在4.X版本之前Redis只支持AOF以及RDB两种形式持久化，但是因为AOF与RDB都存在各自的缺陷，所以在4.x版本之后Redis还提供一种新的持久化机制：混合型持久化（但是最终生成的文件还是.AOF）。
面试官： 那你仔细讲讲这几种持久化机制吧
我：

好的，
```
RDB
```
持久化把内存中当前进程的数据生成快照（
```
.rdb
```
）文件保存到硬盘的过程，有手动触发和自动触发：
- 自动触发：
  - Redis RDB持久化默认开启
    save 900 1 – 900s内存在1个写操作
    save 300 10 – 300s内存在10个写操作
    save 60 10000 – 60s内存在10000个写操作
    如上是RDB的自动触发的默认配置，当操作满足如上条件时会被触发。
- 手动触发：
  - save：阻塞当前 Redis，直到RDB持久化过程完成为止，若内存实例比较大会造成长时间阻塞，线上环境不建议用它
  - bgsave：Redis 进程执行fork操作创建子进程，由子进程完成持久化，阻塞时间很短（微秒级），是save的优化,在执行Redis-cli shutdown关闭Redis服务时或执行flushall命令时，如果没有开启AOF持久化，自动执行bgsave，bgsave执行流程如下：
    
    而且RDB 是在某个时间点将数据写入一个临时文件，持久化结束后，用这个临时文件替换上次持久化的文件，重启时加载这个文件达到数据恢复。
RDB优缺点：
- 优点：使用单独子进程来进行持久化，主进程不会进行任何 IO 操作，保证了 Redis 的高性能；而且RDB文件存储的是压缩的二进制文件，适用于备份、全量复制，可用于灾难备份，同时RDB文件的加载速度远超于AOF文件。
- 缺点：RDB是间隔一段时间进行持久化，如果持久化之间的时间内发生故障，会出现数据丢失。所以这种方式更适合数据要求不严谨的时候，因为RDB无法做到实时持久化，而且每次都要创建子进程，频繁创建成本过高；备份时占用内存，因为Redis 在备份时会独立创建一个子进程，将数据写入到一个临时文件（需要的内存是原本的两倍）；还有一点，RDB文件保存的二进制文件存在新老版本不兼容的问题。

5.1. AOF持久化

我：

而
```
AOF
```
持久化方式能很好的解决
```
RDB
```
持久化方式造成的数据丢失，
```
AOF
```
持久化到硬盘中的并不是内存中的数据快照，而是和
```
MySQL
```
的
```
binlog
```
日志一样记录写入命令，
```
AOF
```
的持久化策略也有三种：
- appendfsync always：同步持久化形式，每次发生数据更改都将命令追加到AOF文件，因为每次写入时都记录会产生大量磁盘IO，从而性能会受到影响，但是数据最安全。
- appendfsync everysec：Redis开启AOF后的缺省配置，异步操作，每秒将写入命令追加到AOF文件，如果在刚持久化之后的一秒内宕机，会造成1S的数据丢失。
- appendfsync no：Redis并不直接调用文件同步，而是交给操作系统来处理，操作系统可以根据buffer填充情况/通道空闲时间等择机触发同步；这是一种普通的文件操作方式。性能较好，在物理服务器故障时，数据丢失量会因OS配置有关。
- ```
AOF
```
  持久化机制优缺点：
  - 优点：根据不同的fsync策略可以保证数据丢失风险降到最低，数据能够保证是最新的，fsync是后台线程在处理，所以对于处理客户端请求的线程并不影响。
  - 缺点：文件体积由于保存的是所有命令会比RDB大上很多，而且数据恢复时也需要重新执行指令，在重启时恢复数据的时间往往会慢很多。虽然fsync并不是共用处理客户端请求线程的资源来处理的，但是这两个线程还是在共享同一台机器的资源，所以在高并发场景下也会一定受到影响。
AOF机制重写：

随着
```
Redis
```
在线上运行的时间越来越久，客户端执行的命令越来越多，
```
AOF
```
的文件也会越来越大，当
```
AOF
```
达到一定程度大小之后再通过
```
AOF
```
文件恢复数据是异常缓慢的，那么对于这种情况
```
Redis
```
在开启
```
AOF
```
持久化机制的时候会存在
```
AOF
```
文件的重写，缺省配置是当
```
AOF
```
文件比上一次重写时的文件大小增长
```
100%
```
并且文件大小不小于
```
64MB
```
时会对整个
```
AOF
```
文件进行重写从而达到“减肥”的目的（这里的
```
100%
```
和
```
64MB
```
可以通过
```
auto-aof-rewrite-percentage 100
```
与
```
auto-aof-rewrite-min-size 64mb
```
来调整）。而
```
AOF rewrite
```
操作就是“压缩”
```
AOF
```
文件的过程，当然
```
Redis
```
并没有采用“基于原
```
aof
```
文件”来重写的方式，而是采取了类似
```
snapshot
```
的方式：基于
```
copy-on-write
```
，全量遍历内存中数据，然后逐个序列到
```
aof
```
文件中。因此
```
AOF rewrite
```
能够正确反应当前内存数据的状态，这正是我们所需要的；*
```
rewrite
```
过程中，对于新的变更操作将仍然被写入到原
```
AOF
```
文件中，同时这些新的变更操作也会被
```
Redis
```
收集起来(
```
buffer，copy-on-write
```
方式下，最极端的可能是所有的
```
key
```
都在此期间被修改，将会耗费
```
2
```
倍内存)，当内存数据被全部写入到新的
```
aof
```
文件之后，收集的新的变更操作也将会一并追加到新的
```
aof
```
文件中，此后将会重命名新的
```
aof
```
文件为
```
appendonly.aof
```
, 此后所有的操作都将被写入新的
```
aof
```
文件。如果在
```
rewrite
```
过程中，出现故障，将不会影响原
```
AOF
```
文件的正常工作，只有当
```
rewrite
```
完成之后才会切换文件，因为
```
rewrite
```
过程是比较可靠的，触发
```
rewrite
```
的时机可以通过配置文件来声明，同时
```
Redis
```
中可以通过
```
bgrewriteaof
```
指令人工干预。
- AOF持久化过程如下：
面试官： 那你项目中Redis采用的是那种持久化方式呢？
我：在我们项目中考虑到了Redis中不仅仅只是用来做缓存，其中还存储着一些MySQL中不存在的数据，所以数据的安全性要求比较高，而RDB因为并不是实时的持久化，会出现数据丢失，但是采用AOF形式在重启、灾备、迁移的时候过程异常耗时，也并不理想，所以在我们线上是同时采用两种形式的，而AOF+RDB两种模式同时开启时Redis重启又该加载谁呢？（说着说着我又掏出了纸笔给面试官画了如下一幅图）：

5.3. 4.x之后的混合型持久化

当然在Redis4.x之后推出了混合型持久化机制，因为RDB虽然加载快但是存在数据丢失，AOF数据安全但是加载缓慢，Redis为了解决这个问题，带来了一个新的持久化选项——混合持久化。将RDB文件的内容和增量的AOF日志文件存在一起。这里的AOF日志不再是全量的日志，而是自持久化开始到持久化结束的这段时间发生的增量AOF日志，通常这部分AOF日志很小。Redis重启的时候，可以先加载RDB的内容，然后再重放增量AOF日志，就可以完全替代之前的AOF全量文件重放，恢复效率因此大幅得到提升（混合型持久化最终生成的文件后缀是.aof，可以通过redis.conf文件中aof-use-rdb-preamble yes配置开启）。 – 混合型持久化优点：结合了RDB和AOF的优点，使得数据恢复的效率大幅提升 – 混合型持久化缺点：兼容性不好，Redis-4.x新增，虽然最终的文件也是.aof格式的文件，但在4.0之前版本都不识别该aof文件，同时由于前部分是RDB格式，阅读性较差

六、`Redis`的事务机制

面试官： 既然Redis是数据库，那么它支不支持事务呢？
我：
```
Redis
```
作为数据库当然是支持事务的，只不过
```
Redis
```
的事务机制是弱事务，相对来说比较鸡肋，官方给出如下几个指令来进行
```
Redis
```
的事务控制：
- MULTI：标记一个事务块的开始
- DISCARD：取消事务，放弃执行事务块内的所有命令
- EXEC：执行所有事务块内的命令
- UNWATCH：取消WATCH命令对所有key的监视
- WATCH key [key ...]：监视一个(或多个)key，如果在事务执行之前这个(或这些)key被其他命令所改动，那么事务将被打断

七、`Redis`内存模型及内存划分

面试官： 嗯嗯，挺不错，那你对于Redis的内存模型以及内存的划分有去了解过嘛？
我：

了解过的，
```
Redis
```
的内存模型我们可以通过客户端连接之后使用内存统计命令
```
info memory
```
去查看，如下：
- used_memory（单位：字节）： Redis分配器分配的内存总量，包括使用的虚拟内存（稍后会详解）
- used_memory_rss（单位：字节）： Redis进程占据操作系统的内存；除了分配器分配的内存之外，used_memory_rss还包括进程运行本身需要的内存、内存碎片等，但是不包括虚拟内存
- 说明： used_memory是从Redis角度得到的量，used_memory_rss是从操作系统角度得到的量。二者之所以有所不同，一方面是因为内存碎片和Redis进程运行需要占用内存，使得used_memory_rss可能更大；另一方面虚拟内存的存在，使得used_memory可能更大
- mem_fragmentation_ratio： 内存碎片比率，该值是used_memory_rss / used_memory；一般大于1，且该值越大，内存碎片比例越大。而小于1，说明Redis使用了虚拟内存，由于虚拟内存的媒介是磁盘，比内存速度要慢很多，当这种情况出现时，应该及时排查，如果内存不足应该及时处理，如增加Redis节点、增加Redis服务器的内存、优化应用等；一般来说，mem_fragmentation_ratio在1.03左右是比较健康的状态（对于jemalloc分配器来说），由于在实际应用中，Redis的数据量会比较大，此时进程运行占用的内存与Redis数据量和内存碎片相比，都会小得多，mem_fragmentation_ratio便成了衡量Redis内存碎片率的参数
- mem_allocator： Redis使用的内存分配器，在编译时指定；可以是libc 、jemalloc或tcmalloc，默认是jemalloc
我接着说：

而
```
Redis
```
作为内存数据库，在内存中存储的内容主要是数据，但除了数据以外，
```
Redis
```
的其他部分也会占用内存。
```
Redis
```
的内存占用可以划分为以下几个部分：
- 数据： 作为数据库，数据是最主要的部分；这部分占用的内存会统计在used_memory中
- 进程本身运行需要的内存： Redis主进程本身运行肯定需要占用内存，如代码、常量池等等，这部分内存大约几兆，在大多数生产环境中与Redis数据占用的内存相比可以忽略。这部分内存不是由jemalloc分配，因此不会统计在used_memory中。除了主进程外，Redis创建的子进程运行也会占用内存，如Redis执行AOF、RDB重写时创建的子进程。当然，这部分内存不属于Redis进程，也不会统计在used_memory和used_memory_rss中。
- 缓冲内存： 缓冲内存包括客户端缓冲区、复制积压缓冲区、AOF缓冲区等；其中，客户端缓冲存储客户端连接的输入输出缓冲；复制积压缓冲用于部分复制功能；AOF缓冲区用于在进行AOF重写时，保存最近的写入命令。在了解相应功能之前，不需要知道这些缓冲的细节；这部分内存由jemalloc分配，因此会统计在used_memory中。
- 内存碎片：
  
  内存碎片是
```
Redis
```
  在分配、回收物理内存过程中产生的。例如，如果对数据的更改频繁，而且数据之间的大小相差很大，可能导致
```
Redis
```
  释放的空间在物理内存中并没有释放，但
```
Redis
```
  又无法有效利用，这就形成了内存碎片。内存碎片不会统计在
```
used_memory
```
  中。
  - 内存碎片的产生与对数据进行的操作、数据的特点等都有关；此外，与使用的内存分配器也有关系：如果内存分配器设计合理，可以尽可能的减少内存碎片的产生。如果Redis服务器中的内存碎片已经很大，可以通过安全重启的方式减小内存碎片：因为重启之后，Redis重新从备份文件中读取数据，在内存中进行重排，为每个数据重新选择合适的内存单元，减小内存碎片。
面试官： 那Redis的共享对象你有了解过吗？
在
```
RedisObject
```
对象中有一个
```
refcount
```
，
```
refcount
```
记录的是该对象被引用的次数，类型为整型。
```
refcount
```
的作用，主要在于对象的引用计数和内存回收。当创建新对象时，
```
refcount
```
初始化为1；当有新程序使用该对象时，refcount加1；当对象不再被一个新程序使用时，
```
refcount
```
减1；当
```
refcount
```
变为0时，对象占用的内存会被释放。
```
Redis
```
中被多次使用的对象(
```
refcount>1
```
)，称为共享对象。
```
Redis
```
为了节省内存，当有一些对象重复出现时，新的程序不会创建新的对象，而是仍然使用原来的对象。这个被重复使用的对象，就是共享对象。目前共享对象仅支持整数值的字符串对象。
- 共享对象的具体实现：
  Redis的共享对象目前只支持整数值的字符串对象。之所以如此，实际上是对内存和CPU（时间）的平衡：共享对象虽然会降低内存消耗，但是判断两个对象是否相等却需要消耗额外的时间。对于整数值，判断操作复杂度为O(1)；对于普通字符串，判断复杂度为O(n)；而对于哈希、列表、集合和有序集合，判断的复杂度为O(n^2)。虽然共享对象只能是整数值的字符串对象，但是5种类型都可能使用共享对象（如哈希、列表等的元素可以使用）。
  就目前的实现来说，Redis服务器在初始化时，会创建10000个字符串对象，值分别是0-9999的整数值；当Redis需要使用值为0-9999的字符串对象时，可以直接使用这些共享对象。10000这个数字可以通过调整参数Redis_SHARED_INTEGERS（4.0中是OBJ_SHARED_INTEGERS）的值进行改变。
  共享对象的引用次数可以通过object refcount命令查看。

八、Redis虚拟内存

面试官： 刚刚听你提到过Redis的虚拟内存，那你能详细讲讲它是怎么会事吗？
我：

首先说明下
```
Redis
```
的虚拟内存与操作系统虚拟内存不是一码事,但是思路和目的都是相冋的。就是暂时把不经常访问的数据从內存交换到磁盘中,从而腾出宝贵的内存空间。对于
```
Redis
```
这样的内存数据库,内存总是不够用的。除了可以将数据分割到多个
```
Redis
```
实例以外。另外的能够提高数据库容量的办法就是使用虚拟内存技术把那些不经常访问的数据交换到磁盘上。如果我们存储的数据总是有少部分数据被经常访问,大部分数据很少被访问,对于网站来说确实总是只有少量用户经常活跃。当少量数据被经常访问时,使用虚拟内存不但能提高单台
```
Redis
```
数据库服务器的容量,而且也不会对性能造成太多影响
```
Redis
```
没有使用操作系统提供的虚拟内存机制而是自己在用户态实现了自己的虚拟内存机制。主要的理由有以下两点：
- 一、操作系统的虚拟内存是以4k/页为最小单位进行交换的。而Redis的大多数对象都远小于4k,所以一个操作系统页上可能有多个Redis对象。另外 Redis的集合对象类型如list,set可能行在于多个操作系统页上。最终可能造成只有10%的key被经常访问,但是所有操作系统页都会被操作系统认为是活跃的,这样只有内存真正耗尽时操作系统才会进行页的交换
- 二、相比操作系统的交换方式，Redis可以将被交换到磁盘的对象进行压缩,保存到磁盘的对象可以去除指针和对象元数据信息。一般压缩后的对象会比内存中的对象小10倍。这样Redis的虛拟内存会比操作系统的虚拟内存少做很多I0操作
我：

而关于
```
Redis
```
虚拟内存的配置也存在于
```
redis.conf
```
文件中，如下：
- vm-enabled ves：#开启虚拟内存功能
- vm-swap-file ../redis.swap：#交换出来value保存的文件路径
- Vm-max-memory 268435456：# Redis使用的最大内存上限(256MB),超过上限后Redis开始交换value到磁盘swap文件中。建议设置为系统空闲内存的60%-80%
- vm-page-size 32：#每个 Redis页的大小32个字节
- vm-pages 134217728：#最多在文件中使用多少个页,交换文件的大小
- vm-max-threads 8：#用于执行value对象换入换出的工作线程数量，0表示不使用工作线程(详情后面介绍)。
我：
```
Redis
```
的虚拟内存在设计上为了保证
```
key
```
的查询速度,只会将
```
value
```
交换到
```
swap
```
文件。如果是由于太多
```
key
```
很小的
```
value
```
造成的内存问题,那么
```
Redis
```
的虚拟内存并不能解决问题。和操作系统一样
```
Redis
```
也是按页来交换对象的。
```
Redis
```
规定同一个页只能保存一个对象。但是一个对象可以保存在多个页中。在
```
Redis
```
使用的内存没超过
```
vm-max-memory
```
之前是不会交换任何
```
value
```
的。当超过最大内存限制后,
```
Redis
```
会选择把较老的对象交换到
```
swap
```
文件中去。如果两个对象一样老会优先交换比较大的对象,精确的交换计算公式
```
swappability=age*1og(size_Inmemory)
```
。对于
```
vm-page-size
```
的设置应该根据自己应用将页的大小设置为可以容纳大多数对象的尺寸。太大了会浪费磁盘空间,太小了会造成交换文件出现过多碎片。对于交换文件中的每个页,
```
Redis
```
会在内存中用一个
```
1bit
```
值来对应记录页的空闲状态。所以像上面配置中页数量(
```
vm pages134217728
```
)会占用
```
16MB
```
内存用来记录页的空內状态。
```
vm-max-threads
```
表示用做交换任务的工作线程数量。如果大于0推荐设为服务器的cpu的核心数。如果是0则交换过程在上线程进行。具体工作模式如下：
- 阻塞模式(
```
vm-max-threads=0
```
  )：
  - 换出：主线程定期检査发现内存超出最大上限后,会直接以阻塞的方式,将选中的对象保存到
```
swap
```
    文件中,并释放对象占用的内存空间,此过程会一直重复直到下面条件满足。
    - 内存使用降到最大限制以下
    - swap文件满了
    - 几乎全部的对象都被交换到磁盘了
  - 换入：当有客户端请求已经被换出的value时,主线程会以阳塞的方式从swap文件中加载对应的value对象,加载时此时会阻塞所客户端。然后处理该客户端的请求
- 非阻塞模式(
```
vm-max-threads>0
```
  )：
  - 换出：当主线程检测到使用内存超过最大上限,会将选中要父换的对象信息放到一个队列中父给工作线程后台处理,主线程会继续处理客户端请求
  - 换入：如果有客户端请求的key已终被换出了,主线程会先阳塞发出命令的客户端,然后将加载对象的信息放到一个队列中,让工作线程去加载。加载完毕后工作线程通知主线程。主线程再执行客户端的命令。这种方式只阻塞请求的value是已经被换出key的客户端总的来说阻塞方式的性能会好些,因为不需要线程同步、创建线程和恢复被阻塞的客户端等开销。但是也相应的牺牡了响应性。工作线稈方式主线程不会阳塞在磁盘1O上,所以响应性更好。如果我们的应用不太经常发生换入换出,而且也不太在意有点延迟的话推荐使用阻塞方式（详细介绍参考）。

九、`Redis`客户端通信RESP协议

面试官： 那你再简单讲讲Redis的客户端通信的RESP协议吧
我：

这个比较简单，
```
RESP
```
是
```
Redis
```
序列化协议，
```
Redis
```
客户端
```
RESP
```
协议与
```
Redis
```
服务器通信。
```
RESP
```
协议在
```
Redis 1.2
```
中引入，但在
```
Redis 2.0
```
中成为与
```
Redis
```
服务器通信的标准方式。这个通信方式就是
```
Redis
```
客户端实现的协议。RESP实际上是一个序列化协议，它支持以下数据类型:简单字符串、错误、整数、大容量字符串和数组。当我们在客户端中像
```
Redis
```
发送操作命令时，比如：
```
set name 竹子爱熊猫
```
这条命令，不会直接以这种格式的形式发送到
```
Redis Server
```
，而是经过
```
RESP
```
的序列化之后再发送给
```
Redis
```
执行，而AOF持久化机制持久化之后生成的AOF文件中也并不是存储
```
set name 竹子爱熊猫
```
这个指令，而是存储
```
RESP
```
序列化之后的指令，
```
RESP
```
的特点如下：
- 实现简单
- 能被计算机快速地解析
- 可读性好能够被人工解析

十、Redis高可用机制：主从复制、哨兵、代理式/分片式集群

10.1. 主从复制

面试官： 如果在整个架构中加入Redis作为缓存层，那么会在Java程序与DB之间多出一层访问，假设Redis挂了那么Java程序这边又会抛出异常导致所有请求死在这里从而导致整个系统的不可用，那么怎么避免Redis出现这类的单点故障呢？
我： Redis既然这么受欢迎那么这些问题它都提供了相关的解决方案的，Redis有提供了主从、哨兵、代理集群与分片集群的高可用机制来保证出现单点问题时能够及时的切换机器以保障整个系统不受到影响。但是后续的三种高可用机制都是基于主从的基础上来实现的，所以我先说说Redis的主从复制。虽然我们之前讲到过持久化机制可以保证数据重启情况下也不丢失，但是由于是存在于一台服务器上的，如果机器磁盘坏了、机房爆炸（玩笑~）等也会导致数据丢失，而主从复制可以将数据同步到多台不同机器，也能够保证在主节点宕机时任然对外提供服务，还可以做到通过读写分离的形式提升整体缓存业务群吞吐量。一般在线上环境时我们去搭建主从环境时，为了保证数据一致性，从节点是不允许写的，而是通过复制主节点数据的形式保障数据同步。所以在整个Redis节点群中只能同时运行存在一台主，其他的全为从节点，示意图如下（读的QPS可以通过对从节点的线性扩容来提升）：
面试官： 那你能详细说下主从数据同步的过程吗？
我：

可以的，
```
Redis2.8
```
之前使用
```
sync[runId][offset]
```
同步命令，
```
Redis2.8
```
之后使用
```
psync[runId][offset]
```
命令。两者不同在于，
```
sync
```
命令仅支持全量复制过程，
```
psync
```
支持全量和部分复制。介绍同步之前，先介绍几个概念：
- runId：每个Redis节点启动都会生成唯一的uuid，每次Redis重启后，runId都会发生变化
- offset：主节点和从节点都各自维护自己的主从复制偏移量offset，当主节点有写入命令时，offset=offset+命令的字节长度。从节点在收到主节点发送的命令后，也会增加自己的offset，并把自己的offset发送给主节点。这样，主节点同时保存自己的offset和从节点的offset，通过对比offset来判断主从节点数据是否一致
- repl_back_buffer：复制缓冲区，用来存储增量数据命令
- 主从数据同步具体过程如下：
我：当然psync命令除了支持全量复制之外还支持部分复制，因为在做主从数据同步时会导致主从机器网络带宽开销非常大，而在2.8之前Redis仅支持全量复制，这样非常容易导致Redis在线上出现网络瓶颈，而在2.8之后的增量（部分）复制，用于处理在主从复制中因网络闪断等原因造成的数据丢失场景，当slave再次连上master后，如果条件允许，master会补发丢失数据给slave。因为补发的数据远远小于全量数据，可以有效避免全量复制的过高开销。部分复制流程图如下（复制缓存区溢出也会导致全量复制）：
PS：
```
psync[runid][offset]
```
命令三种返回值：
- FULLRESYNC：第一次连接，进行全量复制
- CONTINUE：进行部分复制
- ERR：不支持psync命令，进行全量复制
面试官： 那你觉得主从机制有什么好处？存在什么问题？
我：

主从机制其实也是为后续的一些高可用机制打下了基础，但是本身也存在一些缺陷，当然在后续的高可用机制中得到了解决，具体如下：
- 优点：
  - 能够为后续的高可用机制打下基础
  - 在持久化的基础上能够将数据同步到其他机器，在极端情况下做到灾备的效果
  - 能够通过主写从读的形式实现读写分离提升Redis整体吞吐，并且读的性能可以通过对从节点进行线性扩容无限提升
- 缺点：
  - 全量数据同步时如果数据量比较大，在之前会导致线上短暂性的卡顿
  - 一旦主节点宕机，从节点晋升为主节点，同时需要修改应用方的主节点地址，还需要命令所有从节点去复制新的主节点，整个过程需要人工干预
  - 写入的QPS性能受到主节点限制，虽然主从复制能够通过读写分离来提升整体性能，但是只有从节点能够做到线性扩容升吞吐，写入的性能还是受到主节点限制
  - 木桶效应，整个Redis节点群能够存储的数据容量受到所有节点中内存最小的那台限制，比如一主两从架构：master=32GB、slave1=32GB、slave2=16GB，那么整个Redis节点群能够存储的最大容量为16GB

10.2. 哨兵机制

面试官： 你刚刚提到过后续的高可用机制能解决这些问题，你说的是哨兵吗？那你再说说哨兵机制
我：

好的，哨兵机制的确能够解决之前主从存在的一些问题，如图：

上图所示是目前企业中常用的
```
Redis
```
架构，一主两从三哨兵架构，
```
Redis Sentinel
```
（哨兵）主要功能包括主节点存活检测、主从运行情况检测、自动故障转移、主从切换。
```
Redis Sentinel
```
最小配置是一主一从。
```
Redis
```
的
```
Sentinel
```
系统可以用来管理多个
```
Redis
```
节点，该系统可以执行以下四个任务：
- 监控：不断检查主服务器和从服务器是否正常运行
- 通知：当被监控的某个Redis服务器出现问题，Sentinel通过API脚本向管理员或者其他应用程序发出通知
- 自动故障转移：当主节点不能正常工作时，Sentinel会开始一次自动的故障转移操作，它会将与失效主节点是主从关系的其中一个从节点升级为新的主节点，并且将其他的从节点指向新的主节点，这样就不需要人工干预进行主从切换
- 配置提供者：在Sentinel模式下，客户端应用在初始化时连接的是Sentinel节点集合，从中获取主节点的信息
面试官： 那你能讲讲哨兵机制原理吗？
我：

可以的，哨兵的工作原理如下：
- 一、每个哨兵节点每10秒会向主节点和从节点发送info命令获取最级联结构图，哨兵配置时只要配置对主节点的监控即可，通过向主节点发送info，获取从节点的信息，并当有新的从节点加入时可以马上感知到
- 二、每个哨兵节点每隔2秒会向Redis数据节点的指定频道上发送该哨兵节点对于主节点的判断以及当前哨兵节点的信息，同时每个哨兵节点也会订阅该频道，来了解其它哨兵节点的信息及对主节点的判断，其实就是通过消息publish和subscribe来完成的
- 三、隔1秒每个哨兵根据自己info获取的级联结构信息，会向主节点、从节点及其余哨兵节点发送一次ping命令做一次心跳检测，这个也是哨兵用来判断节点是否正常的重要依据
- 四、Sentinel会以每秒一次的频率向所有与其建立了命令连接的实例（master、salve、其他Sentinel）发ping命令，通过判断ping回复是有效回复还是无效回复来判断实例是否在线/存活（对该Sentinel来说是“主观在线”），Sentinel配置文件中的down-after-milliseconds设置了判断主观下线的时间长度，如果实例在down-after-milliseconds毫秒内，返回的都是无效回复，那么Sentinel会认为该实例已（主观）下线，修改其flags状态为SRI_S_DOWN。如果多个Sentinel监视一个服务，有可能存在多个Sentinel的down-after-milliseconds配置不同，这个在实际生产中要注意（主观下线：所谓主观下线，就是单个Sentinel认为某个实例下线（有可能是接收不到订阅，之间的网络不通等等原因））
- 五、当主观下线的节点是主节点时，此时该哨兵3节点会通过指令sentinel is-masterdown-by-addr寻求其它哨兵节点对主节点的判断，如果其他的哨兵也认为主节点主观下线了，则当认为主观下线的票数超过了quorum（选举）个数，此时哨兵节点则认为该主节点确实有问题，这样就客观下线了，大部分哨兵节点都同意下线操作，也就说是客观下线，一般情况下，每个Sentinel会以每10秒一次的频率向它已知的所有主服务器和从服务器发送INFO命令，当一个主服务器被标记为客观下线时，Sentinel向下线主服务器的所有从服务器发送INFO命令的频率，会从10秒一次改为每秒一次
- 六、Sentinel和其他Sentinel协商客观下线的主节点的状态，如果处于SDOWN状态，则自动选出新的主节点，将剩余从节点指向新的主节点进行数据复制
- 新主选举原理（自动故障转移）：
```
Sentinel
```
  状态数据结构中保存了主服务的所有从服务信息，领头
```
Sentinel
```
  按照如下的规则从从服务列表中挑选出新的主服务：
  - 过滤掉主观下线的节点
  - 选择slave-priority最高的节点，如果有则返回没有就继续选择
  - 选择出复制偏移量最大的系节点，因为复制便宜量越大则数据复制的越完整，如果有就返回了，没有就继续下一步
  - 选择run_id最小的节点
  - 通过slaveof no one命令，让选出来的从节点成为主节点；并通过slaveof命令让其他节点成为其从节点
  - 将已下线的主节点设置成新的主节点的从节点，当其回复正常时，复制新的主节点，变成新的主节点的从节点，同理，当已下线的服务重新上线时，Sentinel会向其发送slaveof命令，让其成为新主的从
- 哨兵
```
lerder
```
  选举流程：如果主节点被判定为客观下线之后，就要选取一个哨兵节点来完成后面的故障转移工作，选举出一个
```
leader
```
  的流程如下：
  - 每个在线的哨兵节点都可以成为领导者，当它确认主节点下线时，会向其它哨兵发is-master-down-by-addr命令，征求判断并要求将自己设置为领导者，由领导者处理故障转移
  - 当其它哨兵收到此命令时，可以同意或者拒绝它成为领导者
  - 如果征求投票的哨兵发现自己在选举的票数大于等于num(sentinels)/2+1时，将成为领导者，如果没有超过，继续重复选举…………
- 服务下线注意事项：
  - 主观下线：单个哨兵节点认为某个节点故障时出现的情况，一般出现主观下线的节点为从节点时，不需要与其他哨兵协商，当前哨兵可直接对改节点完成下线操作
  - 客观下线：当一个节点被哨兵判定为主观下线时，这个节点是主节点，那么会和其他哨兵协商完成下线操作的情况被称为客观下线（客观下线只存在于主节点）
面试官： 那你觉得哨兵真正的实现了高可用吗？或者说你认为哨兵机制完美了嘛？
我：

刚刚在之前我提到过，哨兵解决了之前主从存在的一些问题，具体如下：
- 哨兵机制优点：
  - 解决了之前主从切换需要人工干预问题，保证了一定意义上的高可用
- 哨兵机制缺点：
  - 全量数据同步仍然会导致线上出现短暂卡顿
  - 写入QPS仍然受到主节点单机限制，对于写入并发较高的项目无法满足需求
  - 仍然存在主从复制时的木桶效应问题，存储容量受到节点群中最小内存机器限制

10.3. 代理式集群

面试官： 嗯嗯，对于类似于淘宝、新浪微博之类的互联网项目，那么怎么做到真正意义上的高可用呢？
我：

之前的哨兵并不算真正意义上的集群，只解决了人工切换问题，如果需要大规模的写入支持，或者缓存数据量巨大的情况下只能够通过加机器内存的形式来解决，但是长此已久并不是一个好的方案，而在
```
Redis3.0
```
之前官方却并没有相对应的解决方案，不过在
```
Redis3.0
```
之前却有很多其他的解决方案的提出以及落地，比如：
- TwemProxy：TwemProxy是一种代理分片机制，由Twitter开源。Twemproxy作为代理，可接受来自多个程序的访问，按照路由规则，转发给后台的各个Redis服务器，再原路返回。这个方案顺理成章地解决了单个Redis实例承载能力的问题。当然，Twemproxy本身也是单点，需要用Keepalived做高可用方案。这么些年来，Twemproxy是应用范围最广、稳定性最高、最久经考验的分布式中间件。只是，他还有诸多不方便之处。Twemproxy最大的痛点在于，无法平滑地扩容/缩容。这样增加了运维难度:业务量突增，需增加Redis服务器; 业务量菱缩，需要减少Redis服务器。但对Twemproxy而言，基本上都很难操作。或者说，Twemproxy更加像服务器端静态sharding，有时为了规避业务量突增导致的扩容需求，甚至被迫新开一个基于Twemproxy的Redis集群。Twemproxy另一个痛点是，运维不友好，甚至没有控制面板。当然，由于使用了中间件代理，相比客户端直接连服务器方式，性能上有所损耗，实测结果降低20%左右。
- Codis：Codis由豌豆英于2014年11月开源，基于Go、C开发,是近期涌现的、国人开发的优秀开源软件之一。现已广泛用于豌豆英的各种Redis业务场景，从各种压力测试来看，稳定性符合高效运维的要求。性能更是改善很多，最初比Twemproxy慢20%;现在比Twemproxy快近100% (条件:多实例，-般Value长度)。Codis具有可视化运维管理界面。Codis无疑是为解决Twemproxy缺点而出的新解决方案。因此综合方面会由于Twemproxy很多。目前也越来越多公司选择Codis，Codis引入了Group的概念,每个Group包括1个Master及至少1个Slave,这是和Twemproxy的区别之一。这样做的好处是，如果当前Master有问题，则运维人员可通过Dashboard“自助式”切换到Slave,而不需要小心翼翼地修改程序配置文件。为支持数据热迁移(AutoRebalance)，出品方修改了RedisServer源码,并称之为Codis Server，Codis采用预先分片(Pre-Sharding)机制，事先规定好了，分成1024个slots (也就是说，最多能支持后端1024个CodisServer),这些路由信息保存在ZooKeeper中。不足之处有对Redis源码进行了修改，以及代理实现本身会有的问题。
我：实则代理分片的原理也很简单，类似于代理式的分库分表的实现，之前我们是直接连接Redis，然后对Redis进行读写操作，现在则是连接代理，读写操作全部交由代理来处理分发到具体的Redis实例，而集群的组成就很好的打破了之前的一主多从架构，形成了多主多从的模式，每个节点由一个个主从来构建，每个节点存储不同的数据，每个节点都能够提供读写服务，从而做到真正意义上的高可用，具体结构如下：
面试官： 嗯，那么为什么现在一般公司在考虑技术选型的时候为什么不考虑这两种方案呢？
我：

因为使用代理之后能够去解决哨兵存在的问题，但是凡事有利必有弊，代理式集群具体情况如下：
- 优点：
  - 打破了传统的一主多从模型，允许多主存在，写入QPS不再受到单机限制
  - 数据分片存储，每个节点存储的数据都不同，解决之前主从架构存在的存容问题
  - 每个节点都是独立的主从，数据同步并不是真正的“全量”，每个节点同步数据时都只是同步该节点上master负责的一部分数据
- 缺点：
  - 由于使用了代理层来打破之前的架构模型，代理层需要承担所有工作
  - 代理层需要维护，保证高可用
  - 代理层需要实现服务动态感知、注册与监听
  - 代理层需要承载所有客户端流量
  - 代理层需要处理所有分发请求
  - 由于数据并不存在与同一台机器，Redis的很多命令不再完美支持，如set的交集、并集、差集等

10.4. 去中心化分片式集群

面试官： 那么既然代理分片式的集群存在这么多需要考虑解决的问题，现在如果让你做架设，做技术选型你会考虑那种方案呢？
我：我会考虑Redis3.x之后的Redis-cluster去中心化分片式集群，Redis-cluster在Redis3.0中推出，支持Redis分布式集群部署模式。采用无中心分布式架构。所有的Redis节点彼此互联(PING-PONG机制),内部使用二进制协议优化传输速度和带宽节点的fail是通过集群中超过半数的节点检测失效时才生效.客户端与Redis节点直连,不需要中间proxy层.客户端不需要连接集群所有节点连接集群中任何一个可用节点即可，减少了代理层，大大提高了性能。Redis-cluster把所有的物理节点映射到[0-16383]slot上,cluster负责维护node <-> slot <-> key之间的关系。目前Jedis已经支持Redis-cluster。从计算架构或者性能方面无疑Redis-cluster是最佳的选择方案。
面试官： 那你能讲讲Redis-cluster集群的原理吗？
我： Redis Cluster在设计中没有使用一致性哈希（ConsistencyHashing），而是使用数据分片（Sharding）引入哈希槽（hashSlot）来实现；一个RedisCluster包含16384（0~16383）个哈希槽，存储在RedisCluster中的所有键都会被映射到这些slot中，集群中的每个键都属于这16384个哈希槽中的一个，集群使用公式slot=CRC16（key）% 16384来计算key属于哪个槽，其中CRC16(key)语句用于计算key的CRC16校验和。集群中的每个主节点（Master）都负责处理16384个哈希槽中的一部分，当集群处于稳定状态时，每个哈希槽都只由一个主节点进行处理，每个主节点可以有一个到N个从节点（Slave），当主节点出现宕机或网络断线等不可用时，从节点能自动提升为主节点进行处理。

假设我此时向Redis发送一条命令：set name 竹子爱熊猫，那么Redis会使用CRC16算法计算KEY值，CRC16(name)，类似于一个HASH函数，完成后会得到一个数字，假设此时计算完name后得到的结果是26384，那么会拿着这个计算完成之后的结果%总槽数，26384%16384得到结果为10000，那么key=name的这个值应该被放入负责10000这个HashSlot存储，如上图中，会被放入到第三个节点存储，当再次get这个缓存时同理(Redis底层的GossIP原理由于本篇篇幅过长则不再阐述)。

十一、`Redis`版本新特性

面试官： 既然你在前面提到过这么多版本之间都有不同的变化，那么我最后考考你Redis不同的版本之间有什么区别吧
我：（心想：这不就是考我新特性吗，Redis问这么久我都扛不住了，嗓子都冒烟了）好的好的，具体如下：
- ```
Redis3.x
```
  ：
  - 支持集群
  - 在2.6基础上再次加大原子性命令支持
- ```
Redis4.x
```
  ：
  - 主从数据同步机制：4.0之前仅支持pync1，4.x之后支持psync2
  - 线程DEL/FLUSH优化，新的UNLINK与DEL作用相同，FLUSHALL/FLUSHDB中添加了ASYNC选项，Redis现在可以在不同的线程中删除后台的key而不会阻塞服务器
  - 慢日志记录客户端来源IP地址，这个小功能对于故障排查很有用处
  - 混合RDB + AOF格式
  - 新的管理命令：
    - MEMORY：能够执行不同类型的内存分析：内存问题的故障排除（使用MEMORY DOCTOR，类似于LATENCYDOCTOR）,报告单个键使用的内存量，更深入地报告Redis内存使用情况
    - SWAPDB：能够完全立即（无延迟）替换同实例下的两个Redis数据库(目前我们业务没啥用)
  - 内存使用和性能改进：
    - Redis现在使用更少的内存来存储相同数量的数据
    - Redis现在可以对使用的内存进行碎片整理，并逐渐回收空间
- ```
Redis5.x
```
  ：
  - 新的流数据类型(Stream data type)
  - 新的 Redis 模块API：定时器、集群和字典API
  - RDB可存储LFU和LRU信息
  - Redis-cli中的集群管理器从Ruby (redis-trib.rb)移植到了C语言代码。执行redis-cli --cluster help命令以了解更多信息
  - 新的有序集合(sorted set)命令：ZPOPMIN/MAX和阻塞变体(blocking variants)
  - 升级Active defragmentation至v2版本
  - 增强HyperLogLog的实现
  - 更好的内存统计报告
  - 许多包含子命令的命令现在都有一个HELP子命令
  - 优化客户端频繁连接和断开连接时，使性能表现更好
  - 升级Jemalloc至5.1版本
  - 引入CLIENT UNBLOCK和CLIENT ID
  - 新增LOLWUT命令
  - 在不存在需要保持向后兼容性的地方，弃用”master/slave”术语
  - 网络层中的差异优化
  - ```
  Lua
```
  相关的改进：
  - 将Lua脚本更好地传播到replicas / AOF
  - Lua脚本现在可以超时并在副本中进入-BUSY状态
- 引入动态的HZ(Dynamic HZ)以平衡空闲CPU使用率和响应性
- 对Redis核心代码进行了重构并在许多方面进行了改进，许多错误修复和其他方面的改进
- ```
Redis6.x
```
  ：
  - ```
  ACL
```
  ：在
```
  Redis 5
```
  版本之前，
```
  Redis
```
  安全规则只有密码控制还有通过
```
  rename
```
  来调整高危命令比如
```
  flushdb/KEYS*/shutdown
```
  等。
```
  Redis6
```
  则提供
```
  ACL
```
  的功能对用户进行更细粒度的权限控制：
  - 接入权限:用户名和密码
  - 可以执行的命令
  - 可以操作的KEY
- 新的Redis通信协议：RESP3
- Client side caching客户端缓存：基于RESP3协议实现的客户端缓存功能。为了进一步提升缓存的性能，将客户端经常访问的数据cache到客户端。减少TCP网络交互，提升RT
- IO多线程：O多线程其实指客户端交互部分的网络IO交互处理模块多线程，而非执行命令多线程。作者不想将执行命令多线程是因为要避免复杂性、锁的效率低下等等。此次支持IO多线程的设计大体如下:
- 工具支持Cluster集群：Redis6.0版本后redis/src目录下提供的大部分工具开始支持Cluster集群
- Modules API：Redis 6中模块API开发进展非常大，因为Redis Labs为了开发复杂的功能，从一开始就用上Redis模块。Redis可以变成一个框架，利用Modules来构建不同系统，而不需要从头开始写然后还要BSD许可。Redis一开始就是一个向编写各种系统开放的平台
- Disque：Disque作为一个RedisModule使用足以展示Redis的模块系统的强大。集群消息总线API、屏蔽和回复客户端、计时器、模块数据的AOF和RDB等等
面试官： 嗯嗯，小伙子你前途无量呀，今天晚上方便入职吗？
我： …

文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/94155.html

一篇了解Redis