Redis 6 笔记

NoSQL概论

NoSQL全称Not Only SQL,它是一种非关系型的数据库,相比传统SQL关系型数据库,它有以下特点

不保证关系数据的ACID特性
不遵循SQL标准
消除数据间相关性

优势

性能远超传统关系型数据库
灵活易于扩展
数据模型灵活
高可用

=> 海量高并发数据解决方案

NoSQL数据库分为以下几种

键值存储数据库: 所有数据都以键值方式存储,类似于HashMap,使用简单,性能高
列存储数据库: 通常用来应对分布式存储的海量数据; 一个键会指向多个列
文档型数据库: 以一种特定的文档格式存储数据;
图形数据库: 用类似于图的数据结构存储数据,结合相关算法实现高速访问

Redis数据库就是一个开源的键值存储数据库,所有数据都存放在内存中,性能远远高于磁盘IO, 并且支持数据持久化,横向扩展,主从复制等等

实际生产中,需要将MySQL和Redis在不同情况下搭配使用

基本操作

与MySQL不同,Redis不具有像MySQL那样严格的表结构; 它是一个键值数据库,因此,可以用像Map一样的操作方式,通过键值对向Redis数据库中添加数据

在Redis下,数据库是由一个整数索引标示,而不是一个数据库名称; 默认使用0号数据库, 数据库总数可以通过配置文件来修改,默认16个

选择数据库

可以通过select语句切换数据库

1	select 序号;

数据操作

向Redis数据库中添加数据：

1
2
3

set <key> <value>
-- 一次性多个
mset [<key> <value>]...

所有存入的数据默认会以字符串的形式保存，键值具有一定的命名规范，以方便我们可以快速定位我们的数据属于哪一个部分，比如用户的数据：

1 2	-- 使用冒号来进行板块分割，比如下面表示用户XXX的信息中的name属性，值为lbw set user:info:用户ID:name lbw

我们可以通过键值获取存入的值：

get <key>

你以为Redis就仅仅只是存取个数据吗？它还支持数据的过期时间设定：

1 2	set <key> <value> EX 秒 set <key> <value> PX 毫秒

当数据到达指定时间时，会被自动删除。我们也可以单独为其他的键值对设置过期时间：

1	expire <key> 秒

通过下面的命令来查询某个键值对的过期时间还剩多少：

ttl <key>
-- 毫秒显示
pttl <key>
-- 转换为永久
persist <key>

那么当我们想直接删除这个数据时呢？直接使用：

1	del <key>...

删除命令可以同时拼接多个键值一起删除。

当我们想要查看数据库中所有的键值时：

keys *

也可以查询某个键是否存在：

1	exists <key>...

还可以随机拿一个键：

randomkey

我们可以将一个数据库中的内容移动到另一个数据库中：

1	move <key> 数据库序号

修改一个键为另一个键：

1
2
3

rename <key> <新的名称>
-- 下面这个会检查新的名称是否已经存在
renamex <key> <新的名称>

如果存放的数据是一个数字，我们还可以对其进行自增自减操作：

-- 等价于a = a + 1
incr <key>
-- 等价于a = a + b
incrby <key> b
-- 等价于a = a - 1
decr <key>

最后就是查看值的数据类型：

1	type <key>

Redis数据库也支持多种数据类型，但是它更偏向于我们在Java中认识的那些数据类型。

数据类型介绍

一个键值对除了存储一个String类型的值以外，还支持多种常用的数据类型。

Hash

这种类型本质上就是一个HashMap，也就是嵌套了一个HashMap罢了，在Java中就像这样：

#Redis默认存String类似于这样：
Map<String, String> hash = new HashMap<>();
#Redis存Hash类型的数据类似于这样：
Map<String, Map<String, String>> hash = new HashMap<>();

它比较适合存储类这样的数据，由于值本身又是一个Map，因此我们可以在此Map中放入类的各种属性和值，以实现一个Hash数据类型存储一个类的数据。

我们可以像这样来添加一个Hash类型的数据：

