*★如果文章对你有帮助【关注点赞❤️收藏⭐】一起努力！★*
六 个人主页：花棉袄
本章内容：【Redis：基础篇】学习视频推荐官方文档
✍ 版权: 本文由【花棉袄】原创在CSDN首发需要转载请联系博主
⛩️Redis简介
16个数据库
-
Redis有16个数据库，索引【0，15】
-
默认连接使用的是一个索引为【0】的数据库
-
可使用select 进行切换
本章内容：【Redis：基础篇】学习视频推荐官方文档
✍ 版权: 本文由【花棉袄】原创在CSDN首发需要转载请联系博主
16个数据库
Redis有16个数据库，索引【0，15】
默认连接使用的是一个索引为【0】的数据库
可使用select 进行切换
单线程+多路复用
-
多路复用是指使用一个线程来检查多个文件描述符（Socket）的就绪状态
-
比如调用 select 和 poll 函数，传入多个文件描述符，如果有一个文件描述符就绪，则返回，否则阻塞直到超时
-
得到就绪状态后进行真正的操作可以在同一个线程里执行，也可以启动线程执行（比如使用线程池）
☃️Redis安装
- 请参考：Docker部署系列：Redis安装
️数据类型 1️⃣Redis键常用操作 2️⃣字符串
- String类型是二进制安全的：Redis的string可以包含任何数据
- 一个Redis中字符串的value最多可以是512M
常用操作
- set [key] [value]：赋值 - setnx [key] [value]：当不存在时设置值 - append [key] [value]：追加值 - mset [key1] [value1] [key2] [value2]：批量赋值 - msetnx [key1] [value1] [key2] [value2]：当不存在时设置值，批量赋值--一旦一个存在，所有都失败 - setrange [key] [start] [value]：从start处开始赋值 - setex [key] [expire] [value]：赋值&设置有效期 - get [key]：取值 - mget [key1] [key2]：批量取值 - getset [key] [value]：取老值赋新值 - strlen [key]：获取长度 - getrange [key] [start] [end]：获取子字符串--含头也含尾 - incr [key]：值自增1，仅可对整数操作 - decr [key]：值自减1，仅可对整数操作 - incrby [key] [step]：值自增step，仅可对整数操作 - decrby [key] [step]：值自减step，仅可对整数操作
数据结构
-
String 的数据结构为简单动态字符串 (Simple Dynamic String, 缩写 SDS)，是可以修改的字符串，内部结构实现上类似于 Java 的 ArrayList，采用预分配冗余空间的方式来减少内存的频繁分配.
-
如图中所示，内部为当前字符串实际分配的空间 capacity 一般要高于实际字符串长度 len
-
当字符串长度小于 1M 时，扩容都是加倍现有的空间，如果超过 1M，扩容时一次只会多扩 1M 的空间
- 需要注意的是字符串最大长度为 512M
常用操作
- lpush [listname] [value] [value]：从左侧添加元素 - rpush [listname] [value] [value]：从右侧添加元素 - lrange [listname] [start] [end]：遍历获取元素（从左到右） - lindex [listname] [index]：获取索引位置元素 - lset [listname] [index] [value]：设置索引处元素值 - llen [listname]：获取列表长度 - lpop [listname]：从左侧抛出元素（元素全部删除后该列表也删除） - rpop [listname]：从右侧抛出元素（元素全部删除后该列表也删除） - lrem [listname] [count] [value]：从左侧删除指定元素（删除count个） - rpoplpush [listname1] [listname2]：从key1右边吐出一个值插入到key2的左边 - linsert [listname1] [before|after] [value] [value]：在指定元素前插入元素（只插入一个）
数据结构
-
List的数据结构为快速链表quickList
-
首先在列表元素较少的情况下会使用一块连续的内存存储，这个结构是ziplist，也即是压缩列表
-
它将所有的元素紧挨着一起存储，分配的是一块连续的内存
-
当数据量比较多的时候才会改成quicklist
-
因为普通的链表需要的附加指针空间太大，会比较浪费空间
-
比如这个列表里存的只是int类型的数据，结构上还需要两个额外的指针prev和next
常用操作
- sadd [setname] [value] [value]：集合添加添加元素 - srem [setname] value]：删除元素（元素全部删除后该集合也删除） - spop [setname]：随机抛出元素 - smembers [setname]：获取全部元素 - sismember [setname] value]：判断元素是否在集合中 - scard [setname]：集合数量 - srandmember [setname] [count]：随机获取元素 - sinter [setname1] [setname2] [setname3]：获取交集 - sunion [setname1] [setname2] [setname3]：获取并集 - sdiff [setname1] [setname2] [setname3]：获取差集
数据结构
-
底层结构是dict字典，字典使用哈希表实现的
-
Java中的Set的内部实现是用HashMap，只不过所有的value都指向了同一个对象
-
Redis的set结构也是一样，它的内部结构也是hash，所有的value都指向了同一个内部值
常用操作
- hset [hname] [key] [value]：赋值 - hget [hname] [key]：取值 - hkeys [hname]：所有键值 - hvals [hname]：获取所有值value - hlen [hname]：容量 - hexists [hname] key]：是否存在 - hincrby [hname] [key] [len]：为某个键增加步长 - hsetnx [hname] [key] [value]：将hname中的key的值设置成value，当且仅当key不存在的时候设置
数据结构
-
Hash类型对应的数据结构是两种：ziplist（压缩列表） hashtable（哈希表）
-
当field-value长度较短且个数较少时，使用ziplist，否则使用hashtable
常用操作
- zadd [zsetname] [score] [key]：添加元素 - zrem [zsetname] [key]：移除元素 - zrange [zsetname] [start] [end]：遍历元素 - zange [zsetname] [start] [end] withscores：遍历元素（展示分值） - zrangebyscore [zsetname] [start] [end]：按分值排序（从小到大） - zrangebyscore [zsetname] [start] [end] withscores：按分值排序 - zrevrangebyscore [zsetname] [start] [end ]：按分值倒排（从大到小） - zrank [zsetname] [key]：获取元素排名 - zrank [zsetname] [key] withscores：获取元素排名 - zincrby [zsetname] [step] [key]：增加元素分值
数据结构
有序链表
- 要查找值为51的元素，需要从第一个元素开始依次查找、比较才能找到。共需要6次比较
跳跃表
-
从第2层开始，1节点比51节点小，向后比较； 21节点比51节点小，继续向后比较，后面就是NULL了，所以从21节点向下到第1层
-
在第1层，41节点比51节点小，继续向后，61节点比51节点大，所以从41向下
-
在第0层，51节点为要查找的节点，节点被找到，共查找4次
️配置文件详解 1️⃣网络配置 2️⃣通用配置 3️⃣限制配置
发布与订阅
-
订阅端订阅渠道消息，一旦有发布端往渠道中推送消息，订阅端能立即接收到消息
-
订阅信息
subscribe channel1
- 发布信息
publish channel1 hello
Java中使用 1️⃣Jedis使用
引入依赖
redis.clients jedis 3.3.0
Jedis使用
public class JedisDemo1 {
public static void main(String[] args) {
Jedis jedis = new Jedis("127.0.0.1", 6379);
String ping = jedis.ping();
System.out.println(ping);
}
@Test
public void demo() {
Jedis jedis = new Jedis("127.0.0.1", 6379);
String s = jedis.flushDB();
System.out.println(s);
}
@Test
public void demo1() {
Jedis jedis = new Jedis("127.0.0.1", 6379);
Set keys = jedis.keys("*");
System.out.println(keys);
}
@Test
public void demo2() {
Jedis jedis = new Jedis("127.0.0.1", 6379);
String set = jedis.set("name", "xiang");
String name = jedis.get("name");
System.out.println(name);
jedis.mset("k1", "v1", "k2", "v2");
List mGet = jedis.mget("k1", "k2");
System.out.println(mGet);
}
@Test
public void demo3() {
Jedis jedis = new Jedis("127.0.0.1", 6379);
Long lpush = jedis.lpush("list", "v1", "v2", "v3");
System.out.println(lpush);
List list = jedis.lrange("list", 0, -1);
System.out.println(list);
}
@Test
public void demo4() {
Jedis jedis = new Jedis("127.0.0.1", 6379);
jedis.sadd("family", "xiang");
jedis.sadd("family", "yuan");
Set set = jedis.smembers("family");
System.out.println(set);
}
@Test
public void demo5() {
Jedis jedis = new Jedis("127.0.0.1", 6379);
jedis.hset("desc", "price", "12");
jedis.hset("desc", "height", "140");
jedis.hincrBy("desc", "price", 12);
Map desc = jedis.hgetAll("desc");
System.out.println(desc);
}
@Test
public void demo6() {
Jedis jedis = new Jedis("127.0.0.1", 6379);
jedis.zadd("city", 90, "nanjing");
jedis.zadd("city", 95, "shanghai");
jedis.zadd("city", 99, "beijing");
Set city = jedis.zrangeByScoreWithScores("city", 0, 100);
System.out.println(city);
}
}
2️⃣SpringBoot整合Redis
RedisTemplate：操作不同数据类型方法
引入依赖
org.springframework.boot spring-boot-starter-data-redis
配置文件
spring.redis.host=127.0.0.1 spring.redis.port=6379 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
代码测试
@Slf4j
@RestController
public class MyController {
@Autowired
private RedisTemplate redisTemplate;
@GetMapping("test")
public String test(){
redisTemplate.opsForValue().set("name1","xiang");
String name1 = redisTemplate.opsForValue().get("name1");
log.info("name:{}", name1);
return name1;
}
}
- 使用客户端连接Redis查看键值信息
按需自选配置
@EnableCaching
@Configuration
public class RedisConfig extends CachingConfigurerSupport {
@Bean
public RedisTemplate redisTemplate(RedisConnectionFactory factory) {
RedisTemplate template = new RedisTemplate<>();
RedisSerializer redisSerializer = new StringRedisSerializer();
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);
template.setConnectionFactory(factory);
template.setKeySerializer(redisSerializer);
template.setValueSerializer(jackson2JsonRedisSerializer);
template.setHashValueSerializer(jackson2JsonRedisSerializer);
return template;
}
@Bean
public CacheManager cacheManager(RedisConnectionFactory factory) {
RedisSerializer redisSerializer = new StringRedisSerializer();
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);
RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig().entryTtl(Duration.ofSeconds(600))
.serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(jackson2JsonRedisSerializer))
.disableCachingNullValues();
RedisCacheManager cacheManager = RedisCacheManager.builder(factory)
.cacheDefaults(config)
.build();
return cacheManager;
}
}
事务和锁机制 1️⃣事务三大特性
-
单独的隔离操作 ：事务中的所有命令都会序列化、按顺序地执行。事务在执行的过程中，不会被其他客户端发送来的命令请求所打断
-
没有隔离级别的概念：队列中的命令没有提交之前都不会实际被执行，因为事务提交前任何指令都不会被实际执行
-
不保证原子性：事务中如果有一条命令执行失败，其后的命令仍然会被执行，没有回滚
-
Redis 事务是一个单独的隔离操作：事务中的所有命令都会序列化、按顺序地执行
-
事务在执行的过程中，不会被其他客户端发送来的命令请求所打断
-
Redis的三个命令需要搭配使用：multi exec disc!
-
主要的三个命令：multi、exec和discard
-
组队阶段某个命令出现了报告错误，执行时整个的所有队列都会被取消
- 执行阶段某个命令出现错误，则只有报错的命令不会执行，其他的命令都会执行，不会回滚
入队成功
127.0.0.1:6379> multi OK 127.0.0.1:6379(TX)> set k1 v1 QUEUED 127.0.0.1:6379(TX)> set k2 v2 QUEUED 127.0.0.1:6379(TX)> exec 1) OK 2) OK
入队失败
127.0.0.1:6379> multi OK 127.0.0.1:6379(TX)> set k1 v1 QUEUED 127.0.0.1:6379(TX)> set k2 (error) ERR wrong number of arguments for 'set' command 127.0.0.1:6379(TX)> exec (error) EXECABORT Transaction discarded because of previous errors.3️⃣事务冲突问题
-
想想一个场景：有很多人有你的账户，同时去参加双十一抢购
一个请求想给金额减 8000
一个请求想给金额减 5000
一个请求想给金额减 1000 -
最终我们可以发现，总共金额是 10000，如果请求全部执行，那最后的金额变为 - 4000，很明显不合理
-
悲观锁 (Pessimistic Lock)，每次去拿数据的时候都认为别人会修改
-
所以每次在拿数据的时候都会上锁，这样别人想拿这个数据就会 block 直到它拿到锁
-
传统的关系型数据库里边就用到了很多这种锁机制
-
比如行锁，表锁等，读锁，写锁等，都是在做操作之前先上锁
-
乐观锁 (Optimistic Lock)，每次去拿数据的时候都认为别人不会修改
-
所以不会上锁，但是在更新的时候会判断一下在此期间别人有没有去更新这个数据，可以使用版本号等机制
-
乐观锁适用于多读的应用类型，这样可以提高吞吐量
-
Redis 就是利用这种 check-and-set 机制实现事务的
watch key [key …]
- 在执行 multi 之前，先执行 watch key1 [key2]，可以监视一个 (或多个) key
- 如果在事务执行之前这个 (或这些) key 被其他命令所改动，那么事务将被打断
预先设置值
set balance 100
打开两个客户端同时监视balance值
127.0.0.1:6379> watch balance OK 127.0.0.1:6379> multi OK 127.0.0.1:6379(TX)> incrby balance 10 QUEUED 127.0.0.1:6379(TX)> exec 1) (integer) 110
127.0.0.1:6379> watch balance OK 127.0.0.1:6379> multi OK 127.0.0.1:6379(TX)> incrby balance 10 QUEUED 127.0.0.1:6379(TX)> exec (nil)
unwatch：取消 WATCH 命令对所有 key 的监视
- EXEC 命令或 DISCARD 命令先被执行了的话那么就不需要再执行 UNWATCH 了
- 利用乐观锁解决超卖问题，监视库存键
RDB持久化
- 在指定的时间间隔内将内存中的数据集快照写入磁盘， 也就是Snapshot快照
- 它恢复时是将快照文件直接读到内存里
持久化执行流程
- Redis 会单独创建（fork）一个子进程来进行持久化
- 首先会将数据写入到一个临时文件中
- 待持久化过程都结束了，再用这个临时文件替换上次持久化好的文件
- 整个过程中，主进程是不进行任何 IO 操作的，这就确保了极高的性能
Fork 的作用
-
Fork 的作用是复制一个与当前进程一样的进程
-
新进程的所有数据（变量、环境变量、程序计数器等） 数值都和原进程一致
-
但是是一个全新的进程，并作为原进程的子进程
-
在 Linux 程序中，fork () 会产生一个和父进程完全相同的子进程
-
但子进程在此之后多会 exec 系统调用，出于效率考虑，Linux 中引入了 “写时复制技术”
-
一般情况父进程和子进程会共用同一段物理内存，只有进程空间的各段的内容要发生变化时，才会将父进程的内容复制一份给子进程
RDB优点
- 适合大规模的数据恢复
- 对数据完整性和一致性要求不高更适合使用
- 节省磁盘空间
- 恢复速度快
RDB缺点
- Fork的时候，内存中的数据被克隆了一份，大致2倍的膨胀性需要考虑
- 虽然Redis在fork时使用了写时拷贝技术，但是如果数据庞大时还是比较消耗性能
- 在备份周期在一定间隔时间做一次备份，所以如果Redis意外down掉的话，就会丢失最后一次快照后的所有修改
有关的配置
-
在redis.conf中配置文件名称，默认为dump.rdb
-
配置文件中默认的快照配置
- 动态停止RDB：redis-cli config set save “” #save后给空值，表示禁用保存策略
rdb的备份
1.先通过config get dir 查询rdb文件的目录 2.将*.rdb的文件拷贝下来 3.rdb的恢复：将文件名改回原来的名字 4.关闭Redis 5.先把备份的文件拷贝到工作目录下 cp dump2.rdb dump.rdb 6.启动Redis,备份数据会直接加载2️⃣AOF持久化
AOF持久化
- 以日志的形式来记录每个写操作（增量保存），将Redis执行过的所有写指令记录下来(读操作不记录)
- 只许追加文件但不可以改写文件，redis启动之初会读取该文件重新构建数据
- redis 重启的话就根据日志文件的内容将写指令从前到后执行一次以完成数据的恢复工作
AOF持久化流程
- 客户端的请求写命令会被append追加到AOF缓冲区内
- AOF缓冲区根据AOF持久化策略[always,everysec,no]将操作sync同步到磁盘的AOF文件中
- AOF文件大小超过重写策略或手动重写时，会对AOF文件rewrite重写，压缩AOF文件容量
- Redis服务重启时，会重新load加载AOF文件中的写操作达到数据恢复的目的
AOF的开启
- 可以在redis.conf中配置文件名称，默认为 appendonly.aof
# appendfsync always ： 每一次操作都进行持久化 appendfsync everysec ： 每隔一秒进行一次持久化 # appendfsync no ： 不进行持久化
- AOF和RDB同时开启，系统默认取AOF的数据（数据不会存在丢失）
正常恢复
- 修改默认的appendonly no，改为yes
- 把有数据的aof文件复制一份保存到对应目录(查看目录：config get dir)
- 恢复：重启redis然后重新加载
异常恢复
- 修改默认的appendonly no，改为yes
- 如遇到AOF文件损坏，使用redis-check-aof–fix appendonly.aof 命令进行恢复
- 备份被写坏的AOF文件
- 恢复：重启redis，然后重新加载
重写机制
- 是指上就是把rdb 的快照，以二级制的形式附在新的aof头部，作为已有的历史数据，替换掉原来的流水账操作
- AOF采用文件追加方式，文件会越来越大为避免出现此种情况，新增了重写机制
- 当AOF文件的大小超过所设定的阈值时，Redis就会启动AOF文件的内容压缩， 只保留可以恢复数据的最小指令集
- 可以使用命令bgrewriteaof
触发机制，何时重写
-
Redis会记录上次重写时的AOF大小，默认配置是当AOF文件大小是上次rewrite后大小的一倍且文件大于64M时触发
-
重写虽然可以节约大量磁盘空间，减少恢复时间，但是每次重写还是有一定的负担的，因此设定Redis要满足一定条件才会进行重写
-
auto-aof-rewrite-percentage：设置重写的基准值，文件达到100%时开始重写（文件是原来重写后文件的2倍时触发）
-
auto-aof-rewrite-min-size：设置重写的基准值，最小文件64MB。达到这个值开始重写
AOF优点
-
备份机制更稳健，丢失数据概率更低
-
可读的日志文本，通过操作AOF
AOF缺点
- 比起RDB占用更多的磁盘空间
- 恢复备份速度要慢
- 每次读写都同步的话，有一定的性能压力
- 存在个别Bug，造成恢复不能
主从复制概述
- 主机数据更新后根据配置和策略， 自动同步到备机的master/slaver机制，Master以写为主，Slave以读为主
主从复制作用
- 读写分离，性能扩展
- 容灾快速恢复
- 在根目录下创建文件夹myredis，把redis的配置文件复制过来，要把aof持久化关掉
- 创建三个文件redis6379.conf、redis6381.conf、redis6380.conf内容如下
######################redis6379.conf####################### include /myredis/redis.conf pidfile /var/run/redis_6379.pid port 6379 dbfilename dump6379.rdb
######################redis6380.conf####################### include /myredis/redis.conf pidfile /var/run/redis_6380.pid port 6380 dbfilename dump6380.rdb
######################redis6381.conf####################### include /myredis/redis.conf pidfile /var/run/redis_6381.pid port 6381 dbfilename dump6381.rdb
- 同时启动三个端口的redis
- 查看服务是否启动
- 查看三台主机运行情况
info replication # 打印主从复制的相关信息
master和slave互联
- 在从机上配置主机的：ip和端口
- 在6380和6381上执行: slaveof 127.0.0.1 6379
slaveof
- 在主机上写，在从机上可以读取数据，在从机上写数据报错
- 主机挂掉，重启就行，一切如初，从机重启需重设：slaveof 127.0.0.1 6379
slave1、slave2是从头开始复制还是从切入点开始复制
- 比如从k4进来，那之前的k1,k2,k3是否也可以复制
- 从机会全量复制主机的内容，k1,k2,k3,k4都会复制
主机shutdown后情况如何？从机是上位还是原地待命
- 主机shutdown后，从机原地待命，等待主机重新启动，一切回复正常
主机又回来了后，主机新增记录，从机还能否顺利复制
- 可以复制，因为主机重启后和之前一样，主机写内容，会同步到从机中
其中一台从机down后情况如何？依照原有它能跟上大部队吗？
- 从机down后，会脱离大部队，如果重新启动，还想同步主机内容的话，需要重新执行命令：slaveof
-
Slave启动成功连接到master后会发送一个sync命令
-
Master接到命令启动后台的存盘进程，同时收集所有接收到的用于修改数据集命令， 在后台进程执行完毕之后，master将传送整个数据文件到slave，以完成一次完全同步
-
全量复制：而slave服务在接收到数据库文件数据后，将其存盘并加载到内存中
-
增量复制：Master继续将新的所有收集到的修改命令依次传给slave，完成同步
-
但是只要是重新连接master，一次完全同步（全量复制)将被自动执行
-
上一个Slave可以是下一个slave的Master，Slave同样可以接收其他 slaves的连接和同步请求，那么该slave作为了链条中下一个的master，可以有效减轻master的写压力,去中心化降低风险
-
用 slaveof：中途变更转向：会清除之前的数据，重新建立拷贝最新的
-
风险是一旦某个slave宕机，后面的slave都没法备份
-
主机挂了，从机还是从机，无法写数据了
-
当一个master宕机后，后面的slave可以立刻升为master，其后面的slave不用做任何修改
-
用slaveof no one 将从机变为主机
哨兵模式
- 哨兵(sentinel) 是一个分布式系统，用于对主从结构中的每台服务器进行监控，当出现故障时通过投票机制选择新的master并将所有slave连接到新的master
- 能够后台监控主机是否故障，如果故障了根据投票数自动将从库转换为主库
哨兵的作用
-
监控：监控master和slave
不断的检查master和slave是否正常运行
master存活检测、master与slave运行情况检测
-
通知（提醒）：当被监控的服务器出现问题时，向其他（哨兵间，客户端）发送通知
-
自动故障转移：断开master与slave连接，选取一个slave作为master，将其他slave连接新的master，并告知客户端新的服务器地址
-
注意：哨兵也是一台redis服务器，只是不提供数据相关服务，通常哨兵的数量配置为单数
搭建哨兵模式
- 先搭建一主二从的环境
- 自定义的/myredis目录下新建sentinel.conf文件，名字绝不能错
哨兵模式配置信息
- 在sentinel.conf文件中添加
- 设置哨兵监听的主服务器信息， sentinel_number表示参与投票的哨兵数量
sentinel monitor master_name master_host master_port sentinel_number
- 设置判定服务器宕机时长，该设置控制是否进行主从切换
sentinel down-after-milliseconds master_name million_seconds
- 设置故障切换的最大超时时
sentinel failover-timeout master_name million_seconds
- 设置主从切换后，同时进行数据同步的slave数量
- 数值越大，要求网络资源越高，数值越小，同步时间越长
sentinel parallel-syncs master_name sync_slave_number
- 启动哨兵
redis-sentinel filename
复制延时
- 由于所有的写操作都是先在Master上操作，然后同步更新到Slave上
- 所以从Master同步到Slave机器有一定的延迟
- 当系统很繁忙的时候，延迟问题会更加严重
- Slave机器数量的增加也会使这个问题更加严重
故障恢复
-
优先级在redis.conf中默认：slave-priority 100，值越小优先级越高
-
偏移量是指获得原主机数据最全的
-
每个redis实例启动后都会随机生成一个40位的runid
-
Redis 集群实现了对Redis的水平扩容，即启动N个redis节点，将整个数据库分布存储在这N个节点中，每个节点存储总数据的1/N
-
Redis 集群通过分区（partition）来提供一定程度的可用性（availability）： 即使集群中有一部分节点失效或者无法进行通讯， 集群也可以继续处理命令请求
集群作用：
- 分散单台服务器的访问压力，实现负载均衡
- 分散单台服务器的存储压力，实现可扩展性
- 降低单台服务器宕机带来的业务灾难
-
通过算法设计，计算出key应该保存的位置
-
将所有的存储空间计划切割成16384份，每台主机保存一部分
注意：每份代表的是一个存储空间，不是一个key的保存空间
-
将key按照计算出的结果放到对应的存储空间
1️⃣ 制作6个实例
- 主：6379,6380,6381 - 从：6389,6390,6391
- 删除文件夹中的全部持久化文件rdb或者aof
2️⃣ 新建六个配置文件
- 内容如下：(除了端口号不一样，其他都一样)
include /myredis/redis.conf pidfile "/var/run/redis_6391.pid" port 6391 dbfilename "dump6391.rdb" cluster-enabled yes cluster-config-file nodes-6391.conf cluster-node-timeout 15000
3️⃣启动6个服务
redis-server redis_6379.conf redis-server redis_6380.conf redis-server redis_6381.conf
4️⃣ 将六个节点合成一个集群
-
组合之前，请确保所有redis实例启动后，nodes-xxxx.conf文件都生成正常
-
先到redis的src目录中
cd /opt/redis-6.2.1/src
- 运行集成集群命令
redis-cli --cluster create --cluster-replicas 1 192.168.242.110:6379 192.168.242.110:6380 192.168.242.110:6381 192.168.242.110:6389 192.168.242.110:6390 192.168.242.110:6391
- 查看是否集成成功
# 连接Redis redis-cli -c -p 6379 # 查看集群信息 cluster nodes
- 查看集群节点信息
- 是否启用cluster，加入cluster节点
cluster-enabled yes|no
- cluster配置文件名，该文件属于自动生成，仅用于快速查找文件并查询文件内容
cluster-config-file filename
- 节点服务响应超时时间，用于判定该节点是否下线或切换为从节点
cluster-node-timeout milliseconds
- master连接的slave最小数量
cluster-migration-barrier min_slave_number5️⃣集群节点操作
- 查看集群节点信息
cluster nodes
- 更改slave指向新的master
cluster replicate master-id
- 发现一个新节点，新增master
cluster meet ip:port
- 忽略一个没有solt的节点
cluster forget server_id
- 手动故障转移
cluster failover6️⃣集群操作命令
- 添加master到当前集群中，连接时可以指定任意现有节点地址与端口
redis-cli --cluster add-node new-master-host:new-master-port now-host:now-port
- 添加slave到master中
redis-cli --cluster add-node new-slave-host:new-slave-port master-host:master-port --cluster-slave --cluster-master-id masterid
- 删除节点，如果删除的节点是master，必须保障其中没有槽slot
redis-cli --cluster del-node del-slave-host:del-slave-port del-slave-id
- 重新分槽，分槽是从具有槽的master中划分一部分给其他master，过程中不创建新的槽
redis-cli --cluster reshard new-master-host:new-master:port --cluster-from src- master-id1, src-master-id2, src-master-idn --cluster-to target-master-id -- cluster-slots slots
注意：将需要参与分槽的所有masterid不分先后顺序添加到参数中，使用，分隔
指定目标得到的槽的数量，所有的槽将平均从每个来源的master处获取
- 重新分配槽，从具有槽的master中分配指定数量的槽到另一个master中，常用于清空指定master中的槽
redis-cli --cluster reshard src-master-host:src-master-port --cluster-from src- master-id --cluster-to target-master-id --cluster-slots slots --cluster-yes7️⃣故障恢复
- 如果主节点下线？从节点能否自动升为主节点？注意：15秒超时
- 主节点恢复后，主从关系会如何？主节点回来变成从机
- 如果所有某一段插槽的主从节点都宕掉，redis服务是否还能继续?
- cluster-require-full-coverage 为yes：整个集群 都挂掉 - cluster-require-full-coverage 为no：该 插槽数据 全都不能使用，也无法存储，其他插槽依然可以存储
- redis.conf中的参数 cluster-require-full-coverage
-
即使连接的不是主机，集群会自动切换主机存储。主机写，从机读
-
无中心化主从集群。无论从哪台主机写的数据，其他主机上都能读到数据
public class JedisClusterTest {
public static void main(String[] args) {
HostAndPort hostAndPort = new HostAndPort("192.168.242.110", 6381);
JedisCluster jedisCluster = new JedisCluster(hostAndPort);
jedisCluster.set("k5","v5");
String k5 = jedisCluster.get("k5");
System.out.println(k5);
}
}
數应用问题
1️⃣缓存穿透
问题描述
- key对应的数据在数据源并不存在，每次针对此key的请求从缓存获取不到，请求都会压到数据源，从而可能压垮数据源
- 比如用一个不存在的用户id获取用户信息，不论缓存还是数据库都没有，若黑客利用此漏洞进行攻击可能压垮数据库
问题排查
- Redis中大面积出现未命中
- 出现非正常URL访问
问题分析
- 获取的数据在数据库中也不存在，数据库查询未得到对应数据
- Redis获取到null数据未进行持久化，直接返回
- 出现黑客攻击服务器
解决方案
-
对空值缓存：如果一个查询返回的数据为空（不管是数据是否不存在），仍然把这个空结果（null）进行缓存，设置空结果的过期时间会很短，最长不超过五分钟
-
设置可访问的名单（白名单）：使用bitmaps类型定义一个可以访问的名单，名单id作为bitmaps的偏移量，每次访问和bitmap里面的id进行比较，如果访问id不在bitmaps里面，进行拦截，不允许访问
-
进行实时监控：当发现的命中率开始急速降低，需要排查访问对象和访问的数据，和运维人员配合，可以设置黑名单限制服务
问题描述
- key对应的数据存在，但在redis中过期，此时若有大量并发请求过来
- 这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存
- 这个时候大并发的请求可能会瞬间把后端DB压垮。
解决方案
-
预先设置热门数据：在redis高峰访问之前，把一些热门数据提前存入到redis里面，加大这些热门数据key的时长
-
实时调整：现场监控哪些数据热门，实时调整key的过期时长
问题描述
-
key对应的数据存在，但在redis中过期，此时若有大量并发请求过来
-
这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存
-
这个时候大并发的请求可能会瞬间把后端DB压垮
-
缓存雪崩：大量key过期
-
缓存穿透：某个key过期
解决方案
-
构建多级缓存架构：nginx缓存 + redis缓存 +其他缓存（ehcache等）
-
使用锁或队列：用加锁或者队列的方式保证来保证不会有大量的线程对数据库一次性进行读写，从而避免失效时大量的并发请求落到底层存储系统上。不适用高并发情况
-
设置过期标志更新缓存：记录缓存数据是否过期（设置提前量），如果过期会触发通知另外的线程在后台去更新实际key的缓存
-
将缓存失效时间分散开：比如我们可以在原有的失效时间基础上增加一个随机值，比如1-5
- 跨JVM的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题
分布式锁主流的实现方案
- 基于数据库实现分布式锁
- 基于缓存（Redis等）–性能最高
- 基于Zookeeper --可靠性更高
set key value nx ex time # 设置锁并且给锁设置过期时间 del key # 删除锁
java代码实现
@GetMapping("testLock")
public void testLock(){
//1获取锁，setne ,顺便设置过期时间
Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", "111",3,TimeUnit.SECONDS);
//2获取锁成功、查询num的值
if(lock){
Object value = redisTemplate.opsForValue().get("num");
//2.1判断num为空return
if(StringUtils.isEmpty(value)){
return;
}
//2.2有值就转成成int
int num = Integer.parseInt(value+"");
//2.3把redis的num加1
redisTemplate.opsForValue().set("num", ++num);
//2.4释放锁，del
redisTemplate.delete("lock");
}else{
//3获取锁失败、每隔0.1秒再获取
try {
Thread.sleep(100);
testLock();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
3️⃣UUID防止误删
解决方案
-
第一步：uuid表示不同的操作：set look uuid nx ex 10
-
第二步：释放锁的时候，首先判断当前uuid和要释放的uuid是否相同
@GetMapping("testLock")
public void testLock(){
String uuid = UUID.randomUUID().toString();
//1获取锁，setne ,顺便设置过期时间
Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", uuid,3,TimeUnit.SECONDS);
//2获取锁成功、查询num的值
if(lock){
...
String lockUuid = (String)redisTemplate.opsForValue().get("lock");
if(uuid.equals(lockUuid)){
//2.4释放锁，del
redisTemplate.delete("lock");
}
}else{
...
}
}
4️⃣LUA保证原子性
解决方案：使用lua脚本保证删除的原子性
// 定义lua 脚本
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
// 使用redis执行lua执行
DefaultRedisScript redisScript = new DefaultRedisScript<>();
redisScript.setScriptText(script);
// 设置一下返回值类型 为Long
// 因为删除判断的时候，返回的0,给其封装为数据类型。如果不封装那么默认返回String 类型，
// 那么返回字符串与0 会有发生错误。
redisScript.setResultType(Long.class);
// 第一个要是script 脚本 ，第二个需要判断的key，第三个就是key所对应的值。
redisTemplate.execute(redisScript, Arrays.asList(locKey), uuid);
简单示例
@GetMapping("testLockLua")
public void testLockLua() {
//1 声明一个uuid ,将做为一个value 放入我们的key所对应的值中
String uuid = UUID.randomUUID().toString();
//2 定义一个锁：lua 脚本可以使用同一把锁，来实现删除！
String skuId = "25"; // 访问skuId 为25号的商品 100008348542
String locKey = "lock:" + skuId; // 锁住的是每个商品的数据
// 3 获取锁
Boolean lock = redisTemplate.opsForValue().setIfAbsent(locKey, uuid, 3, TimeUnit.SECONDS);
// 第一种： lock 与过期时间中间不写任何的代码。
// redisTemplate.expire("lock",10, TimeUnit.SECONDS);//设置过期时间
// 如果true
if (lock) {
// 执行的业务逻辑开始
// 获取缓存中的num 数据
Object value = redisTemplate.opsForValue().get("num");
// 如果是空直接返回
if (StringUtils.isEmpty(value)) {
return;
}
// 不是空 如果说在这出现了异常！ 那么delete 就删除失败！ 也就是说锁永远存在！
int num = Integer.parseInt(value + "");
// 使num 每次+1 放入缓存
redisTemplate.opsForValue().set("num", String.valueOf(++num));
// 定义lua 脚本
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
// 使用redis执行lua执行
DefaultRedisScript redisScript = new DefaultRedisScript<>();
redisScript.setScriptText(script);
// 设置一下返回值类型 为Long
// 因为删除判断的时候，返回的0,给其封装为数据类型。如果不封装那么默认返回String 类型，
// 那么返回字符串与0 会有发生错误。
redisScript.setResultType(Long.class);
// 第一个要是script 脚本 ，第二个需要判断的key，第三个就是key所对应的值。
redisTemplate.execute(redisScript, Arrays.asList(locKey), uuid);
} else {
// 其他线程等待
try {
// 睡眠
Thread.sleep(1000);
// 睡醒了之后，调用方法。
testLockLua();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
为了确保分布式锁可用
-
我们至少要确保锁的实现同时满足以下四个条件：
-
互斥性：在任意时刻，只有一个客户端能持有锁
-
不会发生死锁：即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁
-
解铃还须系铃人：加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了
-
原子性：加锁和解锁必须具有原子性
*★如果文章对你有帮助【关注点赞❤️收藏⭐】一起努力！★*
全栈小狐狸的逆袭之路：Java学习路线 Redis：基础篇 Redis：实战篇暑期编程PK赛 得CSDN机械键盘等精美礼品！
