一、Nosql概述

1.数据库发展史

1）单机mysql的年代

90年代，一个基本的网站一般不会有太大的访问量，单个数据库就够了
那个时候更多使用的是静态网站。
那么，针对这种情况，整个应用系统的瓶颈是什么？

1. 数据量太大，一个机器装不下（数据库保存数据是持久化操作，是需要保存到硬盘上的）
2. 数据的索引问题，如果一张表有百万条数据，肯定是需要索引的，但是索引太多的话，一个机器也装不下
3. 访问量（读写），一个服务器承受不了
   只要开始出现上述问题，那么就必须要升级。
   记住如果创业开始做网页，从最原始的开始，如果访问量很大的话再逐步升级。

2）Memcached(分布式的高速缓存系统) +mysql + 垂直拆分（读写分离）

memcached是一种缓存技术，它可以把你的数据放入内存，从而通过内存访问提速，因为内存是最快的，缓存技术主要目的就是提速！

在memcached中维护了一张大的hashtable表，该表在内存中，表的结构是key和value
memcached的key一般是字串，不能重复
memcached的value可以放入(字符串，数值，数组，对象，布尔，二进制数据，图片和视频)

网站80%的情况都是在读，每一次都直接查询数据库的话十分麻烦，数据库也承载不了，所以说我们希望减轻数据库的压力，可以使用缓存来保证效率！
该时代的发展过程：先优化数据库结构和索引（对数据库的底层进行的优化）–> 文件缓存（IO，将数据保存在文件中肯定比数据库的性能要好） --> Memcached（当时最热门的技术）

针对单机mysql，为了解决在高并发下读的压力，在中间加入cache缓存，可以很好的解决这个问题。
数据库可以进行垂直拆分，比如，一个读，两个写。写的内容通过主从复制，同步更新到另外两个数据库。

3）分库分表+水平拆分+MySQL集群

技术和业务不断发展，对性能要求越来越高
本质：数据库（读，写）
早些年MyISAM:表锁，十分影响效率！高并发下就会出现严重的锁问题
转战Innodb：行锁
慢慢的就开始使用分库分表来解决写的压力！
mysql的集群满足了那几年的需要

4）最近的年代

mysql等关系型数据库已经不够用了，因为数据量很多，变化很快
Mysql有的使用使用他来存一些比较大的文件，博客，图片！导致数据库表很大，效率就低了，如果有一种数据库来专门处理这种数据，mysql压力就变得十分小。（研究如何处理这些问题）

2.为什么要用nosql？

用户的个人信息，社交网络，地理位置，用户自己产生的数据，用户日志等爆发式增长。
这时候我们需要使用Nosql数据库来解决这些问题

3.什么是nosql？

nosql = not only sql 不仅仅是sql，泛指非关系型数据库，随着web2.0互联网的诞生，传统的关系型数据库很难对付2.0时代，尤其是超大规模的高并发的社区， 暴露出很多问题，所以nosql在当今大数据库下发展的十分迅速。

用户的个人信息，社交网络，地理位置。这些数据类型的存储不需要一个固定的格式!不需要多余的操作就可以横向扩展！

4.nosql有什么特点？

1. 方便扩展（数据之间没有关系，很好扩展）
2. 大数据量高性能（redis一秒可以写8万次，读11万次，nosql的缓存记录级，是一种细粒度的缓存，性能比较高）
3. 数据类型具有多样性（不需要事先设计数据库，随用随取）

5.nosql的四大分类

• KV键值对
• 文档型数据库
• 列存储数据库
• 图形数据库
  

二、redis入门

1.什么是redis？

Redis（Remote Dictionary Server )，即远程字典服务
是一个开源的使用ANSI C语言编写、支持网络、可基于内存亦可持久化的日志型、Key-Value数据库，并提供多种语言的API。

2.redis能干什么？

1. 内存存储、持久化、内存中是断电即失，所以说持久化很重要（rdb,aof）
2. 效率高，可以用于高速缓存
3. 发布订阅系统
4. 地图信息分析
5. 计时器、计数器（浏览量）

3.特性

1.多样的数据类型
2. 持久化
3. 事务

4.学习中需要用到的东西

1、官网：https://redis.io/
2、中文网：http://www.redis.cn/

5.安装

https://blog.csdn.net/weixin_43099842/article/details/124965690?spm=1001.2014.3001.5501

6.性能测试

redis-benchmark是一个压力测试工具！

官方自带的性能测试工具！

redis-benchmark 命令参数，即可测试

首先需要保证redis-server服务开启，并且可以连通

参数如下：

使用下面语句测试
[huizi@localhost bin]$ ./redis-benchmark -h localhost -p 6379 -c 100 -n 100000

7.基础知识

redis默认有16个数据库，从0开始到15个，默认是第0个

可以在配置文件中查看

基本命令

select:切换数据库

dbsize：查看数据库大小

keys *:：获取当前数据多有的key

flushdb：清除当前数据库

flushall：清除所有数据库内容

Redis 是单线程的

Redis是很快的，官方表示，Redis是基于内存操作的，所以CPU并不是Reids的性能瓶颈，Reids的瓶颈是根据机器的内存和网络带宽（请求都需要经过网络）来的。

既然可以使用单线程来实现，所以就使用了单线程！

Redis 是 C 语言写的，官方提供的数据为 100000+ 的QPS(每秒请求数)，完全不比同样是 key-value的Memcache差！

Reids为什么单线程还这么块！

1、误区1：高性能的服务器一定是多线程的？不一定

2、误区2：多线程（CPU会上下文切换，这是耗时的操作！）一定比单线程效率高！是不对的。

需要去了解 CPU、内存、硬盘的速度！

总结：Redis 是将所以的数据全部放在内存中，所以说使用单线程去操作效率就是最高的，因为不需要消耗切换的时间，多线程（CPU上下文会切换：是一个耗时的操作！），对于内存系统来说，如果没有上下文切换效率就是最高的！多次读写都是在一个CPU上，所以在内存情况下，这个就是最佳方案！

三、redis五大数据类型

redis-key命令

127.0.0.1:6379> keys *    # 查看当前数据库所有的key
(empty list or set)
127.0.0.1:6379> set name zxh    # set一个key-value
OK
127.0.0.1:6379> set age 1
OK
127.0.0.1:6379> keys *
1) "age"
2) "name"
127.0.0.1:6379> exists name        # 判断key是否存在
(integer) 1
127.0.0.1:6379> exists name1
(integer) 0
127.0.0.1:6379> move name 1        # 移除key，1表示第0个数据库，从小到大 这个和select有区别
(integer) 1
127.0.0.1:6379> keys *
1) "age"
127.0.0.1:6379> get age        # 根据key获取值
"1"
127.0.0.1:6379> set name zxh
OK
127.0.0.1:6379> expire name 10    # 设置key的过期时间
(integer) 1
127.0.0.1:6379> ttl name    # 查看当前key的剩余时间
(integer) 7
127.0.0.1:6379> ttl name
(integer) 6
127.0.0.1:6379> ttl name
(integer) 5
127.0.0.1:6379> ttl name
(integer) 5
127.0.0.1:6379> ttl name
(integer) 1
127.0.0.1:6379> ttl name
(integer) -2
127.0.0.1:6379> get name
(nil)
127.0.0.1:6379> keys *
1) "age"
127.0.0.1:6379> type age    # 查看当前key的类型

1.String 字符串

字符串常用命令

1.1 追加、判断是否存在、获取长度

set key value：设置值

get key value：获取值

