概念基础

Redis 官网: https://redis.io/

Redis是一种支持key-value等多种数据结构的存储系统。可用于缓存，事件发布或订阅，高速队列等场景。支持网络，提供字符串，哈希，列表，队列，集合结构直接存取，基于内存，可持久化。

Redis 的特点

读写性能优异：读速度110000次/s ，写速度81000/s

数据类型丰富：Redis支持二进制 String ,Lists,hashes , Sets ,Orderer Sets数据类型

原子性

丰富的特性

持久化：支持RDB , AOF等持久化方式。

发布订阅

分布式

使用场景:

热点数据的缓存

限时业务的运用

计数器相关问题

分布式锁

延时操作

排行榜问题

redis 的 SortedSet 进行热点数据的排序

点赞、好友等互相关系的存储

Redis 集合的一些命令可以求交集，并集，差集等。比如可以实现两个用户的共同好友功能。

简单队列

Redis有list push和list pop这样的命令，所以能够很方便的执行队列操作

用法

安装

下载地址: https://redis.io/download

安装c语言编译器
yum install gcc
测试 gcc 版本
gcc --version

上传 Redis 到/opt目录

tar -zxvf redis-6.2.6.tar.gz
cd redis-6.2.6

在 redis-6.2.6 目录下, 执行 make 命令（编译指令）

如 果 没 有 准 备 好 C 语 言 编 译 环 境 ， make 会 报错—Jemalloc/jemalloc.h：没有那个文件 解决方案：运行 make distclean , 在执行 make 指令即可

执行 make install进行安装, 安装完毕安装目录在 /usr/local/bin

redis-benchmark:性能测试工具，可以在自己机器运行，看看自己机器性能如何redis-check-aof：修复有问题的 AOF 文件，rdb 和 aof 后面讲
redis-check-dump：修复有问题的 dump.rdb 文件
redis-sentinel：Redis 集群使用
redis-server：Redis 服务器启动命令
redis-cli：客户端，操作入口

配置启动

拷贝一份 redis.conf 到其他目录, 比如 /etc 目录, 注意执行保证能够定位到redis.conf

cp redis.conf /etc/redis.conf

修改/etc/redis.con 后台启动设置 daemonize no 改成 yes, 并保存退出

访问

用客户端访问：redis-cli

指定端口方式: redis-cli -p 6379

Redis 关闭：

单实例关闭：redis-cli shutdown

多实例关闭，指定端口关闭：redis-cli -p 6379 shutdown

基础操作

Redis 命令十分丰富，包括的命令组有 Cluster、Connection、Geo、Hashes、HyperLogLog、Keys、Lists、Pub/Sub、Scripting、Server、Sets、Sorted Sets、Strings、Transactions 一共14个 redis 命令组两百多个 redis 命令。

set key value : 设置 key，value 数据

get key : 根据 key 查询对应的 value，如果不存在，返回空（null）

clear : 清除屏幕中的信息

quit/exit : 退出客户端 【Redis 服务并没有结束】

help 命令名称: 获取命令帮助文档，获取组中所有命令信息名称

对Key的操作

keys * ： 查看当前库所有 key

exists key：判断某个 key 是否存在

type key ：查看你的 key 是什么类型

del key ： 删除指定的 key 数据

unlink key 根据 value 选择非阻塞删除【仅将 keys 从keyspace 元数据中删除，真正的删除会在后续异步操作】

expire key 10 ： 10 秒钟：为给定的 key 设置过期时间

ttl key 查看还有多少秒过期，-1 表示永不过期，-2 表示已过期

对DB的操作

select： 命令切换数据库，比如 select 2 是切换到 db2

dbsize：查看当前数据库的 key 的数量

flushdb：清空当前库

flushall：清空全部库

5个数据类型

redis 自身是一个 Map，其中所有的数据都是采用key : value 的形式存储。key 是字符串，value 是数据，数据支持多种类型/结构。

string

String 是 Redis 的基本数据类型，一个 key 对应一个 value。Redis 的 string 可以包含任何数据，比如jpg 图片或者序列化的对象。一个 Redis 中字符串value 最多可以是512M。

set <key><value> :添加键值对
get <key> :查询对应键值
append <key><value> :将给定的<value> 追加到原值的末尾
strlen <key> :获得值的长度
setnx <key><value> ;只有在 key 不存在时 设置 key 的值
incr <key> :将 key 中储存的数字值(字符串)增 1， 只能对数字值操作，如果为空，新增值为1
decr <key> :将 key 中储存的数字值(字符串)减 1 ， 只能对数字值操作，如果为空，新增值为-1
incrby / decrby <key><步长> :将 key 中储存的数字值增减。自定义步长5.4.2.9 mset <key1><value1><key2><value2> ....., 同时设置一个或多个 key-value 对5.4.2.10 mget <key1><key2><key3> ..... 同时获取一个或多个 value
msetnx <key1><value1><key2><value2> ..... :同时设置一个或多个key-value 对，当且仅当所有给定 key 都不存在, 原子性，有一个失败则都失败
getrange <key><起始位置><结束位置> :获得值的范围，类似 java 中的substring
setrange <key><起始位置><value> :用 <value> 覆写<key>所储存的字符串值，从<起始位置>开始(索引从 0 开始)。
setex <key><过期时间><value> :设置键值的同时，设置过期时间，单位秒。5.4.2.15 getset <key><value> , 以新换旧，设置了新值同时获得旧值

