Redis非关系型数据库
一、关于Redis
1.什么是NoSql？
NoSQL(NoSQL = Not Only SQL )，意即“不仅仅是SQL”,泛指非关系型的数据库.
它们都有些共同的特征：不需要预定义模式：不需要事先定义数据模式，预定义表结构。数据中的每条记录都可能有不同的属性和格式。当插入数据时，并不需要预先定义它们的模式。
弹性可扩展：可以在系统运行的时候，动态增加或者删除结点。不需要停机维护，数据可以自动迁移。
NoSQL代表MongDB、 Redis、Memcache.
2.什么是Redis?
Redis即远程字典服务，是一个开源的使用ANSI C语言编写、支持网络、可基于内存亦可持久化的日志型、“Key-Value”数据库，并提供多种语言的API；redis会周期性的把更新的数据写入磁盘或者把修改操作写入追加的记录文件。
Redis是现在最受欢迎的NoSQL数据库之一.
3.Redis的优势
当前 Redis 已经成为了主要的 NoSQL 工具，其原因如下。
1）响应快速
Redis 响应非常快，每秒可以执行大约 110 000 个写入操作，或者 81 000 个读操作，其速度远超数据库。如果存入一些常用的数据，就能有效提高系统的性能。
2）支持 8 种数据类型
它们是字符串、哈希结构、列表、集合、可排序集合， Geo 类型，基数和位图。前5种比较常用。
比如对于字符串可以存入一些 Java 基础数据类型，哈希可以存储对象，列表可以存储 List 对象等。这使得在应用中很容易根据自己的需要选择存储的数据类型，方便开发。
对于 Redis 而言，虽然只有 6 种数据类型，但是有两大好处：一方面可以满足存储各种数据结构体的需要；另外一方面数据类型少，使得规则就少，需要的判断和逻辑就少，这样读/写的速度就更快。
3）操作都是原子的
所有 Redis 的操作都是原子的，从而确保当两个客户同时访问 Redis 服务器时，得到的是更新后的值（最新值）。在需要高并发的场合可以考虑使用 Redis 的事务，处理一些需要锁的业务。
4）MultiUtility 工具
Redis 可以在如缓存、消息传递队列中使用（Redis 支持“发布+订阅”的消息模式），在应用程序如 Web 应用程序会话、网站页面点击数等任何短暂的数据中使用。
正是因为 Redis 具备这些优点，使得它成为了目前主流的 NoSQL 技术，在 Java 互联网中得到了广泛使用。
4. Redis应用场景
-
在分布式系统中，服务器有多台，客户的请求可能会发送到不同的服务器，我们可以用redis解决多台服务器间的session共享。
-
在高并发的情况下，所有的请求直接访问数据库，数据库容易出现异常。这个时候，就需要使用redis做一个缓冲操作，让请求先访问到redis，而不是直接访问数据库。
-
我们在碰到需要执行耗时特别久，且结果不频繁变动的SQL，就特别适合将运行结果放入缓存。这样，后面的请求就去缓存中读取，使得请求能够迅速响应。
具体场景：
1、会话缓存（最常用）
2、消息队列（支付）
3、活动排行榜或计数
4、发布，订阅消息（消息通知）
5、商品列表，评论列表
6、秒杀活动，抢红包
二、 Redis的安装和配置
1.安装redis
docker-compose.yml
version: '3.0'
services:
redis:
restart: always
image: daocloud.io/library/redis:5.0.9
container_name: redis
environment:
- TZ=Asia/Shanghai
ports:
- 6379:6379
2.使用Redis-cli连接Redis服务器
先进入redis容器内部：
docker exec -it 容器id bash
打开redis客户端
redis-cli
输入命令测试
3.使用Redis可视化管理工具连接Redis服务器
Redis Desktop Manager 是redis可视化管理工具，redi可视化客户端，redis集群管理工具。
3.1 下载和安装
https://github.com/ 网站上搜索 Redis Desktop Manager
https://github.com/lework/RedisDesktopManager-Windows/releases
下载以后安装，直接下一步下一步就可以。
3.2 创建连接
连接成功后：
三 Redis常用的5种数据类型
Redis支持的8种数据类型中前5种比较常用：string（字符串），hash（哈希），list（列表），set（集合）及zset(sorted set：有序集合)。
Redis是一个key-value形式的存储系统，key是一个“字符串”，而value对应的则是前面提到的5种数据类型。
字符串（string）
这是最常见和最容易理解的一种数据类型，它表示存储在redis中的值是一个“字符串”类型的数据。但实际上，它还能存储整型数据，后面我们将通过INCR命令，对值进行自增操作。
哈希（hash）
又称“散列”，这种数据类型类似于Java中的Map类型。初学者可能会疑惑，前面的“字符串”类型，一个key一个value不就是Map类型么。
实际上，在本文开头提到，redis是一种key-value形式的存储系统，我们所说的redis数据类型指的是value的数据类型。所以哈希（hash）也就是value是类似Map的一种数据类型。在后面的章节中我们会更直观的感受到。
列表（list）
列表（list）也可以理解为数组，和在Java中的List类型类似。略微不同的是，Java中的列表可以是泛型类型，也就是说Java中的List数据结构可以是字符串、整型等。而在redis中列表中的数据类型则只有字符串类型。
集合（set）
set类型在redis中被称为集合，同样它和Java的Set集合相同。和redis的列表（list）类似，不同地是，列表（list）的数据是可以重复的且是插入有序，而集合（set）中的数据是不可重复的且是无序。
有序集合（zset）
有序集合（zset）尽管看起来是集合（set）类型多了“有序”的特性。但实际上，可以说它和哈希（hash）更相似。因为它和哈希（hash）一样也是Map类型，不同地是它的key是实际上的成员，而value则是用于排序的“分值”。这个特性能帮助我们快速的实现“点赞数最高倒序排列”等功能。
四 Redis常用命令
1.Redis 字符串(String) 常用命令
1.set
set uname daimenglaoshi #存值
#set key value
- get
get uname #取值
#get key
- mset
mset sex m age 23 #批量存值
#mset key1 value1 key1 value1...
4.mget
mget sex age #批量取值
#mget key1 key2
5.自增
incr age #age中的值自增1
#incr key
incrby age 3 #age中的值自增3
#incrby key increment
- 自减、
decr age #age中的值自减1
#decr key
decrby age 3 #age中的值自减3
#decrby key increment
- Setnx
exists city #是否存在key city
setnx city shanghai #如果不存在city，则可以赋值，如果存在，则赋值失败
#setnx key value
8.Setex
Setex 命令为指定的 key 设置值及其过期时间。如果 key 已经存在， SETEX 命令将会替换旧的值
setex message 60 hello # 设置message=hello 保存时间60秒
#setex key seconds value
2.Redis 哈希(Hash) 常用命令
- hset
hset goods name huawei #存数据
#hset key field value
- hget
hget goods name #取数据
#hget key field
- hmset
hmset goods name huawei price 3000 #批量存储数据
#hmset key field vlue field2 value2
- hmget
hmget goods name price #批量取数据
#hmget key field1 field2
5.hincrby
hincrby goods price 100 # 增量 正数为+ 负数为-
#hincrby key field increment
- hexists
hexists goods name #判断字段是否存在
#hexists key field
- hsetnx
hsetnx goods num 100 #给不存在的字段赋值 如果字段已存在 赋值失败
#hsetnx key field value
- hkeys
hkeys goods #返回某个对象所有的字段名
#hkeys key
- hvals
hvals goods #返回某个对象所有的字段值
#hvals key
3.Redis 列表(List) 命令
- 存储数据
lpush list1 aa #列表左边添加一个值 也可以添加多个值
rpush list1 cc #列表右边添加一个值 也可以添加多个值
- 取出数据
lpop list1 #移除并返回列表左边第一个元素
rpop list1 #移除并返回列表右边第一个元素
- 列表存在才存储
lpushx list1 bb # list1存在 则左边添加一个bb
rpushx list1 dd # list1存在 则右边添加一个dd
- 获取指定索引范围的数据
lrange list1 0 -1 # 获取列表中所有元素 0代表第一个元素 -1 代表最后一个元素
#lrange list begin end
- 通过索引获取指定元素
lindex list1 0 # 返回下标为0的元素
#lindex list index
- 给指定位置赋值
lset list1 0 aa #设置集合list1的第一个元素为aa
#lset list index value
- 返回列表的长度
llen list1 #返回列表的长度
#llen list
- 在指定位置插入数据
linsert list1 before bb ee #在bb前面添加一个ee
#linsert key BEFORE | AFTER pivot value
- 对列表修剪
ltrim list 1 -1 #只保留列表中第2到最后的元素
#ltrim list begin end
- Rpoplpush
rpoplpush list1 list2 #将list1的最后一个元素弹出 并添加到list2的左边
- Brpoplpush
brpoplpush list1 list2 2 #将list1的最后一个元素弹出 并添加到list2的左边 如果没有元素 则等待2秒
#Brpoplpush 命令从列表中弹出一个值，将弹出的元素插入到另外一个列表中并返回它； 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止。
4.Redis 集合(Set) 命令
- sadd 添加数据
sadd myset user1,user2,user3 #添加数据 也可以添加多个
#sadd set value1 value2 。。
- 返回元素
smembers myset #返回所有数据
srandmember myset #随机返回一个元素，此元素还在集合中
srandmember myset #随机返回两个元素
- spop 随机移除数据
spop myset #随机移除一个元素
spop myset 2 #随机移除两个元素
- srem 删除指定数据
srem myset user1 #移除集合中的一个或多个成员元素，不存在的成员元素会被忽略
- scard 返回集合的数量
scard myset1 返回集合的数量
- smove 移动数据
smove myset1 myset2 user1 #将myset1中的user1移动到myset2中
- sinter 交集
sinter myset1 myset2 #返回myset1和myset2的交集
- sunion 并集
sunion myset1 myset2 #返回myset1和myset2的并集
- sdiff 差集
sdiff myset1 myset2 #返回myset1和myset2的差集
5.Redis 有序集合(sorted set) 命令
- 添加数据
zadd accessset 1 aa.com #添加数据 设置分数为1
#zadd set score value score value。。
- 取数据
zrange accessset 1 3 withscores #返回下标1--3之间的元素和分数 位置按分数从小到大
zrange accessset 0 -1 withscores #返回整个集合 按分数递增
zrevrange accessset 0 -1 withscores #返回整个集合 按分数递减
zrangebyscore accessset 1 100 # 返回分数在1到100之间的数据
zrangebyscore accessset 1 100 withscores #返回分数在1到100之间的数据和分数
zrangebyscore accessset 1 100 withscores limit 0 1 #返回分数在1到100之间的数据和分数的前1个
zrangebyscore accessset -inf +inf withscores #返回全部数据
zrevrangebyscore accessset 1 100 #降序返回分数在1到100之间的数据
- 移除数据
zrem accessset aa.com bb.com #移除有序集中的一个或多个成员，不存在的成员将被忽略
- 增加分数
zincrby accessset 2 aa.com #将aa.com的分数增加2
- 返回指定元素的分数
zscore accessset bb.com #返回bb.com的分数
- 返回指定元素的排名
zrank accessset aa.com #返回aa.com的排名 0代表第一
#有序集成员按分数值递增(从小到大)顺序排列
- 返回集合的数量
zcard accessset 返回集合中元素的数量
6.Keys常用命令
- 查看key
keys * #查看所有的key
keys u* #查看所有u开头的key
keys pattern
- 查看某一个key是否存在
exists uname
- 删除key
del uname
- 修改key的名称
rename id uid # 修改key的名称id为uid
#rename oldkey newkey #在集群模式下，key 和newkey 需要在同一个 hash slot。key 和newkey有相同的 hash tag 才能重命名
- 设置key的过期时间
expire uname 20 # 设置key为name的过期时间为20秒
#expire key_name seconds #设置key的过期时间
- 返回key的过期时间
ttl uname #返回uname的过期时间
#ttl key_name #返回指定key的过期时间 时间单位为秒
#pttl key_name #返回指定key的过期时间 时间单位为毫秒
- 删除指定key的过期时间
persist uname #删除uname的过期时间
- 移动key到指定数据库
数据库有0-15，默认是使用数据库0
可以用select选择数据库
move uname 1 #将数据库0中的uname移动到数据库1
select 1 #使用数据库1
exists uname #查看数据库1是否有uname
- 返回key中值得类型
type uname #返回uname中值的类型 可返回string,hash,list,set,zset等
7.Redis发布/订阅命令
除了使用List实现简单的消息队列功能以外，Redis还提供了发布订阅的消息机制。在这种机制下，消息发布者向指定频道（channel）发布消息，消息订阅者可以收到指定频道的消息，同一个频道可以有多个消息订阅者.
Redis也提供了一些命令支持这个机制.
- 发布消息
publish my_channel "hello" #给频道my_channel发布消息
#publish channe_name msg #用于将消息发送到指定的频道 返回的接收到消息的订阅者数量
- 订阅消息
subscribe my_channel #从my_channel频道中订阅消息
#subscribe channel_name 返回结果包含返回值的类型（订阅成功）、订阅的频道名称、目前已订阅的频道数量
- 取消订阅
unsubscribe my_channel #取消my_channel频道的消息订阅
#unsubscribe channel_name #取消一个或多个频道的订阅
- 按模式订阅消息
psubscribe channel:* #订阅所有以channel开头的频道
- 按模式取消订阅
punsubscribe channel:* #取消订阅所有以channel开头的频道
- 查询活跃频道
pubsub channels #查询至少有一个订阅者的频道
- 查看频道订阅数
pubsub numsub my_channel #查看my_channel频道的订阅数
五 Java和Redis的交互
Jedis是Redis官方首选的Java客户端开发包，库文件实现了对redis各类API进行封装调用。
5.1 使用Jedis连接Redis
Jedis直连相当于一个TCP连接，数据传输完成后关闭连接。
1.创建springboot项目
2.导入相关依赖包
<!-- lombok 包-->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<!-- 默认的测试包不是junit 需要移除默认的测试包-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.junit.vintage</groupId>
<artifactId>junit-vintage-engine</artifactId>
</exclusion>
</exclusions>
</dependency>
<!-- 加入junit 包-->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
<!-- redis client-->
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>3.0.0</version>
</dependency>
3.创建测试类测试
4.String类型的命令应用场景
案例1：计数
如微博的评论数、点赞数、分享数，抖音作品的收藏数，京东商品的销售量、评价数等。
案例2：存对象
对于属性值不常改变的对象，可以将对象转换成json字符串，再存储在string类型中，是个不错的选择，如用户信息、商品信息等。
创建user类：
@Data
@NoArgsConstructor
@AllArgsConstructor
public class User {
private int id;
private String name;
private int age;
}
导入fastjson依赖：
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.15</version>
</dependency>
测试：
5.hash类型的命令应用场景
案例1：用hash存购物车信息
案例2：用hash存对象
存储对象可以用String命令，将对象转换为字节数组或json字符串，也可以用Hash命令。
当对象的属性不需要频繁修改时，可以选择用String方式。
当对象的某个属性需要频繁修改时，不适合用string+json，因为它不够灵活，每次修改都需要重新将整个对象序列化并赋值，如果使用hash类型，则可以针对某个属性单独修改，没有序列化，也不需要修改整个对象。比如，商品的价格、销量、关注数、评价数等可能经常发生变化的属性，就适合存储在hash类型里。
但是，当对象的某个属性不是基本类型或字符串时，使用hash类型就必须手动进行复杂序列化。非常麻烦。
所以可以选择：一般对象用string存储，对象中某些频繁变化的属性抽出来用hash存储。
6.list类型的命令应用场景
list类型是简单的字符串列表，按照插入顺序排序。
list可以用来做消息队列，排行榜，最新评论列表，最新点赞列表等，不过后面有专门的rabbitMQ等消息队列中间件，redis中的list主要是用来做排行榜和最新列表之类。
案例：排行榜
list类型的lrange命令可以分页查看队列中的数据。可将每隔一段时间计算一次的排行榜存储在list类型中，如商品销量排行、成绩排名等。
但是，并不是所有的排行榜都能用list类型实现，只有定时计算的排行榜才适合使用list类型存储，与定时计算的排行榜相对应的是实时计算的排行榜，list类型不能支持实时计算的排行榜，之后在介绍有序集合sorted set的应用场景时会详细介绍实时计算的排行榜的实现。
//list类型存排行榜
@Test
public void testList()
{
//1.创建连接对象
Jedis jedis=new Jedis("42.192.12.30",6379);
//2.操作
jedis.lpush("rankinglist","zhongxing");
jedis.lpush("rankinglist","xiaomi");
jedis.lpush("rankinglist","vivo");
jedis.lpush("rankinglist","huawei");
List<String> list=jedis.lrange("rankinglist",0,2);
for(String s:list)
System.out.println(s);
//3.关闭连接
jedis.close();
}
案例2：
list类型的lpush命令和lrange命令能实现最新列表的功能，每次通过lpush命令往列表里插入新的元素，然后通过lrange命令读取最新的元素列表，如朋友圈的点赞列表、评论列表。
但是，并不是所有的最新列表都能用list类型实现，因为对于频繁更新的列表，list类型的分页可能导致列表元素重复或漏掉，举个例子，当前列表里由表头到表尾依次有（E，D，C，B，A）五个元素，每页获取3个元素，用户第一次获取到（E，D，C）三个元素，然后表头新增了一个元素F，列表变成了（F，E，D，C，B，A），此时用户取第二页拿到（C，B，A），元素C重复了。只有不需要分页（比如每次都只取列表的前5个元素）或者更新频率低（比如每天凌晨更新一次）的列表才适合用list类型实现。对于需要分页并且会频繁更新的列表，需用使用有序集合sorted set类型实现。另外，需要通过时间范围查找的最新列表，list类型也实现不了，也需要通过有序集合sorted set类型实现，如以成交时间范围作为条件来查询的订单列表。之后在介绍有序集合sorted set类型的应用场景时会详细介绍sorted set类型如何实现最新列表。
//list类型存最新列表 如评论列表
@Test
public void testList2()
{
//1.创建连接对象
Jedis jedis=new Jedis("42.192.12.30",6379);
//2.操作
jedis.lpush("commentlist","不错");
jedis.lpush("commentlist","牛b");
jedis.lpush("commentlist","很好");
jedis.lpush("commentlist","棒棒的");
List<String> list=jedis.lrange("commentlist",0,-1);
for(String s:list) {
System.out.println(s);
jedis.lpop(s);
}
//3.关闭连接
jedis.close();
}
7.set类型的命令应用场景
集合类型的元素的特点是唯一且无序。
案例1：好友列表
可以用它存储好友列表
sinter获取两个用户的共同好友，scard可以获取好友数量。
//set 存放好友列表
@Test
public void testSet1()
{
//1.创建连接对象
Jedis jedis=new Jedis("42.192.12.30",6379);
//2.操作
//A的好友
jedis.sadd("friendsA:list","aa","bb","cc");
//B的好友
jedis.sadd("friendsB:list","bb","cc","dd");
//查看共同好友
Set<String> innerSet=jedis.sinter("friendsA:list","friendsB:list");
for(String s:innerSet) {
System.out.println(s);
}
//3.关闭连接
jedis.close();
}
案例2：随机展示，抽奖
网站或app首页展示推荐时，可以提前选一批需要展示的内容，然后从中随机选取一部分展示。
//set 抽奖
@Test
public void testSet2()
{
//1.创建连接对象
Jedis jedis=new Jedis("42.192.12.30",6379);
//2.操作
// 所有用户列表
jedis.sadd("user:list","aa","bb","cc","dd");
//随机抽取一人 抽完此人还在集合中
String luckUser=jedis.srandmember("user:list");
System.out.println(luckUser);
System.out.println("---------------------");
//随机抽取两人
List<String> luckList=jedis.srandmember("user:list",2);
//查看集合中所有元素
Set<String> usersSet=jedis.smembers("user:list");
for(String s:usersSet)
System.out.println(s);
System.out.println("---------------------");
//随机抽取1人 抽完将他移除
String luckUser2=jedis.spop("user:list");
System.out.println(luckUser2);
System.out.println("---------------------");
//随机抽取两人 抽完将他们移除
Set<String> luckList2= jedis.spop("user:list",2);
//查看集合中所有元素
Set<String> usersSet2=jedis.smembers("user:list");
for(String s:usersSet)
System.out.println(s);
//3.关闭连接
jedis.close();
}
8.zset类型的命令应用场景
前面我们用list做过排行榜，但不是所有的排行榜都能用list类型实现，只有定时计算的排行榜才适合使用list类型存储，与定时计算的排行榜相对应的是实时计算的排行榜，实时排行榜可以用zset来实现。
比如：热点新闻排行，直播打赏排行等。
也可以用在需要排序的列表，比如 关注列表，粉丝列表等。
案例：热点新闻排行榜
//zset 排行榜
@Test
public void testZset()
{
//1.创建连接对象
Jedis jedis=new Jedis("42.192.12.30",6379);
jedis.flushDB();//清空数据库
//2.操作
jedis.zadd("hotnews",1000,"新闻1"); //score存放点击数
jedis.zadd("hotnews",2000,"新闻2");
jedis.zadd("hotnews",1500,"新闻3");
jedis.zadd("hotnews",1800,"新闻4");
jedis.zadd("hotnews",2500,"新闻5");
//查询点击数前3的新闻
Set<String> set= jedis.zrevrange("hotnews",0,2);
for(String s: set)
System.out.println(s);
System.out.println("-------------------");
//查询点击数前3的新闻和对应的点击数
Set<Tuple> set2= jedis.zrevrangeWithScores("hotnews",0,2);
for (Tuple t : set2)
System.out.println(t.getElement() + ":" + t.getScore());
//3.关闭连接
jedis.close();
}
5.2 使用 Jedis连接池
采用jedis直连方式，每次都需要创建连接对象和关闭对象，我们可以使用之前学过的线程池概念，采用jedis连接池。使用Jedis线程池可以不需要创建新的Jedis对象连接Redis，可以大大减少对于创建和回收Redis连接的开销。
1 使用连接池
@Test
public void testPool()
{
//1.创建连接池 采用默认配置 连接池中的redis预先定义好
JedisPool jedisPool=new JedisPool("42.192.12.30",6379);
//2.从连接池中获取jedis对象
Jedis jedis=jedisPool.getResource();
//3.使用jedis
String val=jedis.get("aa");
System.out.println(val);
//4.释放jedis到连接池
jedis.close();;
}
2 设置连接池的配置信息
@Test
public void testPool2()
{
//1.创建连接池配置信息
GenericObjectPoolConfig poolConfig=new GenericObjectPoolConfig();
poolConfig.setMaxTotal(50); //连接池最大连接数 默认是8
poolConfig.setMaxIdle(50); //允许的最大空闲数 默认是8 建议MaxTotal=MaxIdle
poolConfig.setMinIdle(2); //最小空闲数 默认是0
// 当连接池中资源用尽时，调用者的最大等待时间（毫秒)
//默认是-1 表示永不超时，不建议使用默认值，这样会消耗大量时间
poolConfig.setMaxWaitMillis(1000);
//2.创建连接池 采用默认配置 连接池中的redis预先定义好
JedisPool jedisPool=new JedisPool(poolConfig,"42.192.12.30",6379);
//3.从连接池中获取jedis对象
Jedis jedis=jedisPool.getResource();
//4.使用jedis
String val=jedis.get("aa");
System.out.println(val);
//5.释放jedis到连接池
jedis.close();;
}
5.3 使用SpringBoot配置连接池信息
在实际开发中，配置信息应该从代码中提出，放在单独的配置文件中，这样不用每次使用时都需要配置，也便于之后修改配置信息.
在application.propertis文件中加入:
jedis.pool.host=42.192.12.30
#Redis服务器连接端口
jedis.pool.port=6379
jedis.pool.config.maxTotal=50
jedis.pool.config.maxIdle=50
jedis.pool.config.minIdle=2
jedis.pool.config.maxWaitMillis=1000
如果使用的是yml格式则为:
jedis:
pool:
host: 42.192.12.30
port： 6379
config:
maxTotal: 50
maxIdle: 50
minIdle: 2
maxWaitMillis: 1000
注意：yml基本格式要求：
1.大小写敏感；
2.使用缩进代表层级关系；
3.缩进只能使用空格，不能使用tab键，不要求空格个数，只需要相同层级左对齐（一般2或4个空格）。
配置项格式为key: value，冒号后要有一个空格：
创建RedisConfiguration类，将配置文件信息注入进来
package com.example.testjedis.config;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
@Configuration
public class RedisConfiguration {
@Autowired
@Qualifier("jedis.pool.config")
private JedisPoolConfig jedisPoolConfig;
@Bean(name= "jedis.pool")
public JedisPool jedisPool(@Value("${jedis.pool.host}")String host,
@Value("${jedis.pool.port}")int port) {
return new JedisPool(jedisPoolConfig, host, port);
}
@Bean(name= "jedis.pool.config")
public JedisPoolConfig jedisPoolConfig (@Value("${jedis.pool.config.maxTotal}")int maxTotal,
@Value("${jedis.pool.config.maxIdle}")int maxIdle,
@Value("${jedis.pool.config.minIdle}")int minIdle,
@Value("${jedis.pool.config.maxWaitMillis}")int maxWaitMillis) {
JedisPoolConfig config = new JedisPoolConfig();
config.setMaxTotal(maxTotal);
config.setMaxIdle(maxIdle);
config.setMaxWaitMillis(maxWaitMillis);
return config;
}
}
创建代码测试：
@RestController
public class TestController {
@Autowired
private JedisPool jedisPool;
@RequestMapping("testRedisPool")
public String testRedisPool()
{
Jedis jedis= jedisPool.getResource();
jedis.set("name","aa");
String val=jedis.get("name");
return val;
}
}
六、设置Redis访问权限
1.设置配置文件路径
docker-compose.yml
ersion: '3.0'
services:
redis:
restart: always
image: daocloud.io/library/redis:5.0.9
container_name: redis
environment:
- TZ=Asia/Shanghai
ports:
- 6379:6379
volumes:
- ./redis/redis.conf:/usr/local/etc/redis/redis.conf
command: ["redis-server","/usr/local/etc/redis/redis.conf"]
2.配置文件中设置密码
在redis目录下创建redis.conf文件
redis.conf文件中添加：
requirepass daimenglaoshi #设置访问密码
3.Jedis连接时需要设置密码
jedis.auth("密码");
作业:springboot整合jedisPool 加密码
七、批量处理多条命令
1. 使用管道
在某些高并发的场景下，网络开销成了Redis速度的瓶颈，所以需要使用管道技术来实现突破。
每个命令的执行时间 = 客户端发送时间+服务器处理和返回时间+一个网络来回的时间
那么如果有多个命令要一起执行，我们可以放入管道，一起发送过去，节约网略来回的时间。
但管道操作会消耗一定的内存，所以管道中命令的数量并不是越大越好（太大容易撑爆内存），而是应该有一个合理的值。
另外管道中某些指令失败，管道也会继续执行。所以不适合存放后面需要前面结果的命令。
//测试不用管道执行一万次命令
@Test
public void test()
{
//1.创建连接池 采用默认配置 连接池中的redis预先定义好
JedisPool jedisPool=new JedisPool("42.192.12.30",6379);
//2.从连接池中获取jedis对象
Jedis jedis=jedisPool.getResource();
long begin=System.currentTimeMillis();
jedis.set("count","1");
//3.使用jedis
for(int i=1;i<=10000;i++)
jedis.incr("count");
long end=System.currentTimeMillis();
System.out.println("执行时间:"+(end-begin)); //82312
//4.释放jedis到连接池
jedis.close();
}
//测试用管道执行一万次命令
@Test
public void test2()
{
//1.创建连接池 采用默认配置 连接池中的redis预先定义好
JedisPool jedisPool=new JedisPool("42.192.12.30",6379);
//2.从连接池中获取jedis对象
Jedis jedis=jedisPool.getResource();
//3.创建管道
Pipeline pipeline=jedis.pipelined();
long begin=System.currentTimeMillis();
jedis.set("count","1");
//4.将命令加入管道
for(int i=1;i<=10000;i++)
pipeline.incr("count");
//5.执行管道中的命令
pipeline.syncAndReturnAll();
long end=System.currentTimeMillis();
System.out.println("执行时间:"+(end-begin));//485
//6.释放jedis到连接池
jedis.close();;
}
2.使用事务
redis事务提供了一种“将多个命令打包， 然后一次性、按顺序地执行”的机制， 并且事务在执行的期间不会主动中断 —— 服务器在执行完事务中的所有命令之后， 才会继续处理其他客户端的其他命令。
redis事务使用了multi、exec、discard、watch、unwatch命令。
2.1.执行事务
2.2取消事务
2.3 事务中有语法错误
若在事务队列中存在语法性错误，则执行exec命令时，其他正确命令会被执行，错误命令抛出异常。
2.4 事务中出现命令错误
若在事务队列中存在命令性错误，则执行exec命令时，所有命令都不会执行。
2.5 watch监听
watch命令用于在事务开始之前监视任意数量的键： 当调用exec命令执行事务时， 如果任意一个被监视的键已经被其他客户端修改了， 那么整个事务不再执行， 直接返回失败。没有改变则正常执行事务。
3.使用lua脚本
Lua是一个小巧的脚本语言，一般用于嵌入式应用，现在越来越多应用于游戏当中，魔兽世界,愤怒的小鸟都有用到。
Lua极易嵌入到其他程序，可当做一种配置语言。还可以应用到很多需要性能的地方，比如：游戏脚本，nginx，redis的脚本
互联网应用场景：
Nginx + lua 开发高性能web应用，限流、防止sql注入、请求过滤，黑白名单限制等等等。
redis + lua 实现原子操作，避免多线程数据不一致的问题
3.1设置脚本对应的数据卷
修改docker-compose.yml文件，添加脚本对应的数据卷
version: '3.0'
services:
redis:
restart: always
image: daocloud.io/library/redis:5.0.9
container_name: redis
environment:
-- TZ=Asia/Shanghai
ports:
- 6379:6379
volumes:
- ./redis/redis.conf:/usr/local/etc/redis/redis.conf
- ./redis/lua/:/usr/local/etc/redis/lua #添加lua文件对应的数据卷
command: ["redis-server","/usr/local/etc/redis/redis.conf"]
重新执行docker-compose
docker-compose down
docker-compose up -d
3.2.在lua数据卷目录下创建lua文件
1.lua文件内容如下：
local key=KEYS[1]
local value=ARGV[1]
local key2=KEYS[2]
local value2=ARGV[2]
redis.call("set",key,value);
redis.call("set",key2,value2);
3.3.执行lua文件
进入到redis容器
docker exec -it 容器id bash
执行lua脚本文件
redis-cli --eval /usr/local/etc/redis/lua/1.lua name age , aa 18
# redis-cli --eval lua文件路径 key1 key2 , value1 value2
# redis-cli -a 密码 --eval lua文件路径 key1 key2 , value1 value2
#注意 key和value之间的逗号前后要有空格
八、Redis持久化机制
redis是一个内存数据库，数据保存在内存中，如果突然宕机，数据就会全部丢失。因此必须有一种机制来保证 Redis 的数据不会因为故障而丢失，这种机制就是 Redis 的持久化机制。
Redis 的持久化机制有两种，第一种是RDB快照(Redis DataBase)，第二种是 AOF 日志(Append Only File)。
1.RDB快照
1.1 RDB快照优缺点
RDB其实就是把数据以快照的形式保存在磁盘上。什么是快照呢，你可以理解成把当前时刻的数据拍成一张照片保存下来。
RDB持久化是指在指定的时间间隔内将内存中的数据集快照写入磁盘。也是默认的持久化方式，这种方式是就是将内存中数据以快照的方式写入到二进制文件中,默认的文件名为dump.rdb。
优点
- RDB文件小，非常适合定时备份，用于灾难恢复
- Redis加载RDB文件的速度比AOF快很多，因为RDB文件中直接存储的是内存数据，而AOF文件中存储的是一条条命令，需要重演命令。
缺点
- RDB无法做到实时持久化，若在两次bgsave间宕机，则会丢失区间（分钟级）的增量数据，不适用于实时性要求较高的场景
- 每次保存RDB的时候，Redis都要fork()出一个子进程，并由子进程来进行实际的持久化工作。在数据集比较庞大时，fork()可能会非常耗时，造成服务器在某某毫秒甚至1秒内停止处理客户端
- 存在老版本的Redis不兼容新版本RDB格式文件的问题
1.2 使用RDB方式实现持久化
修改redis.conf，添加配置信息
添加数据卷，对应保存的快照文件目录
version: '3.0'
services:
redis:
restart: always
image: daocloud.io/library/redis:5.0.9
container_name: redis
environment:
-- TZ=Asia/Shanghai
ports:
- 6379:6379
volumes:
- ./redis/redis.conf:/usr/local/etc/redis/redis.conf
- ./redis/lua/:/usr/local/etc/redis/lua #添加lua文件对应的数据卷
- ./redis/data/:/data #添加保存的*.rdb文件对应的数据卷
command: ["redis-server","/usr/local/etc/redis/redis.conf"]
重新运行docker-compse
docker-compose down
docker-compose up -d
1.3 触发机制
手动触发：
save命令:
在客户端中执行 save命令，就会触发 Redis 的持久化，但同时也是使 Redis 处于阻塞状态，直到 RDB 持久化完成，才会响应其他客户端发来的命令，所以在生产环境一定要慎用。
save
bgsave命令:
bgsave 会 fork() 一个子进程来执行持久化，整个过程中只有在 fork() 子进程时有短暂的阻塞，当子进程被创建之后，Redis 的主进程就可以响应其他客户端的请求了，相对于整个流程都阻塞的 save 命令来说，显然 bgsave 命令更适合我们使用
bgsave
自动触发:
满足redis.conf文件中任何一个条件就自动触发bgSave；
在主从复制场景下，如果从节点执行全量复制操作，则主节点会自动执行 bgsave 命令，并将rdb文件发送给从节点，后面会讲主从复制；
执行shutdown命令时，也会自动执行rdb持久化。
测试：
查看保存出来的数据
数据卷redis/data文件夹下有一个redis.rdb文件
redis重启，数据还在。
2.AOF 日志
AOF(append only file):
AOF是将Redis执行的每次写命令记录到单独的日志文件中；当Redis重启时再次执行AOF文件中的命令来恢复数据。
2.1 AOF执行流程
- 所有的写命令都会追加到aof_buf（缓冲区）中。
- 可以使用不同的策略将AOF缓冲区中的命令写到AOF文件中并持久化。
- 随着AOF文件的越来越大，会对AOF文件进行重写,并替换掉之前的AOF文件
- 当服务器重启的时候，会加载AOF文件并执行AOF文件中的命令用于恢复数据。
2.2 AOF优缺点
优点
- AOF可以更好的保护数据不丢失。一般AOF会每隔1秒，通过一个后台线程执行一次fsync操作，最多丢失1秒钟的数据，Redis进程挂了，最多丢掉1秒钟的数据。
- AOF日志文件以append-only模式写入，所以没有任何磁盘寻址的开销，写入性能非常高，而且文件不容易破损，即使文件尾部破损，也很容易修复。
缺点
- AOF方式生成的日志文件太大，需要不断AOF重写，进行瘦身。
- 即使经过AOF重写瘦身，由于文件是文本文件，文件体积较大（相比于RDB的二进制文件）。
- AOF重演命令式的恢复数据，恢复的速度显然比RDB要慢。
2.3 AOF重写
我们知道 AOF 是不断的将写命令追加到一个后缀叫 .aof 的文件当中的,那么问题来了,随着我们不断的写命令,aof文件越来越大,而里面很多命令是可以优化的
比如
set age 18
del age
incr age
decr age
这4条命令执行完，其实什么数据都没改变。那么我们可以重写这个aof文件，将无意义的命令删除，或者将多个命令合并为一个命令。
AOF重写流程
-
Redis主进程会fork一个子进程执行AOF重写，开销和RDB重写一样。
-
AOF重写过程中，不影响Redis原有的AOF过程，包括写消息到AOF缓存以及同步AOF缓存中的数据到硬盘。
-
AOF重写过程中，主进程收到的写操作还会将命令写到AOF重写缓冲区，注意和AOF缓冲区区分开。
-
由于AOF重写过程中原AOF文件还在陆续写入数据，所以AOF重写子进程只会拿到fork子进程时的AOF文件进行重写。
-
子进程拿到原AOF文件中的数据写到一个临时的AOF文件中。
-
子进程完成AOF重写后会发消息给主进程，主进程会把AOF重写缓冲区中的数据写道AOF缓冲区，并且用新的AOF文件替换旧的AOF文件。
重写命令：
bgrewriteaof
2.4 AOF同步策略
为了提高文件的写入效率，在现代操作系统中，当用户调用write函数，将一些、数据写入到文件的时候，操作系统通常会将写入数据暂时保存在一个内存缓冲区里面。等到缓冲区的空间被填满、或者超过了指定的时
限之后，才真正地将缓冲区中的数据写入到磁盘里面。这种做法虽然提高了效率，但也为写入数据带来了安全问题，因为如果计算机发生停机，那么保存在内存缓冲区里面的写入数据将会丢失。为此，系统提供了fsync和fdatasync两个同步函数，它们可以强制让操作系统立即将缓冲区中的数据写入到硬盘里面，从而确保写入数据的安全性。
AOF提供了三种同步策略
- always：每一条AOF记录都立即同步到文件，性能很低，但较为安全。
- everysec：每秒同步一次，性能和安全都比较中庸的方式，也是redis推荐的方式。如果遇到物理服务器故障，可能导致最多1秒的AOF记录丢失。
- no：Redis永不直接调用文件同步，而是让操作系统来决定何时同步磁盘。一般为30秒，性能较好，但很不安全。
2.5 使用AOF方式实现持久化
配置redis.conf文件
## 此选项为aof功能的开关，默认为“no”，可以通过“yes”来开启aof功能
## 只有在“yes”下，aof重写/文件同步等特性才会生效
appendonly yes
## 指定aof文件名称
appendfilename appendonly.aof
## 指定aof操作中文件同步策略，有三个合法值：always everysec no,默认为everysec
appendfsync everysec
## 在aof-rewrite期间，appendfsync是否暂缓文件同步，"no"表示“不暂缓”，“yes”表示“暂缓”，默认为“no”
no-appendfsync-on-rewrite no
## aof文件rewrite触发的最小文件尺寸(mb,gb),只有大于此aof文件大于此尺寸是才会触发rewrite，默认“64mb”，建议“512mb”
auto-aof-rewrite-min-size 64mb
## 相对于“上一次”rewrite，本次rewrite触发时aof文件应该增长的百分比
## 每一次rewrite之后，redis都会记录下此时“新aof”文件的大小，假设为A
## aof文件增长到A*(1 + p)之后，触发下一次rewrite，每一次aof记录的添加，都会检测当前aof文件的尺寸。
auto-aof-rewrite-percentage 100
触发机制
手动触发 :
redis-cli -h ip -p port bgrewriteaof
自动触发:
根据redis.conf文件中配置的auto-aof-rewrite-min-size和auto-aof-rewrite-percentage参数确定自动触发时机。
九、Redis主从架构
我们前面说过，单台redis服务器，在最优的情况下读的速度可以达到11万次/秒，写的速度8.1万次/秒，但是如果对于每秒超过10万次的读写请求，则单台redis服务器不够。这个时候我们需要配置多台redis。
而一般高并发的应用，写的请求是比较少的，大量的请求都是读，网站的读写比可以高达10：1甚至更高。
那么我们可以采用 主从架构+读写分离来应对10w+QPS系统。
1.配置Redis主从架构实现读写分离
创建docker-compose.yml,开启三个redis容器。
配置一主二从：
version: '3.0'
services:
redis_master:
restart: always
image: daocloud.io/library/redis:5.0.9
container_name: redis_master
environment:
- TZ=Asia/Shanghai
ports:
- 6380:6379
volumes:
- ./redis/redis_master.conf:/usr/local/etc/redis/redis.conf
command: ["redis-server","/usr/local/etc/redis/redis.conf"]
redis_slave_1:
restart: always
image: daocloud.io/library/redis:5.0.9
container_name: redis_slave_1
environment:
- TZ=Asia/Shanghai
ports:
- 6381:6379
volumes:
- ./redis/redis_slave_1.conf:/usr/local/etc/redis/redis.conf
links:
- redis_master:master
command: ["redis-server","/usr/local/etc/redis/redis.conf"]
redis_slave_2:
restart: always
image: daocloud.io/library/redis:5.0.9
container_name: redis_slave_2
environment:
- TZ=Asia/Shanghai
ports:
- 6382:6379
volumes:
- ./redis/redis_slave_2.conf:/usr/local/etc/redis/redis.conf
links:
- redis_master:master
command: ["redis-server","/usr/local/etc/redis/redis.conf"]
在docker-compose.yml 文件同级目录下创建redis文件夹，redis文件夹下创建3个配置文件：
redis_master.conf，redis_slave_1.conf，redis_slave_2.conf
redis_slave_1.conf 和 redis_slave_2.conf中配置:
replicaof master 6379
注意:这里的master是之前在 docker-compose里设置的master服务器的别名，这里也可以用ip
执行docker-compose.yml:
docker-compose up -d
查看：
docker ps
测试:
进入master：
进入slave:
主机master一旦发生增删改操作，那么从机会自动将数据同步到从机slave中
查看节点状态：
info
2.主从复制的原理
Redis主从复制可以根据是否是全量分为全量同步和增量同步。
2.1 全量同步
Redis全量复制一般发生在Slave初始化阶段，这时Slave需要将Master上的所有数据都复制一份。具体步骤如下：
1）从服务器连接主服务器，发送SYNC命令；
2）主服务器接收到SYNC命名后，开始执行BGSAVE命令生成RDB文件并使用缓冲区记录此后执行的所有写命令；
3）主服务器BGSAVE执行完后，向所有从服务器发送快照文件，并在发送期间继续记录被执行的写命令；
4）从服务器收到快照文件后丢弃所有旧数据，载入收到的快照；
5）主服务器快照发送完毕后开始向从服务器发送缓冲区中的写命令；
6）从服务器完成对快照的载入，开始接收命令请求，并执行来自主服务器缓冲区的写命令；
完成上面几个步骤后就完成了从服务器数据初始化的所有操作，从服务器此时可以接收来自用户的读请求。
2.2 增量同步
Redis增量复制是指Slave初始化后开始正常工作时主服务器发生的写操作同步到从服务器的过程。
增量复制的过程主要是主服务器每执行一个写命令就会向从服务器发送相同的写命令，从服务器接收并执行收到的写命令。
2.3 Redis主从同步策略
主从刚刚连接的时候，进行全量同步；全同步结束后，进行增量同步。当然，如果有需要，slave 在任何时候都可以发起全量同步。redis 策略是，无论如何，首先会尝试进行增量同步，如不成功，要求从机进行全量同步。
十、Redis哨兵（Sentinel）模式
1.为什么要用哨兵
当主节点宕机了，整个集群就没有可写的节点了。
但是从节点上备份了主节点的所有数据，那在主节点宕机的情况下，如果能够将从节点变成一个主节点，是不是就可以解决这个问题了呢？是的，这个就是Sentinel哨兵的作用。
2.哨兵的功能
Redis 的 Sentinel 系统用于管理多个 Redis 服务器（instance）， 该系统执行以下三个任务：
2.1监控（Monitoring）
Sentinel 会不断地检查你的主服务器和从服务器是否运作正常。
- Sentinel可以监控任意多个Master和该Master下的Slaves。（即多个主从模式）
- 同一个哨兵下的、不同主从模型，彼此之间相互独立。
- Sentinel会不断检查Master和Slaves是否正常。
2.2 提醒（Notification）
当被监控的某个 Redis 服务器出现问题时， Sentinel 可以通过 API 向管理员或者其他应用程序发送通知。
2.3 自动故障迁移（Automatic failover）
监控同一个Master的Sentinel会自动连接，组成一个分布式的Sentinel网络，互相通信并交换彼此关于被监视服务器的信息。
当一个主服务器不能正常工作时， Sentinel 会开始一次自动故障迁移操作， 它会进行选举，将其中一个从服务器升级为新的主服务器， 并让失效主服务器的其他从服务器改为复制新的主服务器； 当客户端试图连接失效的主服务器时， 集群也会向客户端返回新主服务器的地址， 使得集群可以使用新主服务器代替失效服务器。
故障迁移的步骤：
- 投票（半数原则）
当任何一个Sentinel发现被监控的Master下线时，会通知其它的Sentinel开会，投票确定该Master是否下线（半数以上，所以sentinel通常配奇数个）。
- 选举
当Sentinel确定Master下线后，会在所有的Slaves中，选举一个新的节点，升级成Master节点。
其它Slaves节点，转为该节点的从节点。
- 原Master重新上线
当原Master节点重新上线后，自动转为当前Master节点的从节点。
3、哨兵模式部署和自动选举
在主从架构的基础上配置哨兵，比如上面运行过一个一主二从的结构，现在可以配置三个Sentinel实例，监控同一个Master节点和它的从节点。
3.1.修改docker-compose.yml文件
添加sentinel配置文件的数据卷
version: '3.0'
services:
redis_master:
restart: always
image: daocloud.io/library/redis:5.0.9
container_name: redis_master
environment:
- TZ=Asia/Shanghai
ports:
- 6380:6379
volumes:
- ./redis/redis_master.conf:/usr/local/etc/redis/redis.conf
- ./redis/sentinel1.conf:/usr/local/etc/redis/sentinel.conf
command: ["redis-server","/usr/local/etc/redis/redis.conf"]
redis_slave_1:
restart: always
image: daocloud.io/library/redis:5.0.9
container_name: redis_slave_1
environment:
- TZ=Asia/Shanghai
ports:
- 6381:6379
volumes:
- ./redis/redis_slave_1.conf:/usr/local/etc/redis/redis.conf
- ./redis/sentinel2.conf:/usr/local/etc/redis/sentinel.conf
links:
- redis_master:master
command: ["redis-server","/usr/local/etc/redis/redis.conf"]
redis_slave_2:
restart: always
image: daocloud.io/library/redis:5.0.9
container_name: redis_slave_2
environment:
- TZ=Asia/Shanghai
ports:
- 6382:6379
volumes:
- ./redis/redis_slave_2.conf:/usr/local/etc/redis/redis.conf
- ./redis/sentinel3.conf:/usr/local/etc/redis/sentinel.conf
links:
- redis_master:master
command: ["redis-server","/usr/local/etc/redis/redis.conf"]
3.2.创建三个sentinel的配置文件
在redis目录下创建3个配置文件
sentinel1.conf（主节点）
#daemonize是用来指定redis是否要用守护线程的方式启动，yes代表后台运行redis
daemonize yes
#Redis监控一个叫做master的运行在localhost:6379的主节点，投票达到2则表示master已经挂掉了。#localhost可以改为ip
sentinel monitor master localhost 6379 2
#设置多久监听一下redis状态 毫秒为单位
sentinel down-after-milliseconds master 8000
sentinel2.conf和sentinel3.conf （从节点）
daemonize yes
sentinel monitor master master 6379 2
sentinel down-after-milliseconds master 8000
#这里的master是之前给主节点取的名字，可以改用ip
3.3 运行docker-compose文件
3.4 启动哨兵
分别进入三个容器内部，启动哨兵，可以开三个会话，方便测试
3.5 连接redis服务器，查看节点状态
从节点都启动以后，主节点可以看到有两个从节点连接上了，以及对应的ip，从节点可以看到主节点的ip
3.6 测试主节点关闭，重新选举master
关闭主节点：
从1被选举为新的主节点:
从2还是从节点
扩展: 哨兵也可以单独创建
version: '3.0'
services:
sentinel1:
restart: always
image: daocloud.io/library/redis:5.0.9
container_name: sentinel1
environment:
- TZ=Asia/Shanghai
ports:
- 26380:26379
volumes:
- ./sentinel/sentinel1.conf:/usr/local/etc/redis/sentinel.conf
command: ["redis-sentinel","/usr/local/etc/redis/sentinel.conf"]
sentinel2:
restart: always
image: daocloud.io/library/redis:5.0.9
container_name: sentinel2
environment:
- TZ=Asia/Shanghai
ports:
- 26381:26379
volumes:
- ./sentinel/sentinel2.conf:/usr/local/etc/redis/sentinel.conf
command: ["redis-sentinel","/usr/local/etc/redis/sentinel.conf"]
sentinel3:
restart: always
image: daocloud.io/library/redis:5.0.9
container_name: sentinel3
environment:
- TZ=Asia/Shanghai
ports:
- 26382:26379
volumes:
- ./sentinel/sentinel3.conf:/usr/local/etc/redis/sentinel.conf
command: ["redis-sentinel","/usr/local/etc/redis/sentinel.conf"]
十一 Redis集群
1.为什么要搭建Redis集群?
我们前面设置了主从，可以将读写分离，应对大的访问量，但是当访问量继续增大的情况下，一个主从还是不够，而且由于Redis主从架构每个数据库都要保存整个集群中的所有数据,容易形成木桶效应，所以Redis3.0之后的版本添加了集群(Cluster)架构，也就是可以将数据分布在集群中不同的redis节点，每个redis节点可以再搭建自己的主从。
2.Redis集群方案
Redis集群方案基于分而治之的思想。Redis中数据都是以Key-Value形式存储的，而不同Key的数据之间是相互独立的。因此可以将Key按照某种规则划分成多个分区，将不同分区的数据存放在不同的节点上。这个方案类似数据结构中哈希表的结构。在Redis集群的实现中，使用哈希算法(公式是CRC16(Key) mod 16383)将Key映射到0~16383范围的整数。这样每个整数对应存储了若干个Key-Value数据，这样一个整数对应的抽象存储称为一个槽(slot)。每个Redis Cluster的节点——准确讲是master节点——负责一定范围的槽，所有节点组成的集群覆盖了0~16383整个范围的槽。
Redis Cluster的槽位空间是自定义分配的，类似于Windows盘分区的概念。这种分区是可以自定义大小，自定义位置的。Redis Cluster包含了16384个哈希槽，每个Key通过计算后都会落在具体一个槽位上，而这个槽位是属于哪个存储节点的，则由用户自己定义分配。例如机器硬盘小的，可以分配少一点槽位，硬盘大的可以分配多一点。如果节点硬盘都差不多则可以平均分配。
3.Redis集群实现原理
(1)所有的redis节点彼此互联(PING-PONG机制),内部使用二进制协议优化传输速度和带宽.
(2)节点的fail是通过集群中超过半数的master节点检测失效时才生效，所以集群中的节点数应该是2n+1.
(3)客户端与redis节点直连,不需要中间proxy层.客户端不需要连接集群所有节点,连接集群中任何一个可用节点即可
(4)redis-cluster把所有的物理节点映射到[0-16383]slot上,cluster 负责维护node<->slot<->key
4.Redis集群搭建
我们以搭建三主三备集群为例。
创建redis_cluster文件夹，在文件夹中创建docker-compose.yml文件
4.1 创建docker-compose.yml文件
version: '3.0'
services:
redis_1:
restart: always
image: daocloud.io/library/redis:5.0.9
container_name: redis_1
environment:
- TZ=Asia/Shanghai
ports:
- 7001:7001
- 17001:17001
volumes:
- ./redis/7001/redis.conf:/usr/local/etc/redis/redis.conf
command: ["redis-server","/usr/local/etc/redis/redis.conf"]
redis_2:
restart: always
image: daocloud.io/library/redis:5.0.9
container_name: redis_2
environment:
- TZ=Asia/Shanghai
ports:
- 7002:7002
- 17002:17002
volumes:
- ./redis/7002/redis.conf:/usr/local/etc/redis/redis.conf
command: ["redis-server","/usr/local/etc/redis/redis.conf"]
redis_3:
restart: always
image: daocloud.io/library/redis:5.0.9
container_name: redis_3
environment:
- TZ=Asia/Shanghai
ports:
- 7003:7003
- 17003:17003
volumes:
- ./redis/7003/redis.conf:/usr/local/etc/redis/redis.conf
command: ["redis-server","/usr/local/etc/redis/redis.conf"]
redis_4:
restart: always
image: daocloud.io/library/redis:5.0.9
container_name: redis_4
environment:
- TZ=Asia/Shanghai
ports:
- 7004:7004
- 17004:17004
volumes:
- ./redis/7004/redis.conf:/usr/local/etc/redis/redis.conf
command: ["redis-server","/usr/local/etc/redis/redis.conf"]
redis_5:
restart: always
image: daocloud.io/library/redis:5.0.9
container_name: redis_5
environment:
- TZ=Asia/Shanghai
ports:
- 7005:7005
- 17005:17005
volumes:
- ./redis/7005/redis.conf:/usr/local/etc/redis/redis.conf
command: ["redis-server","/usr/local/etc/redis/redis.conf"]
redis_6:
restart: always
image: daocloud.io/library/redis:5.0.9
container_name: redis_6
environment:
- TZ=Asia/Shanghai
ports:
- 7006:7006
- 17006:17006
volumes:
- ./redis/7006/redis.conf:/usr/local/etc/redis/redis.conf
command: ["redis-server","/usr/local/etc/redis/redis.conf"]
4.2 创建redis配置文件
在redis_cluster文件夹中，创建7001,7002,7003,7004,7005,7006 文件夹，在文件夹中创建redis.conf文件
redis.conf
#指定redis端口号
port 7001
#开启redis集群功能
cluster-enabled yes
#集群信息的文件，但这不是用户可编辑的配置文件，而是Redis群集节点每次发生更改时自动保留群集配置（基本上为状态）的文件，以便能够 在启动时重新读取它
cluster-config-file nodes-7001.conf
#集群对外端口号
cluster-announce-port 7001
#集群的总线端口号 集群内部节点通讯的端口号
cluster-announce-bus-port 17001
#Redis集群中节点允许不可用的最长时间，而不会将其视为失败。 如果主节点超过指定的时间不可达，它将由其从属设备进行故障切换
cluster-node-timeout 5000
另外5个配置文件相同，只需要将端口号修改一下。
4.3 执行docker-compose创建6个redis服务器
docker-compose up -d
4.4 将6个redis节点搭建成集群
进入任何一个redis节点：
docker exec -it 容器id bash
输入命令：
redis-cli --cluster create 42.192.12.30:7001 42.192.12.30:7002 42.192.12.30:7003 42.192.12.30:7004 42.192.12.30:7005 42.192.12.30:7006 --cluster-replicas 1
#--cluster-replicas 1 表示跟随的备用从节点是几个
输入yes
4.5 测试
直接连接某个节点，没有开启自动跳转的话，如果当前输入的key不在此节点就会报错
连接节点时，需要加上-c，代表运行节点间跳转
5.Java怎么连接redis集群
创建测试类：
package com.example.testjedis.test;
import org.junit.Test;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;
import java.util.HashSet;
import java.util.Set;
public class Demo04 {
@Test
public void testCluster()
{
//创建集群中的节点集合
Set<HostAndPort> nodes=new HashSet<HostAndPort>();
nodes.add(new HostAndPort("42.192.12.30",7001));
nodes.add(new HostAndPort("42.192.12.30",7002));
nodes.add(new HostAndPort("42.192.12.30",7003));
nodes.add(new HostAndPort("42.192.12.30",7004));
nodes.add(new HostAndPort("42.192.12.30",7005));
nodes.add(new HostAndPort("42.192.12.30",7006));
//创建集群
JedisCluster jedisCluster=new JedisCluster(nodes);
//集群中取值 可以从集群中任何节点中取
String value=jedisCluster.get("name");
System.out.println(value);
jedisCluster.close();
}
}
6.Springboot中怎么整合RedisCluster
Springboot中怎么整合RedisCluster的方式有很多，比如使用redisTemplate，或者原生jedis，redisTemplate方式配置方便，但性能没有原生jedis好。
我们这里使用的是原生jedis，自己配置。
6.1添加redis集群配置信息
6.2 修改配置文件类
添加集群信息的读取
@Bean
public JedisCluster jedisCluster(@Value("${jedis.cluster.nodesString}")String nodesString)
{
String[] nodes=nodesString.split(",");
Set<HostAndPort> nodeSet=new HashSet<HostAndPort>(nodes.length);
for(String node:nodes)
{
String[] nodeAttrs=node.trim().split(":");
HostAndPort hap=new HostAndPort(nodeAttrs[0],Integer.valueOf(nodeAttrs[1]));
nodeSet.add(hap);
}
JedisCluster jedisCluster=new JedisCluster(nodeSet);
return jedisCluster;
}
6.3 创建测试类测试
十二 Redis应用中常见问题
1.什么是缓存雪崩？
目前电商首页以及热点数据都会用redis做缓存 ，一般缓存都是定时任务去刷新，或者是查不到之后去更新的，定时任务刷新就有一个问题。
举个简单的例子：如果所有首页的Key失效时间都是12小时，中午12点刷新的，然后零点有个秒杀活动大量用户涌入，假设当时每秒 10万 个请求，本来缓存在可以扛住每秒 10万 个请求，但是缓存当时所有的Key都失效了。此时 1 秒 10万 个请求全部落数据库，数据库可能就直接挂了。这就是缓存雪崩。
解决方案:
1.在批量往redis中存数据的时候，把每个Key的失效时间都加个随机值，这样可以保证数据不会在同一时间大面积失效。
2.如果redis是集群部署，将热点数据均匀分布在不同的redis库中也能避免全部失效的问题。
3.设置热点数据永远不过期，一旦有更新操作就更新缓存。
2.什么是缓存穿透？
用户发送的请求数据，在缓存和数据库中都没有，如果同时有大量的这种请求发生，就会导致请求因为在缓存中找不到，而全部都进入数据库，导致数据库宕机。这就是缓存穿透。
比如：加入我们数据库的 id 是从1开始自增上去的，而发起为id值为 -1 的或 id 为特别大不存在的查询。
解决方案:
1.在接口层增加校验，比如用户鉴权校验，参数做校验，不合法的参数直接代码Return，比如：id 做基础校验，id <=0的直接拦截等。
2.将对单个IP每秒访问次数超出阈值的IP拉黑。
3.什么是缓存击穿？
缓存击穿是指一个Key非常热点，在不停的扛着大并发，大并发集中对这一个点进行访问，当这个Key在失效的瞬间，持续的大并发就穿破缓存，直接请求数据库，导致数据库宕机。
解决方案:
1.定时任务主动刷新缓存
启动定时任务，在redis缓存失效之前,主动查数据库更新缓存。
2.使用互斥锁
在根据key获得的value值为空时，先锁上，再从数据库加载，加载完毕，释放锁。若其他线程发现获取锁失败，则睡眠50ms后重试。至于锁的类型，单机环境用并发包的Lock类型就行，集群环境则使用分布式锁( 比如redis的setnx)
3.jvm缓存+redis缓存的多级缓存
服务器只会在jvm缓存失效,且redis缓存也失效的情况下才会查询数据库，而多个服务器的jvm缓存失效时间是随机值,所以很大程度上避免的同时失效去查库的情况，由于所有服务器jvm缓存同时失效redis缓存也失效的可能性极低,所以数据库上重复的查询会很少
ter.get(“name”);
System.out.println(value);
jedisCluster.close();
}
}
[外链图片转存中...(img-65lVE8hw-1669557586934)]
#### 6.Springboot中怎么整合RedisCluster
Springboot中怎么整合RedisCluster的方式有很多，比如使用redisTemplate，或者原生jedis，redisTemplate方式配置方便，但性能没有原生jedis好。
我们这里使用的是原生jedis，自己配置。
##### 6.1添加redis集群配置信息
[外链图片转存中...(img-Csxs0vCO-1669557586934)]
##### 6.2 修改配置文件类
添加集群信息的读取
@Bean
public JedisCluster jedisCluster(@Value(“${jedis.cluster.nodesString}”)String nodesString)
{
String[] nodes=nodesString.split(“,”);
Set<HostAndPort> nodeSet=new HashSet<HostAndPort>(nodes.length);
for(String node:nodes)
{
String[] nodeAttrs=node.trim().split(":");
HostAndPort hap=new HostAndPort(nodeAttrs[0],Integer.valueOf(nodeAttrs[1]));
nodeSet.add(hap);
}
JedisCluster jedisCluster=new JedisCluster(nodeSet);
return jedisCluster;
}
[外链图片转存中...(img-ZE43oEAp-1669557586935)]
##### 6.3 创建测试类测试
[外链图片转存中...(img-VfUi7Cyj-1669557586935)]
#### [外链图片转存中...(img-UknG1VAF-1669557586935)]
### 十二 Redis应用中常见问题
Author：呆萌老师 QQ：2398779723 微信：it_daimeng
#### 1.什么是缓存雪崩？
目前电商首页以及热点数据都会用redis做缓存 ，一般缓存都是定时任务去刷新，或者是查不到之后去更新的，定时任务刷新就有一个问题。
举个简单的例子：如果所有首页的Key失效时间都是12小时，中午12点刷新的，然后零点有个秒杀活动大量用户涌入，假设当时每秒 10万 个请求，本来缓存在可以扛住每秒 10万 个请求，但是缓存当时所有的Key都失效了。此时 1 秒 10万 个请求全部落数据库，数据库可能就直接挂了。这就是缓存雪崩。
解决方案:
1.在批量往redis中存数据的时候，把每个Key的失效时间都加个随机值，这样可以保证数据不会在同一时间大面积失效。
2.如果redis是集群部署，将热点数据均匀分布在不同的redis库中也能避免全部失效的问题。
3.设置热点数据永远不过期，一旦有更新操作就更新缓存。
#### 2.什么是缓存穿透？
用户发送的请求数据，在缓存和数据库中都没有，如果同时有大量的这种请求发生，就会导致请求因为在缓存中找不到，而全部都进入数据库，导致数据库宕机。这就是缓存穿透。
比如：加入我们数据库的 id 是从1开始自增上去的，而发起为id值为 -1 的或 id 为特别大不存在的查询。
解决方案:
1.在接口层增加校验，比如用户鉴权校验，参数做校验，不合法的参数直接代码Return，比如：id 做基础校验，id <=0的直接拦截等。
2.将对单个IP每秒访问次数超出阈值的IP拉黑。
#### 3.什么是缓存击穿？
缓存击穿是指一个Key非常热点，在不停的扛着大并发，大并发集中对这一个点进行访问，当这个Key在失效的瞬间，持续的大并发就穿破缓存，直接请求数据库，导致数据库宕机。
解决方案:
1.定时任务主动刷新缓存
启动定时任务，在redis缓存失效之前,主动查数据库更新缓存。
2.使用互斥锁
在根据key获得的value值为空时，先锁上，再从数据库加载，加载完毕，释放锁。若其他线程发现获取锁失败，则睡眠50ms后重试。至于锁的类型，单机环境用并发包的Lock类型就行，集群环境则使用分布式锁( 比如redis的setnx)
3.jvm缓存+redis缓存的多级缓存
服务器只会在jvm缓存失效,且redis缓存也失效的情况下才会查询数据库，而多个服务器的jvm缓存失效时间是随机值,所以很大程度上避免的同时失效去查库的情况，由于所有服务器jvm缓存同时失效redis缓存也失效的可能性极低,所以数据库上重复的查询会很少