keys *：获取所以的key

append key value：追加字符串，如果当前key不存在，就相当于set一个key

strlen key：获取字符串的长度

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

127.0.0.1:6379> keys *
1) "name"
127.0.0.1:6379> EXISTS name     #判断key是否存在，不存在就是新建
(integer) 1
127.0.0.1:6379> append name hello   # 追加数值
(integer) 10
127.0.0.1:6379> get name
"huizihello"
127.0.0.1:6379> 
127.0.0.1:6379> STRLEN name    # 获取key的value字符串长度
(integer) 10

1.2 字符串类型为整数的增/减量

可能用作浏览量等等

incr key：自增 +1
decr key：自减 -1
incrby key increment：设置步长，指定增量！
decrby key decrement：设置步长，指定减量！

127.0.0.1:6379> set view 0
OK
127.0.0.1:6379> INCR view
(integer) 1
127.0.0.1:6379> get view
"1"
127.0.0.1:6379> INCR view
(integer) 2
127.0.0.1:6379> get view
"2"
127.0.0.1:6379> DECR view
(integer) 1
127.0.0.1:6379> get viewe
(nil)
127.0.0.1:6379> get view
"1"
127.0.0.1:6379> INCRBY view 2
(integer) 3
127.0.0.1:6379> get view
"3"
127.0.0.1:6379> DECRBY view 3
(integer) 0

1.3 截取和替换

getrange key start end：字符串截取，比如 [0,5] 左右都是一个闭区间

setrange key offset value：字符串的替换，替换指定下标的字符

127.0.0.1:6379> getrange name 0 3   # 左右都是闭区间
"huiz"
127.0.0.1:6379> setrange name 4 i   
(integer) 10
127.0.0.1:6379> get name
"huizihello"

1.4 设置时间

setex key seconds value：设置过期时间 setex(set with expire)

ttl key：查看剩余时间

setnx key value：判断是否存在，再创建 setnx(set with no exist)

127.0.0.1:6379> setnx age 19    # 如果key存在就不创建
(integer) 0
127.0.0.1:6379> get age
"18"
127.0.0.1:6379> setnx sex 1   # 如果key不存在就创建
(integer) 1
127.0.0.1:6379> keys *
1) "age"
2) "sex"
3) "name"
4) "view"
127.0.0.1:6379> 

1.5 多个值的set 和获取get以及对象存储

mset [key value…]：同时设置多个k-v

mget [key …]：同时获取多个k-v

msetnx [key value…]：设置多个k-v，msetnx是一个原子操作，要么一起成功，要么一起失败，用于分布式锁

127.0.0.1:6379> mset k1 v1 k2 v2 k3 v3 # 同时设置多个k-v
OK
127.0.0.1:6379> keys *
1) "k1"
2) "k2"
3) "age"
4) "name"
5) "sex"
6) "k3"
7) "view"
127.0.0.1:6379> mget k1 k2 k3  # 同时获取多个key
1) "v1"
2) "v2"
3) "v3"
127.0.0.1:6379> msetnx k1 v1 k4 v4  # 设置多个key，要么成功，要么失败
(integer) 0

可以使用 user:1:name 这种key方式来表示 user1对象的name属性，用来代替json字符串

# 如果保存 user1：{name:zhangsan,age:18}  json表示的对象 那么可以下面这么做。足以看出redis是非常强大的
127.0.0.1:6379> mset user:1:name zhansgan user:1:age 18
OK
127.0.0.1:6379> keys *
1) "age"
2) "k2"
3) "user:1:name"
4) "name"
5) "view"
6) "user:1:age"
7) "k1"
8) "sex"
9) "k3"
127.0.0.1:6379> 

1.6 先获取后设置

getset key value：先获取之前存放的值，再设置，如果存在就是覆盖

127.0.0.1:6379> getset age 18 
"18"
127.0.0.1:6379> get age
"18"
127.0.0.1:6379> get age 17
(error) ERR wrong number of arguments for 'get' command
127.0.0.1:6379> getset age 17
"18"
127.0.0.1:6379> get age
"17"

应用场景

String类似的使用场景：value除了是我们的字符串还可以是我们的数字！

• 计数器
• 统计多单位的数量
• 粉丝数
• 对象缓存存储

2.List

在redlis中的List类型，可以玩成栈、队列、阻塞队列
看下面的图
左端进，右端出 ---->队列
左端进，左端出 ---->栈

所有的List命令基本都是l开头的

##################################################

lpush key value [value…]：将一个或多个值，插入到列表的头部（也就是左端）

lrange key start end：取list中的值！先进后出，后进先出，可以看出一条水平线，从左端依次塞入，做左端依次拿出

rpush key value [value…]：从将一个或多个值，从列表的右端放入

##################################################

lpop key：移除列表的第一元素，从左端移除

rpop key：移除列表的最后一个元素，从右端移除

##################################################

lindex key index：通过下标获取列表的值

llen key：获取列表的长度/大小

##################################################

lrem key count value：移除列表中指定个数的指定的值，精准匹配

##################################################

ltrim key start end：通过下标指定长度，截取列表的值，并且用截取的列表改变原列表

##################################################

rpoplpush resouce destination：移除原列表最后一个元素，将它移动到新的列表中（做左端放入）

##################################################

lset key index value：替换列表中指定下标的值

linsert key before|after pivot value：在指定的值前面或者后面插入值

127.0.0.1:6379> lpush list one 
(integer) 1
127.0.0.1:6379> LPUSH list two
(integer) 2
127.0.0.1:6379> LPUSH list three
(integer) 3
127.0.0.1:6379> LRANGE list 0 -1
1) "three"
2) "two"
3) "one"
127.0.0.1:6379> RPUSH list four
(integer) 4
127.0.0.1:6379> LRANGE list 0 -1
1) "three"
2) "two"
3) "one"
4) "four"
###########################################################################################################
127.0.0.1:6379> lpop list 
"three"
127.0.0.1:6379> rpop list
"four"
127.0.0.1:6379> LRANGE list 0 -1
1) "two"
2) "one"
###########################################################################################################
127.0.0.1:6379> lindex list 0
"two"
127.0.0.1:6379> llen list
(integer) 2
127.0.0.1:6379> lrange list 0 -1
1) "one"
2) "one"
3) "one"
4) "one"
###########################################################################################################
127.0.0.1:6379> lrem list 3 one
(integer) 3
127.0.0.1:6379> lrange list 0 -1
1) "one"
###########################################################################################################
127.0.0.1:6379> ltrim list 0 1
OK
127.0.0.1:6379> lrange list 0 -1
1) "one"
2) "one"
###########################################################################################################
127.0.0.1:6379> RPOPLPUSH list mylist
"one"
127.0.0.1:6379> keys *
1) "mylist"
2) "list"
127.0.0.1:6379> EXISTS mylist
(integer) 1
127.0.0.1:6379> lrange list 0 -1
1) "one"
###########################################################################################################
127.0.0.1:6379> lset mylist 0 two
OK
127.0.0.1:6379> LRANGE mylist 0 -1
1) "two"
###########################################################################################################
127.0.0.1:6379> LINSERT list before one hello
(integer) 2
127.0.0.1:6379> lrange list 0 -1
1) "hello"
2) "one"
127.0.0.1:6379> LINSERT list after one hello1
(integer) 3
127.0.0.1:6379> lrange list 0 -1
1) "hello"
2) "one"
3) "hello1"

小结：

他实际上是一个链表，before Node after ， left，right 都可以插入值