hash

Redis hash 是一个键值对集合，hash 适合用于存储对象，类似Java里面的Map

list

list 类型, 保存多个数据，底层使用双向链表存储结构实现。

lpush/rpush <key><value1><value2><value3> .... 从左边/右边插入一个或多个值
lpop/rpop <key>从左边/右边吐出一个值
rpoplpush <key1><key2>从<key1>列表右边吐出一个值，插到<key2>列表左边
lrange <key><start><stop> 按照索引下标获得元素(从左到右)
lrange mylist 0 -1 0 左边第一个，-1 右边第一个，（0-1 表示获取所有）
lindex <key><index>按照索引下标获得元素(从左到右)
llen <key>获得列表长度
linsert <key> before <value><newvalue>在<value>的前面插入<newvalue>插入值
lrem <key><n><value>从左边删除 n 个 value(从左到右)
lset<key><index><value>将列表 key 下标为 index 的值替换成 value

redis 应用于具有操作先后顺序的数据控制，比如应用在系统通知、按照时间顺序展示、将最近的通知列在前面、微信的最近转发，微博的最新关注等等。

set

set 提供的功能与 list 类似是一个列表的功能，特殊之处在于set 是可以自动排重的, 即值是不允许重复的。

sadd  ..... 将一个或多个 member 元素加入到集合key 中，已经存在的member 元素将被忽略。
smembers <key>取出该集合的所有值。
sismember <key><value>判断集合<key>是否为含有该<value>值，有 1，没有0
scard<key>返回该集合的元素个数。
srem <key><value1><value2> .... 删除集合中的某个元素。
spop <key>随机从该集合中吐出一个值。
srandmember <key><n>随机从该集合中取出 n 个值。不会从集合中删除。
smove <source><destination>value 把集合中一个值从一个集合移动到另一个集合
sinter <key1><key2>返回两个集合的交集元素。
sunion <key1><key2>返回两个集合的并集元素。
sdiff <key1><key2>返回两个集合的差集元素(key1 中的，不包含 key2 中的)