1	hset <key> [<字段> <值>]...

我们可以直接获取：

1
2
3

hget <key> <字段>
-- 如果想要一次性获取所有的字段和值
hgetall <key>

同样的，我们也可以判断某个字段是否存在：

1	hexists <key> <字段>

删除Hash中的某个字段：

1	hdel <key>

我们发现，在操作一个Hash时，实际上就是我们普通操作命令前面添加一个h，这样就能以同样的方式去操作Hash里面存放的键值对了，这里就不一一列出所有的操作了。我们来看看几个比较特殊的。

我们现在想要知道Hash中一共存了多少个键值对：

1	hlen <key>

我们也可以一次性获取所有字段的值：

1	hvals <key>

唯一需要注意的是，Hash中只能存放字符串值，不允许出现嵌套的的情况。

List

我们接着来看List类型，实际上这个猜都知道，它就是一个列表，而列表中存放一系列的字符串，它支持随机访问，支持双端操作，就像我们使用Java中的LinkedList一样。

我们可以直接向一个已存在或是不存在的List中添加数据，如果不存在，会自动创建：

-- 向列表头部添加元素
lpush <key> <element>...
-- 向列表尾部添加元素
rpush <key> <element>...
-- 在指定元素前面/后面插入元素
linsert <key> before/after <指定元素> <element>

同样的，获取元素也非常简单：

-- 根据下标获取元素
lindex <key> <下标>
-- 获取并移除头部元素
lpop <key>
-- 获取并移除尾部元素
rpop <key>
-- 获取指定范围内的
lrange <key> start stop

注意下标可以使用负数来表示从后到前数的数字（Python：搁这儿抄呢是吧）:

1 2	-- 获取列表a中的全部元素 lrange a 0 -1

没想到吧，push和pop还能连着用呢：

1 2	-- 从前一个数组的最后取一个数出来放到另一个数组的头部，并返回元素 rpoplpush 当前数组目标数组

它还支持阻塞操作，类似于生产者和消费者，比如我们想要等待列表中有了数据后再进行pop操作：

1
2

-- 如果列表中没有元素，那么就等待，如果指定时间（秒）内被添加了数据，那么就执行pop操作，如果超时就作废，支持同时等待多个列表，只要其中一个列表有元素了，那么就能执行
blpop <key>... timeout

Set和SortedSet

Set集合其实就像Java中的HashSet一样（我们在JavaSE中已经学习过了，HashSet本质上就是利用了一个HashMap，但是Value都是固定对象，仅仅是Key不同）它不允许出现重复元素，不支持随机访问，但是能够利用Hash表提供极高的查找效率。

向Set中添加一个或多个值：

1	sadd <key> <value>...

查看Set集合中有多少个值：

1	scard <key>

判断集合中是否包含：

-- 是否包含指定值
sismember <key> <value>
-- 列出所有值
smembers <key>

集合之间的运算：

-- 集合之间的差集
sdiff <key1> <key2>
-- 集合之间的交集
sinter <key1> <key2>
-- 求并集
sunion <key1> <key2>
-- 将集合之间的差集存到目标集合中
sdiffstore 目标 <key1> <key2>
-- 同上
sinterstore 目标 <key1> <key2>
-- 同上
sunionstore 目标 <key1> <key2>

移动指定值到另一个集合中：

1	smove <key> 目标 value

移除操作：

-- 随机移除一个幸运儿
spop <key>
-- 移除指定
srem <key> <value>...

那么如果我们要求Set集合中的数据按照我们指定的顺序进行排列怎么办呢？这时就可以使用SortedSet，它支持我们为每个值设定一个分数，分数的大小决定了值的位置，所以它是有序的。

我们可以添加一个带分数的值：

1	zadd <key> [<value> <score>]...

同样的：

-- 查询有多少个值
zcard <key>
-- 移除
zrem <key> <value>...
-- 获取区间内的所有
zrange <key> start stop

由于所有的值都有一个分数，我们也可以根据分数段来获取：