如果key 不存在，创建新的链表

如果key存在，新增内容

如果移除了所有值，空链表，也代表不存在！

在两边插入或者改动值，效率最高！ 中间元素，相对来说效率会低一点~

消息排队！消息队列 （Lpush Rpop）， 栈（ Lpush Lpop）！

3.Set

set中的值是不能重复的！set命令一般是以s开头

sadd key meber [member…]：在set集合中添加元素

smembers key：查看指定set集合的所有值

sismember key member：判断某一个值是否在集合中

scard key：获取指定set集合的大小

srem key member [member…]：移除set集合中指定的一个或多个元素

srandmember key count：随机获取set集合中的一个或者多个元素

spop key cout：随机删除一个或者多个元素

smove resouce destination member：将集合中指定的元素移动到另一个集合中

sdiff key [key…]：差集，保留在第一个集合中其他集合里没有的值

sinter key [key…]：交集

sunion key [key…]：并集

# sadd key meber [member...]：在set集合中添加元素

# smembers key：查看指定set集合的所有值
127.0.0.1:6379> sadd myset hello
(integer) 1
127.0.0.1:6379> sadd myset "hello"
(integer) 0
127.0.0.1:6379> SMEMBERS myset
1) "hello"

#sismember key member：判断某一个值是否在集合中
127.0.0.1:6379> SISMEMBER myset hello
(integer) 1
127.0.0.1:6379> SISMEMBER myset hello
(integer) 1
127.0.0.1:6379> SISMEMBER myset hello1
(integer) 0

#scard key：获取指定set集合的大小
127.0.0.1:6379> scard myset
(integer) 1

#srem key member [member...]：移除set集合中指定的一个或多个元素
127.0.0.1:6379> srem myset hello
(integer) 1
127.0.0.1:6379> smembers myset
(empty array)

#srandmember key count：随机获取set集合中的一个或者多个元素
127.0.0.1:6379> sadd myset huizi huizi1 huzi2 huizi3 huizi4
(integer) 5
127.0.0.1:6379> srandmember myset 1
1) "huzi2"
127.0.0.1:6379> srandmember myset 1
1) "huzi2"
127.0.0.1:6379> srandmember myset 1
1) "huizi1"

#spop key cout：随机删除一个或者多个元素
127.0.0.1:6379> spop myset 1
1) "huizi3"
127.0.0.1:6379> smembers myset
1) "huizi"
2) "huzi2"
3) "huizi1"
4) "huizi4"

#smove resouce destination member：将集合中指定的元素移动到另一个集合中
127.0.0.1:6379> smove myset myset2 huizi
(integer) 1
127.0.0.1:6379> smembers myset2
1) "huizi"

#sdiff key [key...]：差集，保留在第一个集合中其他集合里没有的值
# myset{huizi、huizi1、huizi2、huzi4}和myset2 {huizi}   差集为{huizi1、huizi2、huzi4}
127.0.0.1:6379> sdiff myset myset2
1) "huizi1"
2) "huzi2"
3) "huizi4"

#sinter key [key...]：交集
127.0.0.1:6379> sinter myset myset2
1) "huizi"

#sunion key [key...]：并集
127.0.0.1:6379> sadd myset2 huizi6
(integer) 1
127.0.0.1:6379> smembers myset2
1) "huizi6"
2) "huizi"
127.0.0.1:6379> sunion myset myset2
1) "huzi2"
2) "huizi4"
3) "huizi"
4) "huizi6"
5) "huizi1"

小结:
可以应用在共同关注，共同爱好，随机抽奖应用场景中。

4.Hash

hash类型：相当于map集合，原来存储的是 key-value，现在变成 key-map！本质和String类型没有太大区别，还是一个简单的k-v！

hset key field value：存放一个具体的 k-v

hget key field：获取key中某一个属性值

hmset key field value [field value…]：存放一个或多个 k-v

hmget key field [field…]：获取key中一个或多个属性值

hdel key field [field…]：删除key中一个或多个属性值

hgetall key：获取key的全部的数据，以键值对形式显示

hlen key：获取map的长度

hexists key field：判断map是否有对应的属性

hkeys key：获取map所有的key

hvals key：获取map所有的value

hincrby key field increment：对map中的整数类型进行增量

hsetnx key field value：判断map中是否有对应的属性，如果没有进行存储

# hset key field value：存放一个具体的 k-v

# hget key field：获取key中某一个属性值

# hmset key field value [field value...]：存放一个或多个 k-v

# hmget key field [field....]：获取key中一个或多个属性值
127.0.0.1:6379> hset myhash field1 hello
(integer) 1
127.0.0.1:6379> hget myhash field
(nil)
127.0.0.1:6379> hget myhash field1
"hello"
127.0.0.1:6379> hmset myhash field2 hello2 field3 hello3
OK
127.0.0.1:6379> hmget myhash field field2
1) (nil)
2) "hello2"

# hdel key field [field...]：删除key中一个或多个属性值

# hgetall key：获取key的全部的数据，以键值对形式显示
127.0.0.1:6379> hdel myhash field1
(integer) 1
127.0.0.1:6379> hgetall myhash
1) "field2"
2) "hello2"
3) "field3"
4) "hello3"

# hlen key：获取map的长度

# hexists key field：判断map是否有对应的属性
127.0.0.1:6379> hlen myhash
(integer) 2
127.0.0.1:6379

# hkeys key：获取map所有的key

# hvals key：获取map所有的value
127.0.0.1:6379> hkeys myhash
1) "field2"
2) "field3"
127.0.0.1:6379> hvals myhash
1) "hello2"
2) "hello3"

# hincrby key field increment：对map中的整数类型进行增量
127.0.0.1:6379> hset myhash field5 5
(integer) 1
127.0.0.1:6379> hincrby myhash field5 1
(integer) 6

# hsetnx key field value：判断map中是否有对应的属性，如果没有进行存储
127.0.0.1:6379> hsetnx myhash field5 1
(integer) 0  # 表示已经存在
127.0.0.1:6379> hsetnx myhash field4 2
(integer) 1  # 表示不存在，然后进行存储

小结：
hash主要用于存储用户信息之类的，经常变动的信息！hash更适合对象的存储，String类型更适合字符串存储！

127.0.0.1:6379> hset user:1 name huizi
(integer) 1
127.0.0.1:6379> hget user:1 name
"huizi"

5.Zset（有序集合）

Zset就是在set集合上，增加了一个值，比如：set k1 v1，现在是zset k1 score1 v1

我们就可以针对score1 这个值进行一些操作：比如排序

zadd key score member [score member…]：添加一个或者多个元素

zrange key start end [withscores]：指定区间获取set集合的元素，可以设置是否显示score字段

zrangebyscore key min max [withscores]：根据指定的score进行升序排列

zrevrange key start stop [withscores]：根据指定的score进行降序排列

zrem key member [member…]：删除一个或者多个元素

zcard key：获取set集合的长度/大小

zcount key min max：统计个数，按照score属性的区间进行，在[min，max]之间的个数

# zadd key score member [score member...]：添加一个或者多个元素

# zrange key start end [withscores]：指定区间获取set集合的元素，可以设置是否显示score字段
127.0.0.1:6379> zadd myzset 1 one
(integer) 1
127.0.0.1:6379> zadd myzset 2 two 3 three
(integer) 2
127.0.0.1:6379> zrange myzset 0 -1
1) "one"
2) "two"
3) "three"
# zrangebyscore key min max [withscores]：根据指定的score进行升序排列