sorted_get (有序集合 Zset）

Redis 有序集合 zset 与普通集合 set 非常相似，是一个没有重复元素的字符串集合。

不同之处是有序集合的每个成员都关联了一个评分(score),这个评分(score)被用来按照从最低分到最高分的方式排序集合中的成员。集合的成员是唯一的，但是评分可以是重复了 。

因为元素是有序的, 所以也可以很快的根据评分(score)或者次序(position)来获取一个范围的元素。

访问有序集合的中间元素也是非常快的, 你能够使用有序集合作为一个没有重复成员的列表。

sorted set 常用指令&使用：

zadd … 将一个或多个 member 元素及其score 值加入到有序集 key 当中。

zrange [WITHSCORES] 返回有序集 key 中，下标在之间的元素，带 WITHSCORES，可以让分数一起和值返回到结果集

zscore 返回有序集 key 中，成员 member 的 score 值

zrangebyscore key min max [withscores] [limit offset count] 返回有序集 key 中，所有score 值介于min 和 max 之间(包括等于 min 或 max )的成员。有序集成员按 score 值递增(从小到大)次序排列。

zrevrangebyscore key max min [withscores] [limit offset count] 同上，改为从大到小排列。

zincrby 为元素的 score 加上增量

zrem 删除该集合下，指定值的元素

zcount 统计该集合，分数区间内的元素个数

zrank 返回该值在集合中的排名，从 0 开始。

Redie 配置

https://www.cnblogs.com/nhdlb/p/14048083.html#_label0

发布和订阅

Redis 发布订阅 (pub/sub) 是一种消息通信模式：发送者(pub) 发送消息，订阅者 (sub) 接收消息。

Redis 客户端可以订阅任意数量的频道，当给频道发布消息后，消息就会发送给订阅的客户端。

任务队列

就是"传递消息的队列"，与任务队列进行交互的实体有两类，一类是生产者（producer），另一类则是消费者（consumer）。生产者将需要处理的任务放入任务队列中，而消费者则不断地从任务队列中读入任务信息并执行。

从 Pub/Sub 的机制来看，它更像是一个广播系统，多个订阅者（Subscriber）可以订阅多个频道（Channel），多个发布者（Publisher）可以往多个频道（Channel）中发布消息。

Subscriber：收音机，可以收到多个频道，并以队列方式显示

Publisher：电台，可以往不同的 FM 频道中发消息

Channel：不同频率的 FM 频道

发布/订阅分类

【一个发布者，多个订阅者】

主要应用：通知、公告，可以作为消息队列或者消息管道。

【多个发布者，一个订阅者】

各应用程序作为 Publisher 向 Channel 中发送消息，Subscriber 端收到消息后执行相应的业务逻辑，比如写数据库，显示等

主要应用：排行榜、投票、计数

【多个发布者，多个订阅者】

可以向不同的 Channel 中发送消息，由不同的 Subscriber 接收。

主要应用：群聊、聊天

发布订阅操作

PUBLISH channel msg 将信息 message 发送到指定的频道 channel

SUBSCRIBE channel [channel ...] 订阅频道，可以同时订阅多个频道

UNSUBSCRIBE [channel ...] 取消订阅指定的频道, 如果不指定频道，则会取消订阅所有频道

PSUBSCRIBE pattern [pattern ...] 订阅一个或多个符合给定模式的频道，每个模式以 * 作为匹配符，比如it* 匹配所有以 it 开头的频道( it.news 、 it.blog 、 it.tweets 等等)， news.* 匹配所有以news. 开头的频道( news.it 、 news.global.today 等等)，诸如此类

PUNSUBSCRIBE [pattern [pattern ...]] 退订指定的规则, 如果没有参数则会退订所有规则

Jedis

在线文档 : https://www.mklab.cn/onlineapi/jedis/

- Java 程序操作 Redis 的工具

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

public class Jedis_ {
@Test
public void con() {
/**
* 1. 确保 ip:6379 是连通的, 需要打开防火墙的端口
* 2. 如果 redis 你设置了密码, 需要执行 jedis.auth("hspedu");进行权限验证*/
Jedis jedis = new Jedis("192.168.198.130", 6379);
jedis.auth("testedis");
String pong = jedis.ping();
System.out.println("连接成功：" + pong);
jedis.close();
}
@Test
public void key() {
Jedis jedis = new Jedis("192.168.198.130", 6379);
jedis.auth("testedis");
jedis.set("k1", "v1");
jedis.set("k2", "v2");
jedis.set("k3", "v3");
Set<String> keys = jedis.keys("*");
System.out.println(keys.size());
for (String key : keys) {
System.out.println(key);
}
System.out.println(jedis.exists("k1"));
System.out.println(jedis.ttl("k1"));
System.out.println(jedis.get("k1"));
}
@Test
public void string() {
Jedis jedis = new Jedis("192.168.198.130", 6379);
jedis.auth("testedis");
jedis.mset("k1", "v1", "k2", "v2", "k3", "v3");
System.out.println(jedis.mget("k1", "k2", "k3"));
}
@Test
public void list() {
Jedis jedis = new Jedis("192.168.198.130", 6379);
jedis.auth("testedis");
jedis.lpush("nameList", "jack", "tom", "mary");
List<String> list = jedis.lrange("nameList", 0, -1);
for (String element : list) {
System.out.println(element);
}
}
@Test
public void set() {
Jedis jedis = new Jedis("192.168.198.130", 6379);
jedis.auth("testedis");
jedis.sadd("city", "北京");
jedis.sadd("city", "上海");
jedis.sadd("city", "广州");
jedis.sadd("city", "深圳");
Set<String> smembers = jedis.smembers("city");
for (String order : smembers) {
System.out.println(order);
}
}
@Test
public void hash() {
Jedis jedis = new Jedis("192.168.198.130", 6379);
jedis.auth("testedis");
jedis.hset("hash01", "name", "李白");
System.out.println(jedis.hget("hash01", "name"));
Map<String, String> map = new HashMap<String, String>();
map.put("job", "java 工程师");
map.put("address", "北京");
map.put("email", "hello@sohu.com");
jedis.hmset("hash02", map);
List<String> result = jedis.hmget("hash02", "job", "email");
System.out.println("----------------------------");
for (String element : result) {
System.out.println(element);
}
}
@Test
public void zset() {
Jedis jedis = new Jedis("192.168.198.130", 6379);
jedis.auth("testedis");
jedis.zadd("hero", 2, "张飞");
jedis.zadd("hero", 9, "马超");
jedis.zadd("hero", 7, "赵云");
jedis.zadd("hero", 1, "关羽");
Set<String> zrange = jedis.zrange("hero", 0, -1);
for (String e : zrange) {
System.out.println(e);
}
}
}

连接 Redis 注意事项

确保 ip:6379 是连通的, 注意打开防火墙的 6379 端口

设置开放的端口号
firewall-cmd --add-port=6379/tcp --permanent ● 重启防火墙
firewall-cmd --reload
查看防火墙
firewall-cmd --list-all

如果 redis 设置了密码, 需要执行 jedis.auth("密码");进行权限验证

Spring Boot2 整合 Redis

引入依赖

<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.6.6</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
​
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<!-- 说 明 : 如 果 这 里 是 spring-boot-start 就改成如下spring-boot-start-web-->
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- redis -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!-- spring2.X 集成 redis 所需 common-pool-->
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
<!--不要带版本号,防止冲突-->
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.12.13.2.2</version>
</dependency>
</dependencies>
​
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>

配置：application.properties

#Redis 服务器地址
spring.redis.host=192.168.198.130
#Redis 服务器连接端口
spring.redis.port=6379
#Redis 如果有密码,需要配置, 没有密码就不要写
spring.redis.password=hspedu
#Redis 数据库索引（默认为 0）
spring.redis.database=0
#连接超时时间（毫秒）
spring.redis.timeout=1800000
#连接池最大连接数（使用负值表示没有限制）
spring.redis.lettuce.pool.max-active=20
#最大阻塞等待时间(负数表示没限制)
spring.redis.lettuce.pool.max-wait=-1
#连接池中的最大空闲连接
spring.redis.lettuce.pool.max-idle=5
#连接池中的最小空闲连接
spring.redis.lettuce.pool.min-idle=0

redis 配置类

\redis\config\ RedisConfig.java

@EnableCaching
@Configuration
public class RedisConfig extends CachingConfigurerSupport {
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate<String, Object> template =
                new RedisTemplate<>();
        System.out.println("template=>" + template);
        RedisSerializer<String> redisSerializer =
                new StringRedisSerializer();
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer =
                new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.activateDefaultTyping(
                LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.WRAPPER_ARRAY);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        template.setConnectionFactory(factory);
//key 序列化方式
        template.setKeySerializer(redisSerializer);
//value 序列化
        template.setValueSerializer(jackson2JsonRedisSerializer);
//value hashmap 序列化
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        return template;
    }
​
    @Bean
    public CacheManager cacheManager(RedisConnectionFactory factory) {
        RedisSerializer<String> redisSerializer =
                new StringRedisSerializer();
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = newJackson2JsonRedisSerializer(Object.class);
//解决查询缓存转换异常的问题
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.activateDefaultTyping(
                LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.WRAPPER_ARRAY);
        jackson2JsonRedisSerializer.setObjectMapper(om);
// 配置序列化（解决乱码的问题）,过期时间 600 秒
        RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
                .entryTtl(Duration.ofSeconds(600))
                .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(redi
                        sSerializer))
                .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(jackson2JsonRedisSerializer))
                .disableCachingNullValues();
        RedisCacheManager cacheManager = RedisCacheManager.builder(factory)
        cacheDefaults(config)
                .build();
        return cacheManager;
    }
}
​