-- 通过分数段查看
zrangebyscore <key> start stop [withscores] [limit]
-- 统计分数段内的数量
zcount <key>  start stop
-- 根据分数获取指定值的排名
zrank <key> <value>

https://www.jianshu.com/p/32b9fe8c20e1

有关Bitmap、HyperLogLog和Geospatial等数据类型，这里暂时不做介绍，感兴趣可以自行了解。

持久化

Redis的数据存放在内存中,问题来了,如果突然断电,数据不是丢失了吗?

这个时候就需要引入持久化机制了,持久化机制可以把数据备份到硬盘上

持久化的实现方式有两种方案：

一种是直接保存当前已经存储的数据，相当于复制内存中的数据到硬盘上，需要恢复数据时直接读取即可；
还有一种就是保存我们存放数据的所有过程，需要恢复数据时，只需要将整个过程完整地重演一遍就能保证与之前数据库中的内容一致。

RDB(定时达量保存)

相当于第一种方案,就是将数据本身保存到硬盘

1 2	save -- 在保存期间会占用一定的时间,也可以单独开一个线程来进行保存操作 bgsave -- 后台保存

执行后，会在服务端目录下生成一个dump.rdb文件，而这个文件中就保存了内存中存放的数据，当服务器重启后，会自动加载里面的内容到对应数据库中。保存后我们可以关闭服务器：

shutdown

重启后可以看到数据依然存在。

虽然这种方式非常方便，但是由于会完整复制所有的数据，如果数据库中的数据量比较大，那么复制一次可能就需要花费大量的时间，所以我们可以每隔一段时间自动进行保存；还有就是，如果我们基本上都是在进行读操作，而没有进行写操作，实际上只需要偶尔保存一次即可，因为数据几乎没有怎么变化，可能两次保存的都是一样的数据。

我们可以在配置文件中设置自动保存，并设定在一段时间内写入多少数据时，执行一次保存操作：

1 2	save 300 10 # 300秒（5分钟）内有10个写入 save 60 10000 # 60秒（1分钟）内有10000个写入

配置的save使用的都是bgsave后台执行。

AOF(定时写日志)

虽然RDB能够很好地解决数据持久化问题，但是它的缺点也很明显：每次都需要去完整地保存整个数据库中的数据，同时后台保存过程中也会产生额外的内存开销，最严重的是它并不是实时保存的，如果在自动保存触发之前服务器崩溃，那么依然会导致少量数据的丢失。

而AOF就是另一种方式，它会以日志的形式将我们每次执行的命令都进行保存，服务器重启时会将所有命令依次执行，通过这种重演的方式将数据恢复，这样就能很好解决实时性存储问题。

rdb和aof区别

但是，我们多久写一次日志呢？我们可以自己配置保存策略，有三种策略：

always：每次执行写操作都会保存一次
everysec：每秒保存一次（默认配置），这样就算丢失数据也只会丢一秒以内的数据
no：看系统心情随缘保存

可以在配置文件中配置：

# 注意得改成yes
appendonly yes

# appendfsync always
appendfsync everysec
# appendfsync no

重启服务器后，可以看到服务器目录下多了一个appendonly.aof文件，存储的就是我们执行的命令。

AOF的缺点也很明显，每次服务器启动都需要进行过程重演，相比RDB更加耗费时间，并且随着我们的操作变多，不断累计，可能到最后我们的aof文件会变得无比巨大，我们需要一个改进方案来优化这些问题。

Redis有一个AOF重写机制进行优化,可以将多条语句融合成一条，比如我们执行了这样的语句：

1
2
3

lpush test 666
lpush test 777
lpush test 888

实际上用一条语句也可以实现：

1	lpush test 666 777 888

正是如此，只要我们能够保证最终的重演结果和原有语句的结果一致，无论语句如何修改都可以，所以我们可以通过这种方式将多条语句进行压缩。

我们可以输入命令来手动执行重写操作：

1	bgrewriteaof

或是在配置文件中配置自动重写：