# zrevrange key start stop [withscores]：根据指定的score进行降序排列
127.0.0.1:6379> zadd salary 2500 xiaohong
(integer) 1
127.0.0.1:6379> zadd salary 3000 xiaowang
(integer) 1
127.0.0.1:6379> zadd salary 3100 xiaozhang
(integer) 1
127.0.0.1:6379> zrangebyscore salary -inf +inf
1) "xiaohong"
2) "xiaowang"
3) "xiaozhang"
127.0.0.1:6379> zrangebyscore salary -inf +inf withscores   # 带上 withscores 就带上scores了
1) "xiaohong"
2) "2500"
3) "xiaowang"
4) "3000"
5) "xiaozhang"
6) "3100"
127.0.0.1:6379> zrevrange salary 0 -1
1) "xiaozhang"
2) "xiaowang"
3) "xiaohong"

# zrem key member [member...]：删除一个或者多个元素
127.0.0.1:6379> zrem salary xiaowang
(integer) 1
127.0.0.1:6379> zrange salary 0 -1
1) "xiaohong"
2) "xiaozhang"

# zcard key：获取set集合的长度/大小
127.0.0.1:6379> zcard salary
(integer) 2

# zcount key min max：统计个数，按照score属性的区间进行，在[min，max]之间的个数
127.0.0.1:6379> zcount salary -inf +inf
(integer) 2

其余的api，工作中有用到就去官网查看

小结：

既然时有序的，可以对班级的成绩，工资表进行排序！

比如b站的排行榜，统计视频的播放量，放到Zset进行排序，每天进行操作即可！

比如普通消息是1，重要的消息是2，可以进行带权重进行判断！

四、三种特殊数据类型

1.Geospatal 地理位置

Reids的 Geo 在 Redis3.2 版本就推出了！这个功能可以推算地理位置的信息，两地之间的距离，方圆几里的人！

可以查询一些测试数据（城市的纬度和经度）：http://www.jsons.cn/lngcodeinfo/0706D99C19A781A3/

官方文档：https://www.redis.net.cn/order/3685.htm

只有六个命令

1.1 geoadd 添加地理位置

geoadd key longitude latitude member [longitude latitude member…]：添加城市的经纬度信息

longitude：纬度

latidude：经度

member：成员/元素

1.2 geopos 获取地理位置定位

geopos key member [member…]：查询一个或多个城市的经纬度

1.3 geodist 计算两个人之间的距离

geodist key member1 member2 [unit]：计算两个城市/人h之间的举例

unit表示单位

单位：
m 表示单位为米。
km 表示单位为千米。
mi 表示单位为英里。
ft 表示单位为英尺。

1.4 georadius 以经纬度为中心查找半径以内的元素

georadius：以经纬度为中心查找半径以内的元素
指定半径：可以做到比如：找附近的人，通过半径来搜寻；
设置查询数量：如果找到了几万人，那么显示不下，肯定会只显示几个。

格式：georadius key longitude latitude radius m|km|ft|mi [withcoord] [withdist] [withhash]

1.5 georadiusbymember：以元素为中心查找半径以内的元素

georadiusbymember china:city beijing 1000 km

1.6 geohash：以字符串表示返回的经纬度

将二维的经纬度转换为一维的字符串，如果两个字符串越接近，那么则距离越近！

geo底层的实现原理其实就是Zset！我们可以使用Zset命令来操作geo！

2.Hyperloglog 计数

Redis 2.8.9 版本就更新了 Hyperloglog 数据结构！

Redis Hyperloglog 基数统计的算法！

什么是基数？

两个集合，保留不重复的元素，

A {1,3,5,7,8,7}

B{1,3,5,7,8}

基数（保留不重复的元素） = 5 = {1,3,5,7,8}，可以接受误差！

网页的 UV （一个人访问一个网站多次，但是还是算作一个人！）
传统的方式，set保存用户的id，因为set集合是不重复的，相同的会覆盖，然后就可以统计set中的元素数量来做标准判断！

这个方式如果保存大量的用户id，就会比较麻烦、占用内存！我们的目的就是为了计数，而不是保存用户id。

Hyperloglog它的优点：占用的内存是固定，2^64 不同的元素的基数，只需要花费 12KB内存！如果要从内存角度比较的话 Hyperloglog 首选！

官方说会有 0.81% 错误率！ 统计UV任务，可以忽略不计的！

pfadd key element [element…] 创建元素

pfcount key 查看元素数量

pfmerge destkey sourcekey [ourcekey…] 看并集数量

127.0.0.1:6379> pfadd keys a a a b v v c    # 创建第一组带有重复数据的元素
(integer) 1
127.0.0.1:6379> pfcount keys
(integer) 4
127.0.0.1:6379> pfadd mykey a b c e f g h i j # 创建第一组元素 mykey
(integer) 1
127.0.0.1:6379> pfcount mykey    # 统计 mykey 元素的基数数量
(integer) 9
127.0.0.1:6379> pfadd mykey2 i j z x c v b n m
(integer) 1
127.0.0.1:6379> pfcount mykey2
(integer) 9
127.0.0.1:6379> pfmerge mykey3 mykey mykey2    # 合并两组 mykey mykey2 => mykey3 并集
OK
127.0.0.1:6379> pfcount mykey3    # 看并集的数量
(integer) 14

如果允许容错，那么一定可以使用 Hyperloglog
如果不允许容错，就使用 set 或者自己的 数据类型即可

3.Bitmap 位存储

比如：中国疫情，要统计全国的疫情感染人数，我们只需用 14亿个 0去存储，然后感染的一个就将一个 0 改为 1

14亿个0很大吗？不大只需要1000MB就可以存放83亿个bit位，一个字节 = 8 bit，每个bit位可以存放一个0

所以，Bitmap位图，根据位存储的数据结构算法，可以发现只有 0 和 1两个状态

那么就可以应用：

网站用户的活跃、不活跃

员工 365 天的打卡情况

使用Bitmap 来记录 周一到周日的打卡！

• 使用0和1表示打卡和未打卡：周一：1 周二：0 周三：0 周四：1 …

127.0.0.1:6379> setbit sign 1 1
(integer) 0
127.0.0.1:6379> setbit sign 2 0
(integer) 0
127.0.0.1:6379> setbit sign 3 0
(integer) 0
127.0.0.1:6379> setbit sign 4 1
(integer) 0
127.0.0.1:6379> setbit sign 5 1
(integer) 0
127.0.0.1:6379> setbit sign 6 1
(integer) 0
127.0.0.1:6379> setbit sign 7 0   
(integer) 0
127.0.0.1:6379> getbit sign 5    # 查看某一天是否打卡
(integer) 1
127.0.0.1:6379> bitcount sign   # 统计值为0的数量
(integer) 4

五、事务

Reids事务保证：一次性、顺序性、排他性！执行一系列的命令！

• 一次性：只有当所有的命令进入队列后，再一次性执行。
• 顺序性：按照进入队列的先后顺序执行，先进先出。
• 排他性：事务在执行过程中，是不允许被干扰的。

注意点：

• Redis事务没有隔离级别的概念！
• 所有的命令在事务中，并没有直接执行！只有发起执行的命令的时候才会执行！
• Redis 单条命令是保证原子性的，但是事务不保证原子性！

Redis的事务，需要做3步：

• 开启事务（multi）
• 命令入队（…）
• 开启事务（exec）

正常事务