注意事项和细节

如果没有提供 RedisConfig 配置类 , springboot 会使用默认配置, 也可以使用

如果没有提供 RedisConfig 配置类 , springboot 会使用默认配置, 但是会存在问题, 比如 redisTemplate 模糊查找 key 数据为空

Redis 持久化-RDB

在线文档 : https://redis.io/topics/persistence

持久化方案:

RDB（Redis DataBase）

AOF（Append Of File）

什么是 RDB?

在指定的时间间隔内将内存中的数据集快照写入磁盘，也就Snapshot 快照，恢复时将快照文件读到内存

RDB 持久化流程:

1) redis 客户端执行 bgsave 命令或者自动触发 bgsave 命令；
2) 主进程判断当前是否已经存在正在执行的子进程，如果存在，那么主进程直接返回；
3) 如果不存在正在执行的子进程，那么就 fork 一个新的子进程进行持久化数据，fork过程是阻塞的，fork 操作完成后主进程即可执行其他操作；
4) 子进程先将数据写入到临时的 rdb 文件中，待快照数据写入完成后再原子替换旧的rdb
文件；
5) 同时发送信号给主进程，通知主进程 rdb 持久化完成，主进程更新相关的统计信息
​
小结：
1) 整个过程中，主进程是不进行任何 IO 操作的，这就确保了极高的性能
2) 如果需要进行大规模数据的恢复, 且对于数据恢复的完整性不是非常敏感，那RDB方式要比 AOF 方式更加的高效
3) RDB 的缺点是最后一次持久化后的数据可能丢失
​
-如果你是正常关闭 Redis , 仍然会进行持久化, 不会造成数据丢失
-如果是 Redis 异常终止/宕机, 就可能造成数据丢失
​

Fork&Copy-On-Write

1、Fork 的作用是复制一个与当前进程一样的进程。新进程的所有数据(变量、环境变量、程序计数器等) 数值都和原进程一致，但是是一个全新的进程，并作为原进程的子进程

2、在 Linux 程序中，fork()会产生一个和父进程完全相同的子进程，但子进程在此后多会exec 系统调用，出于效率考虑，Linux 中引入了"写时复制技术即: copy-on-write" , 参考: https://blog.csdn.net/Code_beeps/article/details/92838520

3、一般情况父进程和子进程会共用同一段物理内存，只有进程空间的各段的内容要发生变化时，才会将父进程的内容复制一份给子进程。

RDB 配置

dump.rdb 文件：在 redis.conf 中配置文件名称, 默认为 dump.rdb