# 百分比计算，这里不多介绍
auto-aof-rewrite-percentage 100
# 当达到这个大小时，触发自动重写
auto-aof-rewrite-min-size 64mb

至此，我们就完成了两种持久化方案的介绍，最后我们再来进行一下总结：

AOF：
- 优点：存储速度快、消耗资源少、支持实时存储
- 缺点：加载速度慢、数据体积大
RDB：
- 优点：加载速度快、数据体积小
- 缺点：存储速度慢大量消耗资源、会发生数据丢失

事务和锁机制

和MySQL一样，在Redis中也有事务机制，当我们需要保证多条命令一次性完整执行而中途不受到其他命令干扰时，就可以使用事务机制。

我们可以使用命令来直接开启事务：

multi

当我们输入完所有要执行的命令时，可以使用命令来立即执行事务：

exec

我们也可以中途取消事务：

discard

实际上整个事务是创建了一个命令队列，它不像MySQL那种在事务中也能单独得到结果，而是我们提前将所有的命令装在队列中，但是并不会执行，而是等我们提交事务的时候再统一执行。

锁

又提到锁了，实际上这个概念已经不算是陌生了。实际上在Redis中也会出现多个命令同时竞争同一个数据的情况，比如现在有两条命令同时执行，他们都要去修改a的值，那么这个时候就只能动用锁机制来保证同一时间只能有一个命令操作。

虽然Redis中也有锁机制，但是它是一种乐观锁，不同于MySQL，我们在MySQL中认识的锁是悲观锁，那么什么是乐观锁什么是悲观锁呢？

悲观锁：时刻认为别人会来抢占资源，禁止一切外来访问，直到释放锁，具有强烈的排他性质。
乐观锁：并不认为会有人来抢占资源，所以会直接对数据进行操作，在操作时再去验证是否有其他人抢占资源。

Redis中可以使用watch来监视一个目标，如果执行事务之前被监视目标发生了修改，则取消本次事务：

watch

我们可以开两个客户端进行测试。

取消监视可以使用：

unwatch

至此，Redis的基础内容就讲解完毕了，在之前学过的SpringCloud中, 已经了解了集群相关的知识，包括主从复制、哨兵模式等。

通过Java与Redis交互

既然了解了如何通过命令窗口操作Redis数据库，那么我们如何使用Java来操作呢？

这里我们需要使用到Jedis框架，它能够实现Java与Redis数据库的交互，依赖：

<dependencies>
    <dependency>
        <groupId>redis.clients</groupId>
        <artifactId>jedis</artifactId>
        <version>4.0.0</version>
    </dependency>
</dependencies>

基本操作

我们来看看如何连接Redis数据库，非常简单，只需要创建一个对象即可：

public static void main(String[] args) {
    //创建Jedis对象
    Jedis jedis = new Jedis("localhost", 6379);
  	
  	//使用之后关闭连接
  	jedis.close();
}

通过Jedis对象，我们就可以直接调用命令的同名方法来执行Redis命令了，比如：

public static void main(String[] args) {
    //直接使用try-with-resouse，省去close
    try(Jedis jedis = new Jedis("192.168.10.3", 6379)){
        jedis.set("test", "lbwnb");   //等同于 set test lbwnb 命令
        System.out.println(jedis.get("test"));  //等同于 get test 命令
    }
}

Hash类型的数据也是这样：

public static void main(String[] args) {
    try(Jedis jedis = new Jedis("192.168.10.3", 6379)){
        jedis.hset("hhh", "name", "sxc");   //等同于 hset hhh name sxc
        jedis.hset("hhh", "sex", "19");    //等同于 hset hhh age 19
        jedis.hgetAll("hhh").forEach((k, v) -> System.out.println(k+": "+v));
    }
}

我们接着来看看列表操作：

public static void main(String[] args) {
    try(Jedis jedis = new Jedis("192.168.10.3", 6379)){
        jedis.lpush("mylist", "111", "222", "333");  //等同于 lpush mylist 111 222 333 命令
        jedis.lrange("mylist", 0, -1)
                .forEach(System.out::println);    //等同于 lrange mylist 0 -1
    }
}