127.0.0.1:6379> multi
OK
127.0.0.1:6379(TX)> set name1 huizi
QUEUED
127.0.0.1:6379(TX)> set name2 huizi2
QUEUED
127.0.0.1:6379(TX)> set name3 huizi3
QUEUED
127.0.0.1:6379(TX)> exec
1) OK
2) OK
3) OK
127.0.0.1:6379> keys *
 1) "name2"
 2) "name1"
 3) "mylist"
 4) "myhash"
 5) "list"
 6) "salary"
 7) "myset2"
 8) "sign"
 9) "user:1"
10) "myzset"
11) "name3"
12) "myset"

放弃事务

127.0.0.1:6379> multi
OK
127.0.0.1:6379(TX)> set name1 huizi
QUEUED
127.0.0.1:6379(TX)> set name2 huizi2
QUEUED
127.0.0.1:6379(TX)> set name3 huizi3
QUEUED
127.0.0.1:6379(TX)> dicard

运行时异常

如果队列中存在错误，其他命令是可以正常执行的，错误命令会抛出异常！

编译型异常（代码有问题，命令有错的时候）

127.0.0.1:6379(TX)> set name
(error) ERR wrong number of arguments for 'set' command
127.0.0.1:6379(TX)> set name chengwenying
QUEUED
127.0.0.1:6379(TX)> set name1 huizi
QUEUED
127.0.0.1:6379(TX)> set name2 xiaozhang
QUEUED
127.0.0.1:6379(TX)> set name3 xiaowang
QUEUED
127.0.0.1:6379(TX)> exec
(error) EXECABORT Transaction discarded because of previous errors.

redis监视

测试多线程修改值，使用watch 可以当做redis乐观锁操作！

1. 一个线程正在执行事务

   127.0.0.1:6379> set money 100
   OK
   127.0.0.1:6379> decrby money 20
   (integer) 80
   127.0.0.1:6379> incrby money 20
   (integer) 100
   127.0.0.1:6379> watch money 
   OK
   127.0.0.1:6379> multi
   OK
   127.0.0.1:6379(TX)> decrby money 20
   QUEUED
   

2. 另一个线程插队，将money的值修改为120

   127.0.0.1:6379> get monry
   (nil)
   127.0.0.1:6379> get money
   "100"
   127.0.0.1:6379> set money 120
   OK
   
   

3. 再执行之前的事务

   127.0.0.1:6379> set money 100
   OK
   127.0.0.1:6379> decrby money 20
   (integer) 80
   127.0.0.1:6379> incrby money 20
   (integer) 100
   127.0.0.1:6379> watch money 
   OK
   127.0.0.1:6379> multi
   OK
   127.0.0.1:6379(TX)> decrby money 20
   QUEUED
   127.0.0.1:6379(TX)> exec
   (nil)
   127.0.0
   

如果修改失败，那么就获取最新的值

127.0.0.1:6379> unwatch
OK
127.0.0.1:6379> watch money
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379(TX)> decrby money 20
QUEUED
127.0.0.1:6379(TX)> incrby money 20
QUEUED
127.0.0.1:6379(TX)> exec
1) (integer) 100
2) (integer) 120

六、Jedis连接工具和SpringBoot整合Redis

1.Jedis连接工具

它是官方推荐的java连接开发工具，使用java操作redis中间件

1. 新建maven项目，导入依赖

    <dependencies>
        <!-- https://mvnrepository.com/artifact/redis.clients/jedis -->
        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>4.2.3</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>2.0.3.graal</version>
        </dependency>
    </dependencies>

2. 测试连接

public class test {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("127.0.0.1", 6379);
        System.out.println(jedis.ping());
    }
}

常用API
和Linux命令一样，也就是说linux命令对应着java方法

public class test {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("127.0.0.1", 6379);
        jedis.flushDB();//清空当前数据库
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("username","zxh");
        jsonObject.put("age",18);
        String s = jsonObject.toString();

        //开启事务

        Transaction multi = jedis.multi();
        try {
            multi.set("user1",s);
            multi.set("user2",s);
            multi.exec();
        } catch (Exception e) {

            multi.discard(); // 抛出异常，取消事务
            e.printStackTrace();
        } finally {
            System.out.println(jedis.get("user1"));
            System.out.println(jedis.get("user2"));
            jedis.close();  //关闭连接
        }
    }
}

2.SpringBoot整合Redis

1. 创建Springboot项目，导入redis依赖

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

进入redis依赖可以找到，底层使用的就是redis，但是可以看到另外一个lettuce

说明

Jedis：采用的是直连，所以在多个线程操作的话，是不安全的！如果想要避免不安全，使用 Jedis pool 连接池！ 更像 BIO（Blocking I/O，阻塞IO）

lettuce：采用netty，实例可以在多个线程中进行共享，不存在线程不安全的情况！可以减少线程数据量，更像 NIO（Nonblocking I/O，非阻塞IO）
springboot中默认使用lettuce，因为其支持高并发

NIO和BIO解释

Netty是一款基于NIO（Nonblocking I/O，非阻塞IO）开发的网络通信框架，对比于BIO（Blocking I/O，阻塞IO），他的并发性能得到了很大提高

两张图了解BIO和NIO的区别：
堵塞IO

非堵塞IO

这两张图可以看出，NIO的单线程能处理连接的数量比BIO要高出很多，而为什么单线程能处理更多的连接呢？原因就是图二中出现的Selector

当一个连接建立之后，它有两个步骤要做：

  第一步是接收客户端发过来的全部数据

  第二步是服务端处理完请求之后返回response客户端

NIO和BIO的区别主要是在第一步：

在BIO中，等待客户端发数据这个过程是阻塞的，这样就造成了一个线程只能处理一个请求的情况，而机器能支持的最大线程数是有限的，这就是为什么BIO不能支持高并发的原因。

而NIO中，当一个Socket 建立之后，Thread 并不会阻塞接收这个 Socket，而是将这个请求交给Selector，Selector 会不断的去遍历所有的 Socket，一旦有一个Socket 建立完成，他会通知Thread，然后Thread 处理完数据后再返回给客户端——这个过程是不阻塞的，这样就能让一个 Thread 处理更多的请求了。

RedisAutoConfiguration自动配置源码分析

@AutoConfiguration
@ConditionalOnClass({RedisOperations.class})
@EnableConfigurationProperties({RedisProperties.class})  //自动加载配置类
@Import({LettuceConnectionConfiguration.class, JedisConnectionConfiguration.class})
public class RedisAutoConfiguration {
    public RedisAutoConfiguration() {
    }

    @Bean
    @ConditionalOnMissingBean(
        name = {"redisTemplate"}
    )
    //可以自己定义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)
    //redis比较常用，需要的话直接使用即可
    public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory redisConnectionFactory) {
        return new StringRedisTemplate(redisConnectionFactory);
    }
}

测试

1. 配置文件

2. 测试连接
   在测试类中进行测试，可以直接注入RedisTemplate

   @SpringBootTest
   class RedisApplicationTests {
   
       @Autowired
       private RedisTemplate redisTemplate;
       @Test
       void contextLoads() {
           redisTemplate.opsForValue().set("mykey","huizi");
           System.out.println(redisTemplate.opsForValue().get("mykey"));
       }
   
   }
   

自定义redisTemplate

为什么要自定义？
首先RedisTemplate 它没有过多的配置
默认的RedisTemplate 使用的是jdk序列化，会有一些问题

@Configuration  // 配置类
public class RedisConfig {