默认为 Redis 启动时命令行所在的目录下，进入到/usr/local/bin 目录下, 启动 Redis, 这个 ./ 就是/usr/local/bin , 如果你在/root/ 目录下启动 Redis , 那么 ./ 就是 /root/ 下了

RDB 备份&恢复

Redis 可以充当缓存, 对项目进行优化, 因此重要/敏感的数据建议在Mysql 要保存一份

从设计层面来说, Redis 的内存数据, 都是可以重新获取的(可能来自程序, 也可能来自Mysql)

Redis 启动时, 初始化数据是从dump.rdb 来的, 这个机制

RDB 持久化小结：

优势

1、适合大规模的数据恢复
2、对数据完整性和一致性要求不高更适合使用
3、节省磁盘空间
4、恢复速度快

劣势

1、虽然 Redis 在 fork 时使用了写时拷贝技术(Copy-On-Write), 但是如果数据庞大时还是比较消耗性能。
2、在备份周期在一定间隔时间做一次备份，所以如果 Redis 意外down 掉的话(如果正常关闭 Redis, 仍然会进行 RDB 备份, 不会丢失数据), 就会丢失最后一次快照后的所有修改

Redis 持久化-AOF

AOF 是什么?

1、AOF(Append Only File)
2、以日志的形式来记录每个写操作(增量保存)，将 Redis 执行过的所有写指令记录下来(比如 set/del 操作会记录, 读操作 get 不记录) [后面演示]
3、只许追加文件但不可以改写文件
4、redis 启动之初会读取该文件重新构建数据
5、redis 重启的话就根据日志文件的内容将写指令从前到后执行一次以完成数据的恢复工作

AOF持久化流程

1) 客户端的请求写命令会被 append 追加到 AOF 缓冲区内
2) AOF 缓冲区根据 AOF 持久化策略[always,everysec,no]将操作sync 同步到磁盘的AOF文件中
3) AOF 文件大小超过重写策略或手动重写时，会对 AOF 文件 rewrite 重写，压缩AOF 文件容量
4) Redis 服务重启时，会重新 load 加载 AOF 文件中的写操作达到数据恢复的目的

AOF 开启：

在 redis.conf 中配置文件名称，默认为 appendonly.aof。默认是 appendonly no, 开始是 appendonly yes

AOF 文件的保存路径，同 RDB 的路径一致。

AOF 和 RDB 同时开启，系统默认取 AOF 的数据

当开启 AOF 后, Redis 从 AOF 文件取数据

AOF 启动/修复/恢复：

AOF 的备份机制和性能虽然和 RDB 不同, 但是备份和恢复的操作同RDB 一样, 都是拷贝备份文件, 需要恢复时再拷贝到 Redis 工作目录下，启动系统即加载

正常恢复：

将有数据的 aof 文件定时备份, 需要恢复时, 复制一份保存到对应目录

恢复：重启 redis 然后重新加载

异常恢复：

1、如遇到 AOF 文件损坏，通过/usr/local/bin/redis-check-aof --fix appendonly.aof 进行恢复
2、建议先: 备份被写坏的 AOF 文件
3、恢复：重启 redis，然后重新加载

AOF 持久化小结：

优势 1、备份机制更稳健，丢失数据概率更低。 2、可读的日志文本，通过操作 AOF 稳健，可以处理误操作

劣势 1、比起 RDB 占用更多的磁盘空间 2、恢复备份速度要慢 3、每次读写都同步的话，有一定的性能压力

RDB 还是 AOF?

官方文档地址: https://redis.io/topics/persistence

官方推荐两个都启用

如果只做缓存：如果只希望数据在服务器运行的时候存在, 也可以不使用任何持久化方式

Redis事务锁机制_秒杀

Redis 的事务是什么?

1、Redis 事务是一个单独的隔离操作：事务中的所有命令都会序列化、按顺序地执行

2、事务在执行的过程中，不会被其他客户端发送来的命令请求所打断

3、Redis 事务的主要作用就是串联多个命令防止别的命令插队

Redis 事务三特性：

单独的隔离操作

1、事务中的所有命令都会序列化、按顺序地执行
2、事务在执行的过程中，不会被其他客户端发送来的命令请求所打断

没有隔离级别的概念

队列中的命令(指令), 在没有提交前都不会实际被执行

不保证原子性

事务执行过程中, 如果有指令执行失败，其它的指令仍然会被执行, 没有回滚

事务相关指令 Multi、Exec、discard

1) 从输入 Multi 命令开始，输入的命令都会依次进入命令队列中，但不会执行(类似Mysql
的 start transaction 开启事务)
2) 输入 Exec 后，Redis 会将之前的命令队列中的命令依次执行(类似Mysql 的commit 提交事务)
3) 组队的过程中可以通过 discard 来放弃组队(类似 Mysql 的rollback 回顾事务)
4) Redis 事务和 Mysql 事务本质是完全不同的, 用Mysql 的做类似说明, 是为了好理解

主从复制

上图描述了主机数据更新后, 自动同步到备机的 master/slaver 机制

Master 以写为主，Slaver 以读为主