实际上我们只需要按照对应的操作去调用同名方法即可，所有的类型封装Jedis已经帮助我们完成了。

SpringBoot整合Redis

我们接着来看如何在SpringBoot项目中整合Redis操作框架，只需要一个starter即可，但是它底层没有用Jedis，而是Lettuce：

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

starter提供的默认配置会去连接本地的Redis服务器，并使用0号数据库，当然你也可以手动进行修改：

spring:
  redis:
  	#Redis服务器地址
    host: 192.168.10.3
    #端口
    port: 6379
    #使用几号数据库
    database: 0

starter已经给我们提供了两个默认的模板类：

@Configuration(
    proxyBeanMethods = false
)
@ConditionalOnClass({RedisOperations.class})
@EnableConfigurationProperties({RedisProperties.class})
@Import({LettuceConnectionConfiguration.class, JedisConnectionConfiguration.class})
public class RedisAutoConfiguration {
    public RedisAutoConfiguration() {
    }

    @Bean
    @ConditionalOnMissingBean(
        name = {"redisTemplate"}
    )
    @ConditionalOnSingleCandidate(RedisConnectionFactory.class)
    public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<Object, Object> template = new RedisTemplate();
        template.setConnectionFactory(redisConnectionFactory);
        return template;
    }

    @Bean
    @ConditionalOnMissingBean
    @ConditionalOnSingleCandidate(RedisConnectionFactory.class)
    public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory redisConnectionFactory) {
        return new StringRedisTemplate(redisConnectionFactory);
    }
}

那么如何去使用这两个模板类呢？我们可以直接注入StringRedisTemplate来使用模板：

@SpringBootTest
class SpringBootTestApplicationTests {

    @Autowired
    StringRedisTemplate template;

    @Test
    void contextLoads() {
        ValueOperations<String, String> operations = template.opsForValue();
        operations.set("c", "xxxxx");   //设置值
        System.out.println(operations.get("c"));   //获取值
      	
        template.delete("c");    //删除键
        System.out.println(template.hasKey("c"));   //判断是否包含键
    }

}

实际上所有的值的操作都被封装到了ValueOperations对象中，而普通的键操作直接通过模板对象就可以使用了，大致使用方式其实和Jedis一致。

我们接着来看看事务操作，由于Spring没有专门的Redis事务管理器，所以只能借用JDBC提供的，只不过无所谓，正常情况下反正我们也要用到这玩意：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
</dependency>

@Service
public class RedisService {

    @Resource
    StringRedisTemplate template;

    @PostConstruct
    public void init(){
        template.setEnableTransactionSupport(true);   //需要开启事务
    }

    @Transactional    //需要添加此注解
    public void test(){
        template.multi();
        template.opsForValue().set("d", "xxxxx");
        template.exec();
    }
}

我们还可以为RedisTemplate对象配置一个Serializer来实现对象的JSON存储：

@Test
void contextLoad2() {
    //注意Student需要实现序列化接口才能存入Redis
    template.opsForValue().set("student", new Student());
    System.out.println(template.opsForValue().get("student"));
}

Redis与分布式

在SpringBoot阶段，我们学习了Redis，它是一个基于内存的高性能数据库，我们当时已经学习了包括基本操作、常用数据类型、持久化、事务和锁机制以及使用Java与Redis进行交互等，利用它的高性能，我们还使用它来做Mybatis的二级缓存、以及Token的持久化存储。而这一部分，我们将继续深入，探讨Redis在分布式开发场景下的应用。

主从复制

在分布式场景下，我们可以考虑让Redis实现主从模式：

主从复制，是指将一台Redis服务器的数据，复制到其他的Redis服务器。前者称为主节点(Master)，后者称为从节点(Slave)，数据的复制是单向的，只能由主节点到从节点。Master以写为主，Slave 以读为主。

这样的好处肯定是显而易见的：