    // 自定义RedisTemplate
    @Bean
    @SuppressWarnings("all")
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory){
        // 为了开发的方便，一般直接使用 <String, Object>
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(factory); // redis连接的线程安全工厂，源码也是这样配置的
        // Json序列化配置
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        //解决jackson2无法序列化LocalDateTime的问题，这里扩展一个LocalDateTime类型，它是日期类型对象 jdk1.8出的（并且这个类是不可变的和线程安全的，可以研究一下它的API），当然还需要对这个对象进行json格式的转换，如下图：
        om.disable(SerializationFeature.WRITE_DATE_KEYS_AS_TIMESTAMPS);
        om.registerModule(new JavaTimeModule());
        
        jackson2JsonRedisSerializer.setObjectMapper(om);
        // String 的序列化
         StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        // key采用String的序列化方式
         template.setKeySerializer(stringRedisSerializer);
        // hash的key也采用String的序列化方式
         template.setHashKeySerializer(stringRedisSerializer);
        // value序列化方式采用jackson
         template.setValueSerializer(jackson2JsonRedisSerializer);
        // hash的value序列化方式采用jackson
         template.setHashValueSerializer(jackson2JsonRedisSerializer);
         template.afterPropertiesSet();

        return template;
    }

}

我们使用上面的redisTemplates测试一下

@SpringBootTest
@SuppressWarnings("all")
class RedisApplicationTests {

    @Autowired
    //@Qualifier("redisTemplate")
    private RedisTemplate redisTemplate;
    @Test
    void contextLoads() {
        User user = new User("辉子", 18);
        //redisTemplate.opsForValue().set("辉子","huizi");
        redisTemplate.opsForValue().set("user",user);
        System.out.println(redisTemplate.opsForValue().get("user"));
    }
}

结果

就是转义后的字符了

在我们真实的开发中，或者你们在公司，一般都可以看到一个公司自己封装RedisUtil

七、持久化

1.RDB

1.1 什么是RDB？

Redis是内存数据库，如果不将内存中的数据库保存到磁盘，那么一旦服务器进程退出，服务器中的数据库状态也会消失，所以 Redis 提供了持久化的功能（RDB 和 AOF）！

在指定的时间间隔内将内存中的数据集快照写入磁盘，也就是行话讲的Snapshot快照，它恢复时是将快照文件直接读到内存中。

工作方式：

Redis 会单独创建（fork）一个子进程来进行持久化操作，会先将数据写入到一个临时RDB之中，待持久化过程都结束了，再用这个临时文件替换上次持久化好的文件。整个过程中，主进程是不进行任何IO的操作。这就确保了极高的性能。

如果需要进行大规模的数据恢复，但对于数据恢复的完整性不是非常敏感，那RDB方式要比AOF方式更加的高效。

RDB的缺点就是最后一次持久化后的数据可能丢失。

RDB的触发机制
可在redis.conf中进行配置

可以自定义save规则来触发rdb

60s内修改50次key就会触发rdb

三种触发机制
1、save的规则满足的时候，会自动触发rdb进行持久化

2、执行flushall命令，也会触发我们的rdb持久化

3、退出redis进程后，也会产生rdb文件

生成的dump.rdb默认会存在redis的bin目录下

如果该目录下存在 dump.rdb文件，redis启动就会自动恢复其中的数据。
优缺点

优点

1、适合大规模的数据恢复！

2、对数据的完整性要求不高！

缺点

1、需要一定的时间间隔才会进行持久化操作！如果redis在最后一次意外宕机，那么最后一次修改的数据也就没有了！

2、fork子进程的时候，会占用一定的内存空间！

2.AOF

aof(append only file) 将我们所有的命令都记录下来，相当于历史history，恢复的时候把这个文件全部在执行一遍！

aof的配置

默认是不开启的，我们需要手动进行配置！只需要将上面的no 改成yes 就开启了aof！

触发条件：
配置完后，重启redis服务就会在启动目录下生成appendonly.aof文件
如果文件出错，那么如何修复aof文件呢？
我们可以通过redis-check-aof命令修复，该命令在启动目录下

redis-check-aof --fix appendonly.aof # 修复文件

优缺点

优点

1、每一次修改都可以同步数据，保证了文件的完整性！

2、每一秒修改都会同步一次，但是最后一秒的数据可能会丢失！

3、可以设置从不同步数据，那么效率就会很高！

缺点

1、相对于数据文件来说，aof 文件的容量 远远大于 rdb文件

2、修复速度也比 rdb慢！

3、aof 运行效率也要比 rdb慢

所以redis的默认配置是rdb持久化！

扩展
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分钟备份一次就够了，只保留 save 900 1 这条规则。

• 如果Enable AOF，好处是在最恶劣情况下也只会丢失不超过两秒数据，启动脚本较简单只load自己的AOF文件就可以了，代价一是带来了持续的IO，二是AOF
  rewrite 的最后将 rewrite 过程中产生的新数据写到新文件造成的阻塞几乎是不可避免的。只要硬盘许可，应该尽量减少AOFrewrite的频率，AOF重写的基础大小默认值64M太小了，可以设到5G以上，默认超过原大小100%大小重写可以改到适当的数值。

• 如果不Enable AOF ，仅靠 Master-Slave Repllcation实现高可用性也可以，能省掉一大笔IO，也减少了rewrite时带来的系统波动。代价是如果Master/Slave同时倒掉，会丢失十几分钟的数据，启动脚本也要比较两个 Master/Slave 中的 RDB文件，载入较新的那个，微博就是这种架构。

八、订阅发布

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

应用的领域：微信公众号、微博的关注、还有一些关注系统等！

Redis客户端可以订阅任意数量的频道。

订阅/发布消息图：

总共有三个对象：消息的发送者、频道、消息的订阅者！

下图展示了频道 channel1 ， 以及订阅这个频道的三个客户端 —— client2 、 client5 和 client1 之间的关系

当有消息通过PUBLISH命令发送给频道channel时，这个消息就被发送给订阅它的三个客户端

命令：

订阅端发送消息，发送端就会实时收到消息。

原理：
Redis 是使用 C 实现的，通过分析 Redis 源码里的 pubsub.c 文件，可以了解发布和订阅机制的底层实现，以此加深对 Redis的理解。

Redis通过 PUBLISH、SUBSCRIBE 和 PSUBSCRIBE 等命令实现发布和订阅功能。

以微信的公众号作为举例：

通过SUBSCRIBE 命令订阅某频道后，redis-server 里会维护一个字典，字典的键就是一个个频道！而字典的值则是一个链表，链表中保存了所有订阅这个 channel 频道的客户端。SUBSCRIBE 命令的关键，就是将客户端添加到给定 channel 频道的订阅链表中。

通过 PUBLISH 命令向订阅者发送消息，redis-server 会使用给定的频道作为键，在他锁维护的 channel 字典中查找记录了订阅这个频道的所有客户端的链表，遍历这个链表，将消息发布给所有订阅者。

Pub/Sub 从字面上理解就是发布（Publish）与订阅（Subscribe）

在Redis中，你可以设定对某一个key值进行消息发布及消息订阅，当一个key值上进行了消息发布后，所有订阅它的客户端都会收到相应的消息。

这一功能最明显的用法就是用作实时消息系统，比如普通的即时聊天，群聊等功能。

使用场景

1、实时消息系统！

2、实时聊天！（频道当做聊天室，将信息回显给所有人即可！）

3、订阅，关注系统都是可以的！

稍微复杂的场景我们就会使用消息中间件MQ

九、主从复制

主从复制，就是指一台Redis服务器的数据，复制到其他的Redis服务器。目前称为主节点（master/leader），后者称为从节点（slave/follower）