好处:

读写分离, 提升效率 (理解: 读写分离后, 将读和写操作分布到不同的Reids, 减少单个 Redis 的压力, 提升效率

容灾快速恢复 (理解: 如果某个 slaver , 不能正常工作, 可以切换到另一个slaver)

主从复制, 要求是一主多从, 不能有多个 Master( 理解: 如果有多个主服务器Master, 那么 slaver 不能确定和哪个 Master 进行同步, 出现数据紊乱) 
要解决主服务器的高可用性, 可以使用 Redis 集群

搭建一主多从

创建目录, 并拷贝 redis.conf 到 /testredis

vi /testredis/redis.conf , 进行如下设置

关闭 appendonly no 只使用 RDB

创建 3 个文件/testredis/redis6379.conf /testredis/redis6380.conf /testredis/redis6381.conf并编辑

include /testredis/redis.conf
pidfile /var/run/redis_6379.pid
port 6379
dbfilename dump6379.rdb

启动三台 redis 服务器

redis-server redis6379.conf
redis-server redis6380.conf
redis-server redis6381.conf

连接到 3 个 Redis 服务, info replication 打印主从复制的相关信息

如果从服务器 down 了, 重新启动, 仍然可以获取 Master 的最新数据

如果主服务器 down 了, 从服务器并不会抢占为主服务器, 当主服务器恢复后, 从服务器仍然指向原来的主服务器

薪火相传

1. 上一个 Slave 可以是下一个 slave 的 Master，Slave 同样可以接收其他slaves 的连接和同步请求，那么该 slave 作为了链条中下一个的 master, 可以有效减轻master 的写压力,去中心化降低风险
2. slaveof <master_ip><master_port>
3. 风险是一旦某个 slave 宕机，后面的 slave 都没法同步
4. 主机挂了，从机还是从机，无法写数据了

反客为主

在薪火相传的结构下, 当一个 master 宕机后, 指向 Master 的slave 可以升为master, 其后面的 slave 不用做任何修改

用 slaveof no one 将从机变为主机(后面可以使用哨兵模式, 自动完成切换)

哨兵模式

哨兵模式(如图): 反客为主的自动版，能够后台监控主机是否故障，如果故障了根据投票数自动将从库转换为主库

调整为一主二仆模式，6379 带着 6380、6381
创建 /testredis/sentinel.conf
sentinel monitor redis_master 127.0.0.1 6379 1
​
说明
redis_master 为监控对象起的服务器名称
1 表示至少有多少个哨兵同意迁移的数量, 这里我配置1 表示只要有1个哨兵同意迁移就可以切换

启动哨兵, 注意看哨兵的端口是 26379，当主机挂掉，从机选举中产生新的主机，如果原来的主机重启, 会自动成为从机

集群

为什么需要集群-高可用性

1、生产环境的实际需求和问题

- 容量不够，redis 如何进行扩容？
- 并发写操作， redis 如何分摊？
- 主从模式，薪火相传模式，主机宕机，会导致 ip 地址发生变化，应用程序中配置需要修改对应的主机地址、端口等信息

redis3.0 提供解决方案-无中心化集群配置

1) 各个 Redis 服务仍然采用主从结构
2) 各个 Redis 服务是连通的, 任何一台服务器, 都可以作为请求入口3) 各个 Redis 服务器因为是连通的, 可以进行请求转发
4) 这种方式, 就无中心化集群配置, 可以看到，只需要 6 台服务器即可搞定
5) 无中心化集群配置, 还会根据 key 值, 计算 slot , 把数据分散到不同的主机, 从而缓解单个主机的存取压力
6) Redis 推荐使用无中心化集群配置
7) 在实际生成环境 各个 Redis 服务器, 应当部署到不同的机器(防止机器宕机, 主从复制失效)

集群介绍

1、Redis 集群实现了对 Redis 的水平扩容，即启动 N 个 redis 节点，将整个数据库分布存储在这 N 个节点中，每个节点存储总数据的 1/N。
2、Redis 集群通过分区（partition）来提供一定程度的可用性（availability）：即使集群中有一部分节点失效或者无法进行通讯， 集群也可以继续处理命令请求

Redis 集群搭建

搭建步骤：

将 rdb、aof 文件都删除掉

rm -rf dump63**.rdb

redis cluster 配置修改

cluster-enabled yes 打开集群模式
cluster-config-file nodes-6379.conf 设定节点配置文件名
cluster-node-timeout 15000 设定节点失联时间，超过该时间（毫秒），集群自动进行主从切换

vi /hspredis/redis6379.conf , 删除不必要的内容, 增加 cluster 配置

复制5份conf文件

使用查找替换修改另外 5 个文件

替换指令 :%s/6379/6380
vi redis6380.conf

启动 6 个 Redis 服务

redis-server redis6379.conf
redis-server redis6380.conf
redis-server redis6381.conf
redis-server redis6382.conf
redis-server redis6383.conf
redis-server redis6384.conf

将六个节点合成一个集群