实现了读写分离，提高了性能。
在写少读多的场景下，我们甚至可以安排很多个从节点，这样就能够大幅度的分担压力，并且就算挂掉一个，其他的也能使用。

那么我们现在就来尝试实现一下，这里我们还是在Windows下进行测试，打开Redis文件夹，我们要开启两个Redis服务器，修改配置文件redis.windows.conf：

1
2
3

# Accept connections on the specified port, default is 6379 (IANA #815344).
# If port 0 is specified Redis will not listen on a TCP socket.
port 6001

一个服务器的端口设定为6001，复制一份，另一个的端口为6002，接着我们指定配置文件进行启动，打开cmd：

现在我们的两个服务器就启动成功了，接着我们可以使用命令查看当前服务器的主从状态，我们打开客户端：

输入info replication命令来查看当前的主从状态，可以看到默认的角色为：master，也就是说所有的服务器在启动之后都是主节点的状态。那么现在我们希望让6002作为从节点，通过一个命令即可：

可以看到，在输入replicaof 127.0.0.1 6001命令后，就会将6001服务器作为主节点，而当前节点作为6001的从节点，并且角色也会变成：slave，接着我们来看看6001的情况：

可以看到从节点信息中已经出现了6002服务器，也就是说现在我们的6001和6002就形成了主从关系（还包含一个偏移量，这个偏移量反应的是从节点的同步情况）

主服务器和从服务器都会维护一个复制偏移量，主服务器每次向从服务器中传递 N 个字节的时候，会将自己的复制偏移量加上 N。从服务器中收到主服务器的 N 个字节的数据，就会将自己额复制偏移量加上 N，通过主从服务器的偏移量对比可以很清楚的知道主从服务器的数据是否处于一致，如果不一致就需要进行增量同步了。

那么我们现在可以来测试一下，在主节点新增数据，看看是否会同步到从节点：

可以看到，我们在6001服务器插入的a，可以在从节点6002读取到，那么，从节点新增的数据在主节点能得到吗？我们来测试一下：

可以看到，从节点压根就没办法进行数据插入，节点的模式为只读模式。那么如果我们现在不想让6002作为6001的从节点了呢？

可以看到，通过输入replicaof no one，即可变回Master角色。接着我们再来启动一台6003服务器，流程是一样的：

可以看到，在连接之后，也会直接同步主节点的数据，因此无论是已经处于从节点状态还是刚刚启动完成的服务器，都会从主节点同步数据，实际上整个同步流程为：

从节点执行replicaof ip port命令后，从节点会保存主节点相关的地址信息。
从节点通过每秒运行的定时任务发现配置了新的主节点后，会尝试与该节点建立网络连接，专门用于接收主节点发送的复制命令。
连接成功后，第一次会将主节点的数据进行全量复制，之后采用增量复制，持续将新来的写命令同步给从节点。

当我们的主节点关闭后，从节点依然可以读取数据：

但是从节点会疯狂报错：

当然每次都去敲个命令配置主从太麻烦了，我们可以直接在配置文件中配置，添加这样行即可：

1	replicaof 127.0.0.1 6001

这里我们给6002和6003服务器都配置一下，现在我们重启三个服务器。

当然，除了作为Master节点的从节点外，我们还可以将其作为从节点的从节点，比如现在我们让6003作为6002的从节点：

也就是说，现在差不多是这样的的一个情况：

采用这种方式，优点肯定是显而易见的，但是缺点也很明显，整个传播链路一旦中途出现问题，那么就会导致后面的从节点无法及时同步。

哨兵模式

前面我们讲解了Redis实现主从复制的一些基本操作，那么我们接着来看哨兵模式。

经过之前的学习，我们发现，实际上最关键的还是主节点，因为一旦主节点出现问题，那么整个主从系统将无法写入，因此，我们得想一个办法，处理一下主节点故障的情况。实际上我们可以参考之前的服务治理模式，比如Nacos和Eureka，所有的服务都会被实时监控，那么只要出现问题，肯定是可以及时发现的，并且能够采取响应的补救措施，这就是我们即将介绍的哨兵：