需要注意：

1、数据的复制是单向的，只能从主节点到从节点，Master以写为主，slave以读为主

2、默认情况下，每台Redis 服务器都是主节点。且一个主节点可以有多个从节点，但是一个从节点只有一个主节点。

主从复制的作用主要包括：

1、数据冗余：主从复制实现了数据的热备份，是持久化以外的一种数据冗余方式。

2、故障恢复：当主节点出现问题时，可以由从节点提供服务，实现快速故障恢复；实际上是一种服务的冗余。

3、负载均衡：在主从复制的基础上，配合读写分离，可以由主节点提供写服务，由从节点提供读服务（及写Redis数据时应用连接主节点，读Redis数据时应用连接从节点），以此来分担服务器负载；尤其是在少写读多的情况下，通过多个从节点分担读负载，可以大大提高Redis服务器的并发量。

4、高可用（集群）基石：除了上述作用以外，主从复制还是哨兵和集群能够实施的基础，因此说主从复制是Redis高可用的基础。

一般来说，要将Redis运行于工程项目中，只是用一台Redis是万万不能的，因为会宕机，原因如下：

1、从结构上，单个Redis服务器会发生单点故障，并且一台服务器需要处理所有的请求负载，压力较大；

2、从容量上，单个Redis服务器内存容量有限，就算一台Redis服务器内存容量为256G，也不能将所有的内存用作Redis存储内存，一般来说，单台Redis最大使用内存不应该超过20G。

比如电商网站上的商品，一般都是一次上传，无数次浏览的，说专业点也就是"多读少写"

主从复制，读写分离！ 80% 的情况下都是在进行读操作！减缓服务器的压力！架构中经常使用！ 最低配置：一主二从！

只要在公司中，主从复制就是必须要使用的，因为在真实的项目中不可能单机使用Redis！

环境搭建

1. 可以通过info replication来查看服务信息
   只配置从库，不需要配置主库

127.0.0.1:6379> info replication
# Replication
role:master   # 主机
connected_slaves:0   # 没有从机
master_failover_state:no-failover
master_replid:e1f0c835244a7ac008f5add83258143e750ff042
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:0
second_repl_offset:-1
repl_backlog_active:0
repl_backlog_size:1048576
repl_backlog_first_byte_offset:0
repl_backlog_histlen:0
127.0.0.1:6379> 

2. 复制三个配置文件，用于模拟3个Redis集群。
   
3. 分别修改5处配置文件的信息：为了不重复
   端口
   后台运行开启
   pid进行文件名称
   log日志文件名称
   dump.rdb持久化文件名称
4. 修改完毕之后，启动我们的3个redis服务器，可以通过进程信息查看！
5. 配置一主二从
   默认情况下，每台redis服务器都是主节点，我们只需要配置从机就好了
   命令格式： slaveof host port
   通过命令的方式只是临时的，所以需要通过配置文件进行配置主从机！

# 在两个从机上进行配置
127.0.0.1:6380> slaveof 127.0.0.1 6379    # 找谁当自己的老大
OK
127.0.0.1:6380> info replication
# Replication
role:slave    # 变成了从机
master_host:127.0.0.1    # 主机的ip地址
master_port:6379    # 主机的端口port
master_link_status:up
master_last_io_seconds_ago:2
master_sync_in_progress:0
slave_repl_offset:14
slave_priority:100
slave_read_only:1
connected_slaves:0
master_replid:4864fe3562d3d55a78052a4c27e8afb8a7ace8cd
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:14
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:14

# 在主机中查看
127.0.0.1:6379> info replication
# Replication
role:master
connected_slaves:2    # 多了从机的配置
slave0:ip=127.0.0.1,port=6380,state=online,offset=182,lag=0    # 多了从机的配置
slave1:ip=127.0.0.1,port=6381,state=online,offset=182,lag=0
master_replid:4864fe3562d3d55a78052a4c27e8afb8a7ace8cd
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:182
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:182

复制原理
slave 启动成功连接到 master 后会发送一个sync 同步命令

master 接到命令，启动后台的存盘进程，同时收集所有接收到的用于修改数据集的命令，在后台进程执行完毕之后，master将传送整个数据文件到slave，并完成一次完全同步。

全量复制：而slave服务在接收到数据文件之后，将其存盘并加载到内存中。

增量复制：Master继续将所有收集到的写命令依次传给slave，完成同步。

只要slave是重新连接master，一次完全同步（全量复制）将被自动执行！ 我们的数据一定可以在从机中看到！

十、哨兵模式

（自动选举老大）
之前，当主机发生宕机后，是手动进行配置，从 从机中选择一个当做主机，然后将其当做其他从机的主机。这就需要人工干预，费时费力，还会造成一段时间内服务器不可用，所以这不是一种推荐的方式。
更多的时候我会选择哨兵模式!
哨兵模式工作原理：哨兵模式是一种特殊的模式，首先Redis提供了哨兵的命令，哨兵是一个独立的进程，作为进程，它会独立运行。其原理是哨兵通过发送命令，等待Redis服务器响应，从而监控运行的多个Redis实例。
这里哨兵有两个作用：

• 通过发送命令，让Redis服务器返回监控其运行状态，包括主服务器和从服务器

• 当哨兵监测到master宕机，会自动将slave 切换为master，然后通过订阅发布方式通知其他的从Redis服务器，修改配置文件，让他们切换主机。

然而一个哨兵进程对Redis服务器进行监控，可能会出现问题，就是如果哨兵挂了怎么办？

为此，我们可以使用多个哨兵进行监控。各个哨兵之间还会进行监控（是否有哨兵挂了 和 投票选举 主服务器的操作），这样就形成了多少哨兵模式。

假如主服务器宕机，哨兵1先检测到这个结果，系统并不会马上进行failover[故障转移]操作，仅仅是哨兵1主观的认为主服务器不可用，这个现象称为主观下线。

当后面的哨兵也检测到主服务器不可用，并且数量达到一定值时，那么哨兵之间就会进行一次投票，投票的结果由一个哨兵发起，进行failover[故障转移]操作。切换成功后，就会通过发布订阅模式，让各个哨兵把自己监控的从服务器实现切换主机，这个过程称为客观下线。

当宕机后的主机 重新恢复后，他就会归并到新的主机上，而不是重新成为主机。

优缺点
优点
1、哨兵集群，基于主从复制模式，所有的主从配置优点，它全有

2、主从可以复制，故障可以转移，系统的可用性就会更好。

3、哨兵模式就是主从模式的升级，手动到自动，更加健壮！

缺点
1、Redis 不好在线扩展，集群容量一旦达到上限，在线扩容就十分麻烦！

2、实现哨兵模式的配置其实是很麻烦的，里面有很多选择！

如果现在有10台Redis服务器都满了，再加1台，因为配置都是写死的，那么就需要改很多的配置文件

# Example sentinel.conf

# 哨兵sentinel实例运行的端口 默认26379 
port 26379 

# 哨兵sentinel的工作目录 
dir /tmp

# 哨兵sentinel监控的redis主节点的 ip port 
# master-name 可以自己命名的主节点名字 只能由字母A-z、数字0-9 、这三个字符".-_"组成。 
# quorum 配置多少个sentinel哨兵统一认为master主节点失联 那么这时客观上认为主节点失联了 
# sentinel monitor <master-name> <ip> <redis-port> <quorum>
sentinel monitor mymaster 127.0.0.1 6379 2