将六个节点合成一个集群 的指令:
redis-cli --cluster create --cluster-replicas 1 192.168.198.130:6379192.168.198.130:6380 192.168.198.130:6381 192.168.198.130:6389192.168.198.130:6390 192.168.198.130:6391
​
-注意事项和细节
1) 组合之前，请确保所有 redis 实例启动后，nodes-xxxx.conf 文件都生成正常
2) 此处不要用 127.0.0.1, 请用真实 IP 地址, 在真实生产环境, IP 都是独立的
3) replicas 1 采用最简单的方式配置集群，一台主机，一台从机，正好三组
4) 搭建集群如果没有成功, 把 sentinel 进程 kill 掉, 再试一下
5) 分析主从对应关系

集群方式登录

指令: redis-cli -c -p 6379

指令: cluster nodes 命令查看集群信息, 主从的对应关系,

注意事项和细节
1、一个集群至少要有三个主节点
2、选项 --cluster-replicas 1 表示我们希望为集群中的每个主节点创建一个从节点。3、分配原则：尽量保证主服务器和从服务器各自运行在不同的IP 地址(机器), 防止机器故障导致主从机制失效, 高可用性得不到保障

集群的 Jedis 开发

1、即使连接的不是主机，集群会自动切换主机存储。主机写，从机读。

2、无中心化主从集群。无论从哪台主机写的数据，其他主机上都能读到数据

\jedis\JedisCluster_.java

public class JedisCluster_ {
    public static void main(String[] args) {
​
//1. 这里 set 也可以加入多个入口
//2. 因为没有配置日志, 会有提示, 但是不影响使用
//3. 如果使用集群，需要把集群相关的端口都打开,
//否则会报 No more cluster attempts lef
//4. JedisCluster 看源码有多个构造器, 也可以直接传入一个 HostAndPort

HostAndPort hostAndPort = new HostAndPort("192.168.198.130", 6379); * JedisCluster jedisCluster = new JedisCluster(hostAndPort); */
        Set<HostAndPort> set = new HashSet<HostAndPort>();
        set.add(new HostAndPort("192.168.198.130", 6379));
        JedisCluster jedisCluster = new JedisCluster(set);
        jedisCluster.set("k1", "v1");
        String k1 = jedisCluster.get("k1");
        System.out.println("k1= " + k1);
    }
}

Redis 集群的优缺点

优点 1、实现扩容 2、分摊压力 3、无中心配置相对简单。

缺点 1、多键操作是不被支持的 2、多键的 Redis 事务是不被支持的。lua 脚本不被支持 3、由于集群方案出现较晚，很多公司已经采用了其他的集群方案，而其它方案想要迁移至 redis cluster，需要整体迁移而不是逐步过渡，复杂度较大。

Redis 应用问题&解决方案

缓存穿透

缓存穿透的原因

1) key 对应的数据在数据源并不存在，每次针对此 key 的请求从缓存获取不到，请求都会压到数据源, 可能压垮数据源
2) 比如: 用一个不存在的用户 id 获取用户信息，不论缓存还是数据库都没有，若黑客利用此漏洞进行攻击可能压垮数据库
3) 也就是说：如果从存储层查不到数据则不会写入缓存，这将导致这个不存在的数据每次请求都要到存储层去查询, 失去了缓存的意义

缓存穿透的现象/表象

1) 应用服务器压力变大
2) Redis 命中率降低
3) 一直查数据库

解决方案/思路

1) 对空值缓存
如果一个查询返回的数据为空，我们仍然把这个空结果（null）进行缓存，设置空结果的过期时间应该短些，最长不超过五分钟
2) 设置可访问的名单(白名单)
定义一个可以访问的名单，每次访问和白名单的 id 进行比较，如果访问id 不在白名单里面，进行拦截，不允许访问, 比如使用 bitmaps 实现. 
3) 采用布隆过滤器
布隆过滤器可以用于检索一个元素是否在一个集合中。它的优点是空间效率和查询时间都远远超过一般的算法，缺点是有一定的误识别率和删除困难
4) 进行实时监控
当发现 Redis 的命中率开始急速降低, 需要排查访问对象和访问的数据, 和运维人员配合,
可以设置黑名单限制服务

缓存击穿

缓存击穿的原因

1) key 对应的数据存在，但在 redis 中过期，此时若有大量并发请求过来，这些请求发现缓存过期, 会从后端 DB 加载数据并回设到缓存，这时大并发的请求可能会瞬间把后端DB压垮
2) 比如某个热点数据, 可能会在某些时间点, 被超高并发地访问, 容易出现缓存击穿

缓存击穿的现象/表象

1) 数据库访问压力瞬时增加
2) Redis 里面没有出现大量 key 过期
3) Redis 正常运行状态, 但是数据库可能瘫痪了

解决方案

1) 预先设置热门数据
在 redis 高峰访问之前，把一些热门数据提前存入到 redis 里面，加大这些热门数据key的时长
2) 实时调整
现场监控哪些数据热门，实时调整 key 的过期时长
3) 使用锁