注意这里的哨兵不是我们之前学习SpringCloud Alibaba的那个，是专用于Redis的。哨兵会对所有的节点进行监控，如果发现主节点出现问题，那么会立即让从节点进行投票，选举一个新的主节点出来，这样就不会由于主节点的故障导致整个系统不可写（注意要实现这样的功能最小的系统必须是一主一从，再小的话就没有意义了）

那么怎么启动一个哨兵呢？我们只需要稍微修改一下配置文件即可，这里直接删除全部内容，添加：

1	sentinel monitor lbwnb 127.0.0.1 6001 1

其中第一个和第二个是固定，第三个是为监控对象名称，随意，后面就是主节点的相关信息，包括IP地址和端口，最后一个1我们暂时先不说，然后我们使用此配置文件启动服务器，可以看到启动后：

可以看到以哨兵模式启动后，会自动监控主节点，然后还会显示那些节点是作为从节点存在的。

现在我们直接把主节点关闭，看看会发生什么事情：

可以看到从节点还是正常的在报错，一开始的时候不会直接重新进行选举而是继续尝试重连（因为有可能只是网络小卡一下，没必要这么敏感），但是我们发现，经过一段时间之后，依然无法连接，哨兵输出了以下内容：

可以看到哨兵发现主节点已经有一段时间不可用了，那么就会开始进行重新选举，6003节点被选为了新的主节点，并且之前的主节点6001变成了新的主节点的从节点：

当我们再次启动6001时，会发现，它自动变成了6003的从节点，并且会将数据同步过来：

那么，这个选举规则是怎样的呢？是在所有的从节点中随机选取还是遵循某种规则呢？

首先会根据优先级进行选择，可以在配置文件中进行配置，添加replica-priority配置项（默认是100），越小表示优先级越高。
如果优先级一样，那就选择偏移量最大的
要是还选不出来，那就选择runid（启动时随机生成的）最小的。

要是哨兵也挂了咋办？没事，咱们可以多安排几个哨兵，只需要把哨兵的配置复制一下，然后修改端口，这样就可以同时启动多个哨兵了，我们启动3个哨兵（一主二从三哨兵），这里我们吧最后一个值改为2：

1	sentinel monitor lbwnb 192.168.0.8 6001 2

这个值实际上代表的是当有几个哨兵认为主节点挂掉时，就判断主节点真的挂掉了

现在我们把6001节点挂掉，看看这三个哨兵会怎么样：

可以看到都显示将master切换为6002节点了。

那么，在哨兵重新选举新的主节点之后，我们Java中的Redis的客户端怎么感知到呢？我们来看看，首先还是导入依赖：

<dependencies>
    <dependency>
        <groupId>redis.clients</groupId>
        <artifactId>jedis</artifactId>
        <version>4.2.1</version>
    </dependency>
</dependencies>

