NoSQL(NoSQL = Not Only SQL )，意即“不仅仅是SQL”
泛指非关系型的数据库。随着互联网web2.0网站的兴起，传统的关系数据库在应付web2.0网站，特别是超大规模和高并发 的SNS类型的web2.0纯动态网站已经显得力不从心，暴露了很多难以克服的问题，而非关系型的数据库则由于其本身的特 点得到了非常迅速的发展。NoSQL数据库的产生就是为了解决大规模数据集合多重数据种类带来的挑战，尤其是大数据应 用难题，包括超大规模数据的存储。 (例如谷歌或Facebook每天为他们的用户收集万亿比特的数据)。这些类型的数据存储不需要固定的模式，无需多余操作 就可以横向扩展。
NoSQL数据库种类繁多，但是一个共同的特点都是去掉关系数据库的关系型特性。数据之间无关系，这样就非常容易扩展。也无形之间，在架构的层面上带来了可扩展的能力。
NoSQL数据库都具有非常高的读写性能，尤其在大数据量下，同样表现优秀。
这得益于它的无关系性，数据库的结构简单。
一般MySQL使用Query Cache,每次表的更新Cache就失效，是一种大粒度的Cache, 在针对web2.0的交互频繁的应用，Cache性 能不高。而NoSQL 的Cache是记录级的， 是一种细粒度的Cache，所以NoSQL在这个层面上来说就要性能高很多了
RDBMS (关系型数据库)
-高度组织化结构化数据
-结构化查询语言(SQL)
-数据和关系都存储在单独的表中。
-数据操纵语言，数据定义语言
-严格的一致性
-基础事务
NoSQL (非关系型的数据库)
-代表着不仅仅是SQL
-没有声明性查询语言
-没有预定义的模式
-键-值对存储，列存储，文档存储，图形数据库
-最终一致性，而非ACID属性
-非结构化和不可预知的数据
-高性能，高可用性和可伸缩性
Reiids
Redis（Remote Dictionary Server )，即远程字典服务，是一个开源的使用ANSI C语言编写、支持网络、可基于内存亦可持久化的日志型、Key-Value数据库，并提供多种语言的API。从2010年3月15日起，Redis的开发工作由VMware主持。从2013年5月开始，Redis的开发由Pivotal赞助。
Redis作用
- 内存存储，持久化（rdb，aof）
- 效率高，可以用于高速缓存
- 发布订阅
- 地图信息分析
- 计时器，计数器
Redis特性
- 多样的数据类型
- 持久化
- 集群
- 事务
Redis是单线程的
Redis是很快的,官方表示, Redis是基于内存操作, CPU不是Redis性能瓶颈, Redis的瓶颈是根据机器的内存和网络带宽，既然可以使用单线程来实现,就使用单线程了!所有就使用了单线程了!
- 核心：redis将所有的数据全部放入内存里面，所以使用单线程去操作效率就是最高的，对于内存系统来说，如果没有上下文切换效率就是最高的，多次读写都是在一个cpu上的，在内存情况下这个就是最佳方案
五大数据类型以常用命令
首先Redis默认有16个数据库 0-15 ，默认使用第 0 个，可以使用select进行切换
move key db 把key移除到数据库
exists key 判断当前key是否存在
expire name seconds 设置当前key的过期时间
ttl key 查看当前key的剩余时间
tyoe 判断当前key的类型
String
set key
get key
del key
exists key
Append key value 追加字符 如果不存在创建
strlen 获得长度
incr 自增1
decr 自减1
incrby 设置步长 指定增量
decrby 指定减量
getrange key index index 截取
setrange key index value 替换指定位置的字符串
setex（set with expire） 设置过期时间
setnx（set if not exists） 不存在再设置 存在则创建失败 分布式锁中经常使用
mest 批量增加
mget 批量获取
Msetnx： 原子操作 都不存在再设置 存在任意一个值则创建失败
设置一个json字符串来保存对象
set user:y1 {name:y1,age:1}
set user:y1 {name:y1,age:1}
还可以用mset巧妙设计key 对象：id：属性
比如设置当前文章浏览量 对象：id：属性的格式
set article:1:views 0
getset 如果不存在值返回null设置新的值 如果存在就获取原来的值并设置新的值
getset
String 类似的使用场景：(value除了是字符串还可以是数字)
计数器
统计多单位的数量
粉丝数
对象缓存存储
List
在redis里面 可以把List当成 栈 和队列 阻塞队列
所有的list 命令都是以 “ l ” 开头的
List是先进后出的
lpush key value 将一个值或者多个值插入列表头部
rpush key value 将一个值或者多个值插入列表尾部
lrange key index index 返回队列的指定index之间的值
lpush 1 ———> 1 list 4 <——— 4 rpush
lpush 2 ———> 2 1 list 4 5 <——— 5 rpush
lpush 3 ———> 3 2 1 list 4 5 6 <——— 6 rpush
lrange list 0 -1
3 2 1 4 5 6
lpop key 左移
rpop key 右移
lpop 3 <——— 3 2 1 list 4 5 6 ———> 6 rpop
lpop 2 <——— 2 1 list 4 5 ———> 5 rpop
lpop 1 <——— 1 list 4 ———> 4 rpop
lindex key index 返回index 下标的值
lien key 获得list的长度
lrem key 1 value 移除一个值
lrem key n value 移除n个value
trim key index index 截取index下标指定的长度，list被改变，只剩下截取后的
rpoplpush key newKey 移除列表最后一个元素到新的列表
Exists Listkey 是否存在list
lset key index element 替换key的index下标的位置的值 如果key不存在报错
linsert key before/after element value 往key的element前面或者后面插入 value
list实际上是一个链表 before node after，left right 都可以插入
如果key不存在，创建新的链表
如果key存在，新增内容
如果移除了所有值，空链表，也代表不存在
在两边插入或者改动效率最高，中间元素效率偏低
你可以把List消看成息队列，消息排队
lpush rpop 队列
lpush lpop 栈
Set
Set中的值不能重复 无序
sadd key value set存入值
smembers key set查看所有值
sismember key value 判断是否set中存在value return 1 & 0
scard key 获取set中元素个数
srem key value 移除set中value元素
srandmember key 随机抽取一个元素。（做抽奖）
srandmember key num 随机抽取num个元素
spop key num 随机弹出 num个元素
smove key1 ket2 member 将member从key1移动到key2. 成功返回1失败0（当有相同的元素时失败）
数字集合类
差集 sdiff key1 key2
交集 sinter key1 key2
并集 suniom key1 key2
应用
共同关注（并集）
共同关注 A用户将所有关注的人放在一个集合当中 将他的粉丝关注的人放在一个集合中
公共爱好 共同好友 推荐好友
Hash
k-map k-<key-value>
hset key field value 设置一个名字为key的hash，字段名(key)是 field 值是value 如果存在field就替换
hget key field field 取出key里面的 field对应的 value
hgetall key 获取全部key里面的键值对
hdel key field 删除hash指定的字段
hlen key 获取hash里面键值对的数量
hexists key field 判断hash的指定字段是否存在 return 1,0
hkeys key 只获得所有field
hvals 只获得所有的value
hincrby key Field num key里面的field字段加num
hsetnx key field value 如果key里面不存在field子段就创建并且设置value 如果存在返回0不能设置
hash作为用户信息 变更信息保存
user 当成 hash-key，name age 作为field
string更适合字符串存储
ZSet（有序集合）
在set的基础上 (sadd k1 v1) 增加了一个值
zadd k1 score1 v1 score2 v2 添加一个值或者多个值
zrange key index index 查看index区间内的值
zrevrange Key 0 -1 从大到小排序
zrevrange Key 0 -1 withscores 从大到小排序附带score
zrevrange Key 0 num withscores 从大到num排序附带score
zrangebyscore key -inf +inf 显示所有的值 从小到大排序
zrangebyscore key -inf +inf withscores 显示所有的值 从小到大排序 并且附带score
zrangebyscore key -inf num withscores 显示所有的值 从小到num排序 并且附带score
zrem key member 移除一个指定元素 member
zcard key 获取有序集合中的元素数量
zcount key min max 获取指定区间的members数量
案例 存储班级成绩 工资
消息权重 score分数权重判断
排行榜 实现
geospatial （地理位置）
六个命令
GEOADD
GEODIST
GEOHASH
GEOPOS
GEORADIUS
GEORADIUSBYMEMBER
GEOADD 添加地理位置
规则两级无法添加，一般下载城市数据，直接通过java程序一次性导入
参数。key 值（经度，纬度，名称）
GEOADD key 经度 纬度 名称
GEOPOS key 名称
获得当前定位，一定是一个坐标值
两人之间的距离（直线距离）
单位
m km ml英里 ft英尺
GEODIST key member member (单位) 默认是米
GEORADIUS 以给定的经度纬度为中心，找出半径内的元素
附近的人实现
获得所有附近的人的地址，手机定位 ，通过半径来查询
GEORADIUS key 经度 纬度 半径 单位 (参数)
带了参数的
GEORADIUSBYMEMBER 根指定元素为中心 查找半径内的城市
GEORADIUSBYMEMBER key member 半径 单位 （参数）
GEOHASH 返回一个或者多个位置元素的GEOHASH表示 11个长度的GEOHASH 字符串 字符串越像表示距离越近
GEOHASH key member
GEO 底层实现原理其实就是zset 可以使用zset来操作GEO
查看地图中所有元素
移除北京的定位（移除一个元素）
hyperloglog
什么是基数
基数 ：一个集合内不重复的元素个数
redis 2.8.9版本就更新了 hyperloglog 基数统计算法
网页 uv（一个人访问一个网站多次，但是还是算做一个人）
传统方式，set保存用户id，统计set中元素数量
很多分布式id很长所以保存起来比较麻烦，我们目的是计数，为不是保存用户id
hyperloglog优点 占用内存固定 比如 放入 2^64不同的元素的基数，只需要12kb内存 如果从内存角度比较的话 hyperloglog首选。错误率 0.81%
使用
pfadd key [value,value…..] 创建一组元素
pfcount key 统计元素数量
pfcount key key2 统计key，key2 元素并集数量
pfmerge newkey key1 key2 合并key1 key2 的并集到newkey
不允许容错 就用set
Bitmap 位存储
比如统计疫情感染人数：0 0 0 0 0 0 0 0 0 1 0 1 0 1 0
0未感染 1感染
统计用户信息 活跃 不活跃 登录 未登录 365打卡
两个状态的都可以使用 bitmap
bitmaps位图 数据结构都是操作2进制位来进行记录
1字节=8bit 一个人一年365bit 约等于 46字节 效率高 省内存
0-6 表示 周一到周六 0未打卡 1打卡
查看某一天是否打卡
统计打卡天数
Bitcount key start end
事务
事务的本质：一组命令的集合 一个事务中的所有命令都会被序列化 在事务执行的过程中，会按照顺序执行
一次性，顺序性，排他性执行一系列命令 [set,set,set]
redis事务没有隔离级别的概念 所有命令在入队的时候并没有被执行 只有发起执行命令的时候才会被执行 Exec
redis事务：
开启事务 multi
命令入队 命令
执行事务 exec
正常执行事务
命令报错事物停止 命令不会执行 （编译型异常）
运行时异常 运行的时候其他命令可以执行
放弃事物 discard 队列中的命令都不会执行
监控
redis单条命令保证原子性，但是事务不保证原子性
redis的监视测试 redis可以实现乐观锁
正常执行成功 事物正常结束 数据没有发生变动 watch 监视money对象
开启两个进程终端
第一个不做执行
第二个进程直修改money数据
此时执行第一个进程 修改失败
使用watch可以当做乐观锁
如果事务执行失败先解锁 获取最新的值 再次监视 比对监视的值是否发生变化 如果没变化可以执行成功
jedis
要使用java来操作redis
什么是jedis:
官方推荐的java连接开发工具 使用java操作redis的中间件
1 导入对应的依赖
<!-- 导入jedis的包-->
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>3.2.0</version>
</dependency>
2编码测试
连接--->操作--->结束
public class TestPing
{
public static void main(String[] args)
{
//new一个jedis对象
Jedis jedis = new Jedis("127.0.0.1",6379);
//所有的命令就是我们上面的所有指令
System.out.println("测试链接："+jedis.ping());
System.out.println("清空当前数据库数据："+jedis.flushDB());
System.out.println("判断某个键是否存在："+jedis.exists("username"));
System.out.println("新增'username'，'Y1'的键值对："+jedis.set("username","Y1"));
System.out.println("新增'password'，'Y1'的键值对："+jedis.set("password","Y1"));
System.out.println("系统中所有的键如下："+"\n"+jedis.keys("*"));
System.out.println("删除键password："+jedis.del("password"));
System.out.println("判断键password是否存在："+jedis.exists("password"));
System.out.println("查看username所储存值的类型："+jedis.type("username"));
System.out.println("随机返回key空间的一个："+jedis.randomKey());
System.out.println("重命名key："+jedis.rename("username","name"));
System.out.println("取出改后的name："+jedis.get("name"));
System.out.println("选择0号数据库："+jedis.select(0));
System.out.println("删除当前库所有keys："+jedis.flushDB());
System.out.println("返回当前库keys数量："+jedis.dbSize());
System.out.println("删除所有keys："+jedis.flushAll());
System.out.println("增加："+jedis.set("k1","v1"));
System.out.println("增加："+jedis.set("k2","v2"));
System.out.println("增加："+jedis.set("k3","v3"));
System.out.println("删除k2："+jedis.del("k2"));
System.out.println("获取k2："+jedis.get("k2"));
System.out.println("修改k1："+jedis.set("k1","newV1"));
System.out.println("获取k1："+jedis.get("k1"));
System.out.println("在k3后面增加："+jedis.append("k3","append"));
System.out.println("获取k3："+jedis.get("k3"));
System.out.println("增加多个键值对："+jedis.mset("k4","v4","k5","v5","k6","v6"));
System.out.println("获取多个键值对："+jedis.mget("k4","kk5","k6"));
System.out.println("删除多个键值对："+jedis.del("k4","k5"));
System.out.println("获取多个键值对："+jedis.mget("k4","kk5","k6"));
System.out.println("清库："+jedis.flushAll());
System.out.println("==========分割线==========");
System.out.println("新增防止覆盖之前的：");
System.out.println("k1："+jedis.setnx("k1","v1"));
System.out.println("k2："+jedis.setnx("k2","v2"));
System.out.println("k2："+jedis.setnx("k2","v3"));
System.out.println("获取k1："+jedis.get("k1"));
System.out.println("获取k2："+jedis.get("k2"));
System.out.println("==========分割线==========");
System.out.println("新增键值对设置有效时间");
System.out.println("set k1："+jedis.setex("k1",2,"2s"));
System.out.println("get k1："+jedis.get("k1"));
try
{
TimeUnit.SECONDS.sleep(3);
}
catch (InterruptedException e)
{
e.printStackTrace();
}
System.out.println("get k1："+jedis.get("k1"));
System.out.println("获取原值更新为新值");
System.out.println("getset k2："+jedis.getSet("k2","getset k2"));
System.out.println("get k2："+jedis.get("k2"));
System.out.println("获取k2的剪切字串："+jedis.getrange("k2",2,4));
}
}
所有api都是上面的命令 一个都没有变化 全部一样
Jedis事务
public class TestTX
{
public static void main(String[] args)
{
Jedis jedis = new Jedis("127.0.0.1",6379);
System.out.println(jedis.ping());
JSONObject jsonObject = new JSONObject();
jsonObject.put("hello","world");
jsonObject.put("name","Y1");
jedis.flushDB();
//jedis.watch("user1","user2");
//开启事物
Transaction multi = jedis.multi();
String result = jsonObject.toJSONString();
try
{
multi.set("user1",result);
multi.set("user2",result);
//代码跑出编译异常事务执行失败
int i =1/0;
//执行事物
multi.exec();
}
catch (Exception e)
{
//放弃事物
multi.discard();
e.printStackTrace();
}
finally
{
System.out.println(jedis.get("user1"));
System.out.println(jedis.get("user2"));
jedis.close();
}
}
}
SpringBoot整合
SpringBoot操作数据：
SpringData：jpa jdbc mongodb redis
说明在springboot2.X之后我们原来的用的jedis被换成了 lettuce
jedis ： 采用直接连接，多个线程操作的话，是不安全的，如果想要避免不安全，使用jedis 的 pool 连接池 BIO
lettuce：采用netty，实例可以在多个线程中共享，不存在线程不安全的情况，可以减少线程数量，更像 NIO
源码分析：
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(RedisOperations.class)
@EnableConfigurationProperties(RedisProperties.class)
@Import({ LettuceConnectionConfiguration.class, JedisConnectionConfiguration.class })
public class RedisAutoConfiguration {
@Bean
@ConditionalOnMissingBean(name = "redisTemplate”) //我们可以自己定一个redisTemplate 替换默认的
public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory)
throws UnknownHostException {
//默认的redisTemplate 没有过多的设置， redis的对象保存 都是需要序列化的
//两个范性都是object类型后面使用需要强转<String, Object>
RedisTemplate<Object, Object> template = new RedisTemplate<>();
template.setConnectionFactory(redisConnectionFactory);
return template;
}
@Bean
@ConditionalOnMissingBean。//由于String类型是redis最最常用的一个类型，所以单独提出来一个bean
public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory redisConnectionFactory)
throws UnknownHostException {
StringRedisTemplate template = new StringRedisTemplate();
template.setConnectionFactory(redisConnectionFactory);
return template;
}
}
整合测试一下
springboot 所有的配置 都有一个自动配置类 autoconfigure
自动配置类会绑定一个 properties 配置文件
1，导入依赖，配置连接，测试
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
spring:
redis:
host: 127.0.0.1
port: 6379
@SpringBootTest
class RedisSpringbootApplicationTests
{
@Resource
RedisTemplate redisTemplate;
@Test
void contextLoads()
{
// 操作字符串
redisTemplate.opsForValue();
//操作list
redisTemplate.opsForList();
//hash
redisTemplate.opsForHash();
//set
redisTemplate.opsForSet();
//zset
redisTemplate.opsForZSet();
//geo
redisTemplate.opsForGeo();
//hyperloglog
redisTemplate.opsForHyperLogLog();
//bitmap
redisTemplate.opsForValue().setBit("key",new Long(12345678),true);
//连接对象
RedisConnection connection = redisTemplate.getConnectionFactory().getConnection();
connection.flushAll();
connection.flushDb();
}
}
值的序列化
默认的jdk序列化方式（字符串会转译） 我会要使用json来序列化
所有的对象需要序列化 不然会报错
关于对象的保存:
Pojo 类都会序列化
redis 里实现 Serializable 接口
默认是jdk序列化
我们要自己实现序列化方式
源码的序列化方式 默认使用jdk序列化方式
自己的redistemplate序列化配置模版
@Configuration
public class RedisConfig
{
//编写我们自己的 ridesTemplate
@Bean
@SuppressWarnings("all")
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory)
throws UnknownHostException
{
RedisTemplate<String, Object> template = new RedisTemplate<>();
template.setConnectionFactory(redisConnectionFactory);
//序列化配置
//使用json去解析任意的对象 用json做序列化
Jackson2JsonRedisSerializer jsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
//用obojectmapper进行转译
ObjectMapper objectMapper = new ObjectMapper();
objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
jsonRedisSerializer.setObjectMapper(objectMapper);
//string的序列化
StringRedisSerializer stringRedisTemplate = new StringRedisSerializer();
// key 采用string的序列化方式
template.setKeySerializer(stringRedisTemplate);
// hash 的key 也采用string的序列化方式
template.setHashKeySerializer(stringRedisTemplate);
//value的序列化采用json的方式
template.setValueSerializer(jsonRedisSerializer);
//hash的value序列化采用json的方式
template.setHashValueSerializer(jsonRedisSerializer);
//把properties set 进去
template.afterPropertiesSet();
//返回template
return template;
}
}
序列化之后的 redis-cli 正常显示key
Redis config 的配置
启动的时候，就通过配置文件来启动
单位 配置文件 对单位大小写不敏感
网络
端口
通用配置
快照
主从复置
安全
验证
通过命令来设置
客户端
1、volatile-lru：只对设置了过期时间的key进行LRU（默认值）
2、allkeys-lru ： 删除lru算法的key
3、volatile-random：随机删除即将过期key
4、allkeys-random：随机删除
5、volatile-ttl ： 删除即将过期的
6、noeviction ： 永不过期，返回错误
AOF 配置
一些小小的配置可以改变性能！！！
持久化
在指定的时间间隔内将内存中的数据集快照写入磁盘,也就是行话讲的Snapshot快照,它恢复时是将快照文件直接读到内存里。Redis会单独创建( fork ) 一个子进程来进行持久化,会先将数据写入到一个临时文件中,待持久化过程都结束了,再用这个临时文 件替换上次持久化好的文件。整个过程中,主进程是不进行任何I0操作的。这就确保了极高的性能。如果需要进行大规模数据的恢复，且对于数据恢复的完整性不是非常敏感,那RDB方式要比AOF方式更加的高效。RDB的缺点是最后一次持久化后的数据可能丢失。
一般情况下 不需要修改配置
Rdb保存的文件是 dump.rdb
自定义 rdb 机制 60s 内修改了 5次 触发快照策略
触发机制
1 save的规则满足的情况下 会自动触发rdb规则 生成dump.rdb文件
2执行了flushall命令 会出发rdb规则 生成dump.rdb文件
3退出redis的时候 也会触发rdb规则 生成dump.rdb文件
如何恢复rdb文件
只需要将rbd文件放入redis的启动目录就可以了，redis启动的时候会自动检查dump.rdb文件 恢复其中数据
查看dump文件存放的位置
默认配置基本够用
优点：适合大规模数据恢复，对于数据恢复的完整性不是非常敏感,那RDB方式要比AOF方式更加的高效
缺点：需要一定的时间间隔操作，如果意外宕机最后一次修改数据丢失，fork进程会占用一定的内存空间
生产环境会将rdb备份
AOF 记录所有的命令，恢复的时候就把这个文件全部的再执行一遍
以日志的形式来记录每个写操作，将Reids执行过的写入指令记录下来，只许追加文件但不可以改写文件，redis启动之初会读取aof文件重新构建数据，换言之，redis重启的话就根据日志文件的内容将写指令全部执行一次以完成数据的恢复工作。
如果aof文件被恶心修改、损坏，redisg是不能启动的，此时可以使用redis-check-aof来进行修复
默认不开启需要手动配置
若aof文件大于指定值（64mb），会fork一个新子进程将文件重写
重启redis aof生效 生成appendonly.aof文件 以日志级别记录写操作
如果aof文件出错 redis无法启动 需要修复文件
但是修复后，可能会丢失一些数据
优点
每一次修改都同步，确保文件的完整性
每秒同步一次，可能会丢失一秒的数据
从不同步，效率最高
但是修复后，可能会丢失一些数据
优点
每一次修改都同步，确保文件的完整性
每秒同步一次，可能会丢失一秒的数据
从不同步，效率最高
缺点
相对于数据文件来说，aof远远大于rdb，修复速度比rdb慢
aof运行效率要比rdb慢（涉及到大量的IO操作）
拓展
1.Redis能对AOF文件进行重写，使得AOF文件体积不至于过大
2.只做缓存，只希望数据在服务器运行的时候存在，也可以不做任何持久化
3.同时开启两种持久化方式：
redis重启会优先载入AOF文件来恢复原始数据，因为通常情况下AOF文件保存的数据比RDB要完整
RDB数据不实时，更适合用于备份数据库（AOF在不断变化不好备份），快速重启，而且不会有AOF可能潜在的bug
1、RDB持久化方式能够在指定的时间间隔内对你的数据进行快照存储
2、AOF持久化方式记录每次对服务器写的操作,当服务器重启的时候会重新执行这些命令来恢复原始的数据, AOF命令以Redis协议追加保存每次写的操作到文件末尾，Redis还能对AOF文件进行后台重写,使得AOF文件的体积不至于过大。
3、只做缓存,如果你只希望你的数据在服务器运行的时候存在,你也可以不使用任何持久化
4、同时开启两种持久化方式
●在这种情况下,当redis重启的时候会优先载入AOF文件来恢复原始的数据,因为在通常情况下AOF文件保存的数据集要比RDB 文件保存的数据集要完整。
●RDB的数据不实时,同时使用两者时服务器重启也只会找AOF文件,那要不要只使用AOF呢?作者建议不要,因为RDB更适合 用于备份数据库( AOF在不断变化不好备份) ,快速重启,而且不会有AOF可能潜在的Bug ,留着作为一个万一的手段。
5、性能建议
●因为RDB文件只用作后备用途,建议只在Slave.上持久化RDB文件,而且只要15分钟备份一次就够了
●如果Enable AOF ,好处是在最恶劣情况下也只会丢失不超过两秒数据,启动脚本较简单只load自己的AOF文件就可以了,代价 一是带来了持续的IO ,二是AOF rewrite的最后将rewrite过程中产生的新数据写到新文件造成的阻塞几乎是不可避免的。只要 硬盘许可,应该尽量减少AOF rewrite的频率, AOF重写的基础大小默认值64M太小了,可以设到5G以上,默认超过原大小 100%大小重写可以改到适当的数值。
●如果不Enable AOF，仅靠Master-Slave Repllcation实现高可用性也可以,能省掉一大笔IO，也减少了rewrite时带来的系统波动。代价是如果Master/Slave 同时倒掉，会丢失十几分钟的数据，启动脚本也要比较两个Master/Slave中的RDB文件,载入较新的那个，微博就是这种架构。
发布订阅
发送
订阅
原理
Redis是使用C实现的,通过分析Redis 源码里的pubsub.c文件，了解发布和订阅机制的底层实现,籍此加深对Redis 的理解。
Redis通过PUBLISH、SUBSCRIBE 和PSUBSCRIBE等命令实现发布和订阅功能。
通过SUBSCRIBE命令订阅某频道后, redis-server 里维护了一个字典,字典的键就是一个个channel , 而字典的值则是一个链 表，链表中保存了所有订阅这个channel的客户端。SUBSCRIBE 命令的关键,就是将客户端添加到给定channel的订阅链表中。
通过PUBLISH命令向订阅者发送消息, redis-server会使用给定的频道作为键,在它所维护的channel字典中查找记录了订阅这个频道的所有客户端的链表,遍历这个链表,将消息发布给所有订阅者。
Pub/Sub从字面上理解就是发布( Publish )与订阅( Subscribe ) , 在Redis中,你可以设定对某-个key值进行消息发布及消息订 阅，当一个key值上进行了消息发布后,所有订阅它的客户端都会收到相应的消息。这一功能最明显的用法就是用作实时消息系 统，比如普通的即时聊天,群聊等功能。
简单应用有：
实时消息系统
聊天室
关注系统
主从复制
概念
主从复制,是指将一台Redis服务器的数据,复制到其他的Redis服务器。前者称为主节点(master/leader) ,后者称为从节点(slave/follower) ;数据的复制是单向的,只能由主节点到从节点。Master以写为主, Slave以读为主。
默认情况下,每台Redis服务器都是主节点
且一个主节点可以有多个从节点(或没有从节点) ,但一个从节点只能有一个主节点。
主从复制的作用主要包括:
1、数据冗余:主从复制实现了数据的热备份,是持久化之外的一种数据冗余方式。
2、故障恢复:当主节点出现问题时,可以由从节点提供服务,实现快速的故障恢复;实际上是一种服务的冗余。
3、负载均衡:在主从复制的基础上,配合读写分离,可以由主节点提供写服务,由从节点提供读服务(即写Redis数据时应用连接 主节点,读Redis数据时应用连接从节点) , 分担服务器负载;尤其是在写少读多的场景下,通过多个从节点分担读负载,可以大 大提高Redis服务器的并发量。
4、高可用基石:除了上述作用以外，主从复制还是哨兵和集群能够实施的基础,因此说主从复制是Redis高可用的基础。 一般来说,要将Redis运用于工程项目中,只使用一台Redis是万万不能的,原因如下:
1、从结构上,单个Redis服务器会发生单点故障,并且一台服务器需要处理所有的请求负载,压力较大;
2、从容量上,单个Redis服务器内存容量有限,就算一台Redis服务器内存 容量为256G ,也不能将所有内存用作Redis存储内存, 一般来说 ,单台Redis最大使用内存不应该超过20G。 电商网站上的商品，一般都是一次上传,无数次浏览的,说专业点也就是"多读少写"。
主从复制，读写分离，80%的情况下都是在进行读操作，减缓服务器压力，架构中经常使用，一主二从
只要在公司中，主从复置必须要使用，因为在真实的项目中不可能单机使用redis
环境配置
只配置从库，不配置主库
查看当前库的信息，redis默认是主库
修改
端口 pid名字 log文件名 dump.rdb名字 开启后台启动
修改主机的配置文件
port 6379
daemonize yes
pidfile /var/run/redis_6379.pid
logfile "6379.log"
dbfilename dump6379.rdb
修改从机的配置文件
port 6380
daemonize yes
pidfile /var/run/redis_6380.pid
logfile "6380.log"
dbfilename dump6380.rdb
通过配置文件开启服务
一主二从
开启三个服务发现默认都是主机
一般情况下只用配置从机就可以了
6379为主机
80 81 为从机
这个是暂时的配置
真实的配置是应该修改配置文件的
⚠️
主机写 从机读
主机宕机从机还是可以拿到值，只是没有了写操作
主机从新开机，从机仍然可以获取主机的写入
⚠️ 用命令行配置的主从
从机宕机 主机在宕机期间写入的从机取不到值
因为断开以后重启 读取未修改的配置文件默认还是主节点启动
复制原理
Slave启动成功连接到master后会发送一个sync同步命令
Master接到命令,启动后台的存盘进程,同时收集所有接收到的用于修改数据集命令,在后台进程执行完毕之后, master将传送 整个数据文件到slave ,并完成一次完全同步。
全量复制:slave服务在接收到数据库文件数据后,将其存盘并加载到内存中。
增量复制: Master继续将新的所有收集到的修改命令依次传给slave ,完成同步
但是只要是重新连接master , 一次完全同步(全量复制)将被自动执行
主机断开连接使用这个命令是自己变为主机
SLAVEOF no one
哨兵模式
哨兵配置
配置哨兵配置文件 vim sentinel.conf
配置远远不止这么少
启动哨兵
redis-sentinel sentinel.conf
开启三个redis服务 79 80 81
设置主从后 哨兵日志更新
模拟主机宕机
failover 故障转移
哨兵选举6380为新的主机 （投票算法）
主机从新链接了也只能当作是6380的从机 （哨兵模式规则）
优点:
1、哨兵集群,基于主从复制模式,所有的主从配置优点,它全有
2、主从可以切换,故障可以转移,系统的可用性就会更好
3、哨兵模式就是主从模式的升级,手动到自动,更加健壮!
缺点:
1、Redis 不好在线扩容的,集群容量一但到达上限,在线扩容就十分麻烦!
2、实现哨兵模式的配置其实是很麻烦的,里面有很多选择!
基础配置
protected-mode no #关闭保护模式
port 26479 #端口
daemonize yes #使用后台模式启动
pidfile "/var/run/redis-sentinel_26479.pid" #进程id文件
logfile "/usr/local/redis/sentinel/sentinel_26479.log" #日志文件
dir "/usr/local/redis/sentinel" #工作目录
核心配置
1、sentinel monitor <master-name> <ip> <port> <quorum>
master-name：redis主节点昵称。
ip：redis主机ip。
port：redis主机端口。
quorum：哨兵判断主节点是否发生故障的票数。如果设置为2，表示2个哨兵节点认为主节点发生了故障，一般设置为：哨兵节点数/2+1。
2、sentinel down-after-milliseconds <master-name> <times>
哨兵会定期的向redis节点发送ping命令来判断redis是否可达，若超过指定的times毫秒内还未得到pong回复，则判读该redis不可达。
3、sentinel parallel-syncs <master-name> <nums>
当redis主节点挂了后，哨兵会选出新的master，此时，剩余的slave会向新的master发起同步数据，这个设置表示允许并行同步的slave个数。
4、sentinel failover-timeout <master-name> <times>
进行故障转移时，如果超过设置的times毫秒，表示故障转移失败。
5、sentinel auth-pass <master-name> <password>
如果redis主节点设置了密码，则需要进行这个配置。
*****备注配置redis主从复制、读写分离*******
配置思路：master配置文件不需要动，修改slave的配置文件。
1、添加一行：replicaof <masterip> <masterport>
2、如果master配置有密码，则需要配置这一行
masterauth <master-password>
3、replica-read-only yes #表示slave中的数据是只读的
*****springboot整合redis哨兵模式*******
添加yml配置文件
spring:
redis:
database: 0
password: 12345678
sentinel:
master: mymaster
nodes: 192.168.0.1:26379,192.168.0.1:26479,192.168.0.1:26579
雪崩 击穿 穿透
1、缓存处理流程
接收到查询数据请求时，优先从缓存中查询，若缓存中有数据，则直接返回，若缓存中查不到则从DB中查询，将查询的结果更新到缓存中，并返回查询结果，若DB中查不到，则返回空数据
缓存穿透
当缓存与数据库中都不存在该数据时，由于当数据库查询不到数据就不会写入缓存，这个时候如果用户不断的恶意发起请求，就会导致这个不存在的数据每次请求都会查询DB，请求量大的情况下，就会导致DB压力过大，直接挂掉。
解决方案：
1、当查询返回一个空数据时，直接将这个空数据存到缓存中，过期时间不宜设置过长，建议不超过5分钟
2、采用布隆过滤器：将所有可能存在数据，分别通过多个哈希函数生成多个哈希值，然后将这些哈希值存到一个足够大的bitmap中，此时一个一定不存在的数据就会被这个bitmap拦截，从而减少了数据库的查询压力。
参考链接：https://www.jianshu.com/p/2104d11ee0a2
缓存击穿
某一个数据缓存中没有但数据库中有的数据（一般是缓存时间到期），这时由于并发用户特别多，同时读缓存没读到数据，又同时去数据库去取数据，引起数据库压力瞬间增大，严重情况下会直接挂掉。
解决方案：
1、添加互斥锁:
- ReentrantLock公平锁
- 根据key值加锁，这样线程之间会不影响，不会因为某一个线程获取了锁，其它线程就处于等待时间，也就是线程A从数据库取key1的数据并不妨碍线程B取key2的数据
2、设置热点数据永不过期（物理上的不过期、“逻辑上”的不过期（缓存到期动态构建缓存））
简单的互斥锁例子：
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Autowired
private Jedis jedis;
private final String MUTEX_KEY = "MUTEX_";
public String getData(String key) throws InterruptedException {
String value = stringRedisTemplate.opsForValue().get(key);
//缓存失效
if (StringUtils.isBlank(value)) {
//设置分布式锁，只允许一个线程去查询DB，同时指定过期时间为1min，防止del操作失败，导致死锁，缓存过期无法加载DB数据
if (tryLock(MUTEX_KEY + key, 60L)) {
//从数据库查询数据,将查询的结果缓存起来
value = getValueFromDB();
stringRedisTemplate.opsForValue().set(key, value);
//释放分布式锁
stringRedisTemplate.delete(MUTEX_KEY + key);
} else {
//当锁被占用时，睡眠5s继续调用获取数据请求
Thread.sleep(5000);
getData(key);}
}
return value;
}
/**
* redis实现分布式事务锁 尝试获取锁
*
* @param lockName 锁
* @param expireTime 过期时间
* @return
*/
public Boolean tryLock(String lockName, long expireTime) {
//RedisCallback redis事务管理，将redis操作命令放到事务中处理，保证执行的原子性
String result = stringRedisTemplate.opsForValue().getOperations().execute(new RedisCallback<String>() {
/**
* @param key 使用key来当锁，因为key是唯一的。
* @param value 请求标识，可通过UUID.randomUUID().toString()生成,解锁时通value参数可识别出是哪个请求添加的锁
* @param nx 表示SET IF NOT EXIST，即当key不存在时，我们进行set操作；若key已经存在，则不做任何操作
* @param ex 表示过期时间的单位是秒
* @param time 表示过期时间
*/
@Override
public String doInRedis(RedisConnection connection) throws DataAccessException {
return jedis.set(lockName, UUID.randomUUID().toString(), "NX", "EX", expireTime);
}
});
if ("OK".equals(result)) {
return true;
}
return false;
}
public String getValueFromDB() {
return "";
}
缓存雪崩
缓存中大批量的数据都到了过期时间，从而导致查询数据量巨大，引起数据库压力过大甚至down机。和缓存击穿不同，缓存击穿是指某一条数据到了过期时间，大量的并发请求都来查询这一条数据，缓存雪崩是不同数据都过期了，很多数据都查不到从而查数据库
解决方案：
1、设置热点数据永不过期
2、缓存数据的过期时间设置随机，可以在原有的过期时间上加上一个随机值，比如1-3min，防止同一时间大量缓存数据集体失效，导致数据库压力过大。
3、如果是分布式部署缓存数据库，可将热点数据分别存放到不同的缓存数据库中，避免某一点由于压力过大而down掉。