就是在缓存失效的时候（判断拿出来的值为空），不是立即去load db。-先使用缓存工具的某些带成功操作返回值的操作（比如 Redis 的SETNX）去set 一个mutexkey
当操作返回成功时，再进行 load db 的操作，并回设缓存,最后删除mutex key；-当操作返回失败，证明有线程在 load db，当前线程睡眠一段时间再重试整个get 缓存的方法
使用锁效率会有影响

缓存雪崩

缓存雪崩的原因

1) key 对应的数据存在，但在 redis 中过期，此时若有大量并发请求过来，这些请求发现缓存过期一般都会从后端 DB 加载数据并回设到缓存
2) 这个时候大并发的请求可能会瞬间把后端 DB 压垮。
3) 缓存雪崩与缓存击穿的区别在于这里针对很多 key 缓存，前者则是某一个key

缓存雪崩的现象/表象

1) 数据库访问压力变大, 服务器崩溃
2) 在极短时间内, 访问大量 Key, 而这些 Key 集中过期

解决方案/思路

1) 构建多级缓存架构
nginx 缓存 + redis 缓存 +其他缓存（ehcache 等） , 这种方式开发/维护成本较高
2) 使用锁或队列
用加锁或者队列的方式来保证不会有大量的线程对数据库一次性进行读写，从而避免失效时大量的并发请求落到底层存储系统上。不适用高并发情况
3) 设置过期标志更新缓存
记录缓存数据是否过期，如果过期会触发通知另外的线程在后台去更新实际key 的缓存。4) 将缓存失效时间分散开
比如我们可以在原有的失效时间基础上增加一个随机值，比如1-5 分钟随机，这样每一个缓存的过期时间的重复率就会降低，就很难引发集体失效的事件

分布式锁

问题：

单体单机部署的系统被演化成分布式集群系统后 ，由于分布式系统多线程、多进程并且分布在不同机器上，这将使原单机部署情况下的并发控制锁策略失效 。单纯的 Java API 并不能提供分布式锁的能力，为了解决这个问题就需要一种跨 JVM 的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题。

单体结构项目

1、单体项目,锁是在某
个项目
2、当多个请求来的
时，获取到锁，再进行
业务操作
3.这个锁可以控制的范
围只是当前项目.

1、分布式项目,锁需要控制多
个 子项目/子模块
2、当多个请求来的时，获取
到锁，再进行业务操作
3.这时锁的范围需要控制整个
分布式项目

分布式锁主流实现方案

1) 基于数据库实现分布式锁
2) 基于缓存（Redis 等）
3) 基于 Zookeeper
​
-每一种分布式锁解决方案都有各自的优缺点：
1) 性能：redis 最高
2) 可靠性：zookeeper 最高

实例： Redis 实现分布式锁-基本实现

指令: setnx key value

解读: - setnx 可以理解是上锁/加锁指令
- key 是锁的键
- value 是锁的值
- 在这个 key 没有删除前, 不能执行相同 key 的上锁指令.

指令: del key
解读
- 就是删除 key, 可以理解成就是释放锁

指令: expire key seconds
解读
- 给锁-key, 设置过期时间
- 目的是防止死锁

指令: ttl key
解读
- 查看某个锁-key, 过期时间

指令: set key value nx ex seconds
解读
- 设置锁的同时, 指定该锁的过期时间，防止死锁
- 这个指令是原子性的，防止 setnx key value / expire key seconds 两条指令, 中间执行被打断. - 过期时间到后, 会自动删除

实例： Redis 实现分布式锁-代码实现

1) 在 SpringBoot+Redis 实现分布式锁的使用
2) 获取锁, key 为 lock, 示意图

注意事项：

1、定义锁的 key, key 可以根据业务, 分别设置，比如操作某商品, key 应该是为每个sku定义的，也就是每个 sku 有一把锁
2、为了确保分布式锁可用，要确保锁的实现同时满足以下四个条件：- 互斥性。在任意时刻，只有一个客户端能持有锁。
- 不会发生死锁。即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁。
- 加锁和解锁必须是同一个客户端，A 客户端不能把 B 客户端加的锁给解了- 加锁和解锁必须具有原子性

Redis 新功能

ACL

官网：https://redis.io/topics/acl

基本介绍

1、Redis ACL 是 Access Control List（访问控制列表）的缩写，该功能根据可以执行的命令和可以访问的键来限制某些连接。

2、在 Redis 5 版本之前，Redis 安全规则只有密码控制 还有通过rename 来调整高危命令比如 flushdb ， KEYS* ， shutdown 等。

3、Redis 6 则提供 ACL 的功能对用户进行更细粒度的权限控制：-接入权限:用户名和密码 -可以执行的命令 -可以操作的 KEY

常用指令：

acl list 命令展现用户权限列表

acl cat 命令查看添加权限指令类别

acl whoami 命令查看当前用户

acl setuser 命令创建和编辑用户 ACL