public class Main {
    public static void main(String[] args) {
        //这里我们直接使用JedisSentinelPool来获取Master节点
        //需要把三个哨兵的地址都填入
        try (JedisSentinelPool pool = new JedisSentinelPool("lbwnb",
                new HashSet<>(Arrays.asList("192.168.0.8:26741", "192.168.0.8:26740", "192.168.0.8:26739")))) {
            Jedis jedis = pool.getResource();   //直接询问并得到Jedis对象，这就是连接的Master节点
            jedis.set("test", "114514");    //直接写入即可，实际上就是向Master节点写入

            Jedis jedis2 = pool.getResource();   //再次获取
            System.out.println(jedis2.get("test"));   //读取操作
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

这样，Jedis对象就可以通过哨兵来获取，当Master节点更新后，也能得到最新的。

集群搭建

如果我们服务器的内存不够用了，但是现在我们的Redis又需要继续存储内容，那么这个时候就可以利用集群来实现扩容。

因为单机的内存容量最大就那么多，已经没办法再继续扩展了，但是现在又需要存储更多的内容，这时我们就可以让N台机器上的Redis来分别存储各个部分的数据（每个Redis可以存储1/N的数据量），这样就实现了容量的横向扩展。同时每台Redis还可以配一个从节点，这样就可以更好地保证数据的安全性。

那么问题来，现在用户来了一个写入的请求，数据该写到哪个节点上呢？我们来研究一下集群的机制：

首先，一个Redis集群包含16384个插槽，集群中的每个Redis 实例负责维护一部分插槽以及插槽所映射的键值数据，那么这个插槽是什么意思呢？

实际上，插槽就是键的Hash计算后的一个结果，注意这里出现了计算机网络中的CRC循环冗余校验，这里采用CRC16，能得到16个bit位的数据，也就是说算出来之后结果是0-65535之间，再进行取模，得到最终结果：

Redis key的路由计算公式：slot = CRC16（key） % 16384

结果的值是多少，就应该存放到对应维护的Redis下，比如Redis节点1负责0-25565的插槽，而这时客户端插入了一个新的数据a=10，a在Hash计算后结果为666，那么a就应该存放到1号Redis节点中。简而言之，本质上就是通过哈希算法将插入的数据分摊到各个节点的，所以说哈希算法真的是处处都有用啊。

那么现在我们就来搭建一个简单的Redis集群，这里创建6个配置，注意开启集群模式：

# Normal Redis instances can't be part of a Redis Cluster; only nodes that are
# started as cluster nodes can. In order to start a Redis instance as a
# cluster node enable the cluster support uncommenting the following:
#
cluster-enabled yes

接着记得把所有的持久化文件全部删除，所有的节点内容必须是空的。

然后输入redis-cli.exe --cluster create --cluster-replicas 1 127.0.0.1:6001 127.0.0.1:6002 127.0.0.1:6003 127.0.0.1:7001 127.0.0.1:7002 127.0.0.1:7003，这里的--cluster-replicas 1指的是每个节点配一个从节点：

输入之后，会为你展示客户端默认分配的方案，并且会询问你当前的方案是否合理。可以看到6001/6002/6003都被选为主节点，其他的为从节点，我们直接输入yes即可：

最后分配成功，可以看到插槽的分配情况：

现在我们随便连接一个节点，尝试插入一个值：

在插入时，出现了一个错误，实际上这就是因为a计算出来的哈希值（插槽），不归当前节点管，我们得去管这个插槽的节点执行，通过上面的分配情况，我们可以得到15495属于节点6003管理：

在6003节点插入成功，当然我们也可以使用集群方式连接，这样我们无论在哪个节点都可以插入，只需要添加-c表示以集群模式访问：

可以看到，在6001节点成功对a的值进行了更新，只不过还是被重定向到了6003节点进行插入。

我们可以输入cluster nodes命令来查看当前所有节点的信息：

那么现在如果我们让某一个主节点挂掉会怎么样？现在我们把6001挂掉：

可以看到原本的6001从节点7001，晋升为了新的主节点，而之前的6001已经挂了，现在我们将6001重启试试看：

可以看到6001变成了7001的从节点，那么要是6001和7001都挂了呢？

这时我们尝试插入新的数据：

可以看到，当存在节点不可用时，会无法插入新的数据，现在我们将6001和7001恢复：

可以看到恢复之后又可以继续正常使用了。

最后我们来看一下如何使用Java连接到集群模式下的Redis，我们需要用到JedisCluster对象：

public class Main {
    public static void main(String[] args) {
        //和客户端一样，随便连一个就行，也可以多写几个，构造方法有很多种可以选择
        try(JedisCluster cluster = new JedisCluster(new HostAndPort("192.168.0.8", 6003))){
            System.out.println("集群实例数量："+cluster.getClusterNodes().size());
            cluster.set("a", "yyds");
            System.out.println(cluster.get("a"));
        }
    }
}

操作基本和Jedis对象一样，这里就不多做赘述了。

学习视频:

Redis 缓存技术已完结（2021版本）4K蓝光画质+杜比音效从内卷到开摆