# 当在Redis实例中开启了requirepass foobared 授权密码 这样所有连接Redis实例的客户端都要提供 密码
# 设置哨兵sentinel 连接主从的密码 注意必须为主从设置一样的验证密码
# sentinel auth-pass <master-name> <password>
sentinel auth-pass mymaster MySUPER--secret-0123passw0rd

# 指定多少毫秒之后 主节点没有应答哨兵sentinel 此时 哨兵主观上认为主节点下线 默认30秒
# sentinel down-after-milliseconds <master-name> <milliseconds>
sentinel down-after-milliseconds mymaster 30000

# 这个配置项指定了在发生failover主从切换时，最多可以有多少个slave同时对新的master进行同步，这个数字越小，完成failover所需的时间就越长，但是如果这个数字越大，就意味着越 多的slave因为replication而不可用。可以通过将这个值设为 1 来保证每次只有一个slave 处于不能处理命令请求的状态。
# 通俗来说：就是发生了故障后，有一个新的master节点产生，其他的slave需要同步这个主节点的命令，那么这里为1个，就是每次只能有1个同步命令，在同步命令的时候，该slave节点是不可用的，所以设置为1保证只有一个slave节点不可用，其他的slave节点先提供服务，慢慢的依次进行同步。所以会慢。
# sentinel parallel-syncs <master-name> <numslaves>
sentinel parallel-syncs mymaster 1

# 故障转移的超时时间 failover-timeout 可以用在以下这些方面：
#1. 同一个sentinel对同一个master两次failover之间的间隔时间。
#2. 当一个slave从一个错误的master那里同步数据开始计算时间。直到slave被纠正为向正确的master那 里同步数据时。
#3. 当想要取消一个正在进行的failover所需要的时间。 
#4. 当进行failover时，配置所有slaves指向新的master所所需的最大时间。不过，即使过了这个超时， slaves依然会被正确配置为指向master，但是就不按parallel-syncs所配置的规则来了
# 默认三分钟
# sentinel failover-timeout <master-name> <milliseconds>
sentinel failover-timeout mymaster 180000

# SCRIPTS EXECUTION
#配置当某一事件发生时所需要执行的脚本，可以通过脚本来通知管理员，例如当系统运行不正常时发邮件通知 相关人员。
#对于脚本的运行结果有以下规则： 
#若脚本执行后返回1，那么该脚本稍后将会被再次执行，重复次数目前默认为10 
#若脚本执行后返回2，或者比2更高的一个返回值，脚本将不会重复执行。 
#如果脚本在执行过程中由于收到系统中断信号被终止了，则同返回值为1时的行为相同。 
#一个脚本的最大执行时间为60s，如果超过这个时间，脚本将会被一个SIGKILL信号终止，之后重新执行。 
#通知型脚本:当sentinel有任何警告级别的事件发生时（比如说redis实例的主观失效和客观失效等等）， 将会去调用这个脚本，这时这个脚本应该通过邮件，SMS等方式去通知系统管理员关于系统不正常运行的信息。调用该脚本时，将传给脚本两个参数，一个是事件的类型，一个是事件的描述。如果sentinel.conf配置文件中配置了这个脚本路径，那么必须保证这个脚本存在于这个路径，并且是可执行的，否则sentinel无法正常启动成功。
# 通知脚本 
# 简单的来说，发生故障转移后就会发送邮件给管理员，我们可以自己编写，但是需要学习linux的shell编程才可以编写
# sentinel notification-script <master-name> <script-path>
sentinel notification-script mymaster /var/redis/notify.sh


# 客户端重新配置主节点参数脚本 
# 当一个master由于failover而发生改变时，这个脚本将会被调用，通知相关的客户端关于master地址已经发生改变的信息。 
# 以下参数将会在调用脚本时传给脚本: 
# <master-name> <role> <state> <from-ip> <from-port> <to-ip> <to-port> 
# 目前<state>总是“failover”, 
# <role>是“leader”或者“observer”中的一个。 
# 参数 from-ip, from-port, to-ip, to-port是用来和旧的master和新的master(即旧的slave)通信的
# 这个脚本应该是通用的，能被多次调用，不是针对性的。 
# sentinel client-reconfig-script <master-name> <script-path>
sentinel client-reconfig-script mymaster /var/redis/reconfig.sh # 一般都是由运维来配 置！

十一、缓存穿透和雪崩

缓存的穿透的概念很简单，用户想要查询一个数据，发现redis内存数据库没有，也就是缓存未命中，于是向持久层数据库查询。发现也没有，于是本次查询结束。当很多的用户都发起这个请求，缓存都没有命中（比如秒杀！），于是都去请求了持久层数据库，这会给持久层数据库造成很大的压力，这时候就相当于出现了缓存穿透。

通俗来说：缓存没有，直接去查数据库，这是没有问题的，但是高并发的情况下，会压垮数据库，导致宕机。

解决方案：

1. 布隆过滤器是一种数据结构，布隆过滤器对所有可能查询的参数以hash形式存储，在控制层先进行校验，不符合则丢弃，从而避免了对底层存储系统的查询压力
   
2. 当存储层不命中后，即使返回空的对象，也进行缓存起来，同时会设置一个过期时间，之后访问这个数据将会从缓存中获取，保护了后端数据源
   
   缓存击穿
   这里需要注意和缓存穿透的区别。

缓存的击穿，是指一个key非常的热点，在不停的扛着大并发，大并发集中对一个点进行访问，当这个key失效的瞬间，持续的大并发就穿破缓存，直接请求数据库，就像在一堵墙上凿开了一个洞。

当某个key在过期的瞬间，有大量的请求并发访问，之类数据一般都是热点数据，由于缓存过期，会同时访问数据库来查询最新的数据，并且回写缓存，会导致数据库瞬间压力过大。

通俗来说：缓存中的数据还有10s过期，现在有上百万的并发请求正在读取，当缓存失效的一瞬间，大量的并发请求直接对存储层进行访问，一瞬间服务器扛不住，就会宕机。

解决方案：

1. 热点数据永不过期
2. 加互斥锁

缓存雪崩
缓存雪崩，是指在某一时间段，缓存集中过期实失效，或者服务宕机（断电，扫地大妈又把服务器插头拔了）！

产生雪崩的原因之一：比如在写文本的时候，马上就要到双十一零点，很快就会迎来一波抢购，这波商品时间比较集中的放入缓存，假如这波商品的缓存为一个小时，那么到了凌晨一点钟的时候，这波商品的缓存都过期了。而对这波商品的查询，都落到了数据库上，对于数据库而言，就会产生周期性的压力波峰。对于所有的请求都会达到存储层，存储层的调用量会暴增，造成存储层也会挂掉的情况。

解决方案
1、redis高可用
让这个思想的含义是，既然redis有可能挂掉，那我多增加几台redis，这样一台挂掉之后其他的还可以继续工作，其实就是搭建集群。

2、限流降级
这个解决方案的思想是，在缓存失效之后，通过加锁或者队列来控制读数据库和写缓存的线程数量。比如对某个key只允许一个线程查询数据库和写缓存，其他线程等待。

3、数据预热
数据加热的含义就是在正式部署之前，我先把可能的数据先访问一遍，这样部分可能大量访问的数据就会加载到缓存中。在即将发生的大并发访问前，手动触发加载缓存不同的key，设置不同的过期时间，让缓存失效的时间点尽量均匀，不要一瞬间都失效了，那么又会造成缓存雪崩了。

引用学习：https://space.bilibili.com/95256449/