一、Nosql概述

 

为什么使用Nosql？

1、单机Mysql时代

90年代,一个网站的访问量一般不会太大，单个数据库完全够用。随着用户增多，网站出现以下问题

• 数据量增加到一定程度，单机数据库就放不下了
• 数据的索引（B+ Tree）,一个机器内存也存放不下
• 访问量变大后（读写混合），一台服务器承受不住。 

2、Memcached(缓存) + Mysql + 垂直拆分（读写分离） 

网站80%的情况都是在读，每次都要去查询数据库的话就十分的麻烦！所以说我们希望减轻数据库的压力，我们可以使用缓存来保证效率！ 

优化过程经历了以下几个过程：

• 优化数据库的数据结构和索引(难度大)
• 文件缓存，通过IO流获取比每次都访问数据库效率略高，但是流量爆炸式增长时候，IO流也承受不了
• MemCache,当时最热门的技术，通过在数据库和数据库访问层之间加上一层缓存，第一次访问时查询数据库，将结果保存到缓存，后续的查询先检查缓存，若有直接拿去使用，效率显著提升。

3、分库分表 + 水平拆分 + Mysql集群

4、如今最近的年代

​如今信息量井喷式增长，各种各样的数据出现（用户定位数据，图片数据等），大数据的背景下关系型数据库（RDBMS）无法满足大量数据要求。Nosql数据库就能轻松解决这些问题。

目前一个基本的互联网项目 

为什么要用NoSQL ？ 

用户的个人信息，社交网络，地理位置。用户自己产生的数据，用户日志等等爆发式增长！
这时候我们就需要使用NoSQL数据库的，Nosql可以很好的处理以上的情况！

什么是Nosql 

NoSQL = Not Only SQL（不仅仅是SQL）

​Not Only Structured Query Language

关系型数据库：列+行，同一个表下数据的结构是一样的。

非关系型数据库：数据存储没有固定的格式，并且可以进行横向扩展。

NoSQL泛指非关系型数据库，随着web2.0互联网的诞生，传统的关系型数据库很难对付web2.0时代！尤其是超大规模的高并发的社区，暴露出来很多难以克服的问题，NoSQL在当今大数据环境下发展的十分迅速，Redis是发展最快的。

Nosql特点

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

传统的 RDBMS(关系型数据库)
- 结构化组织
- SQL
- 数据和关系都存在单独的表中 row col
- 操作，数据定义语言
- 严格的一致性
- 基础的事务
- ...

Nosql
- 不仅仅是数据
- 没有固定的查询语言
- 键值对存储，列存储，文档存储，图形数据库（社交关系）
- 最终一致性
- CAP定理和BASE
- 高性能，高可用，高扩展
- ...

了解：3V + 3高

大数据时代的3V ：主要是描述问题的 

1. 海量(Velume)
2. 多样(Variety)
3. 实时(Velocity)

 大数据时代的3高 ： 主要是对程序的要求

1. 高并发
2. 高可扩
3. 高性能

真正在公司中的实践：NoSQL + RDBMS 一起使用才是最强的。

阿里巴巴演进分析

推荐阅读：阿里云的这群疯子: https://yq.aliyun.com/articles/653511 

# 商品信息
- 一般存放在关系型数据库：Mysql,阿里巴巴使用的Mysql都是经过内部改动的。# 商品描述、评论(文字居多)
- 文档型数据库：MongoDB# 图片
- 分布式文件系统 FastDFS
- 淘宝：TFS
- Google: GFS
- Hadoop: HDFS
- 阿里云: oss# 商品关键字 用于搜索
- 搜索引擎：solr,elasticsearch
- 阿里：Isearch 多隆# 商品热门的波段信息
- 内存数据库：Redis，Memcache# 商品交易，外部支付接口
- 第三方应用

Nosql的四大分类

KV键值对 

• 新浪：Redis
• 美团：Redis + Tair
• 阿里、百度：Redis + Memcache

文档型数据库（bson数据格式）：

• MongoDB(掌握)
  • 基于分布式文件存储的数据库。C++编写，用于处理大量文档。
  • MongoDB是RDBMS和NoSQL的中间产品。MongoDB是非关系型数据库中功能最丰富的，NoSQL中最像关系型数据库的数据库。
• ConthDB

列存储数据库 

• HBase(大数据必学)
• 分布式文件系统

图形关系数据库

用于广告推荐，社交网络 

• Neo4j
• InfoGrid

二、Redis入门

概述

Redis是什么？

Redis（Remote Dictionary Server )，即远程字典服务。

是一个开源的使用ANSI C语言编写、支持网络、可基于内存亦可持久化的日志型、Key-Value数据库，并提供多种语言的API。

与memcached一样，为了保证效率，数据都是缓存在内存中。区别的是redis会周期性的把更新的数据写入磁盘或者把修改操作写入追加的记录文件，并且在此基础上实现了master-slave(主从)同步。

Redis能该干什么？

1. 内存存储、持久化，内存是断电即失的，所以需要持久化（RDB、AOF）
2. 高效率、用于高速缓冲
3. 发布订阅系统
4. 地图信息分析
5. 计时器、计数器(eg：浏览量)
6. 。。。

特性 

1. 多样的数据类型

2. 持久化

3. 集群

4. 事务

   …

环境搭建

官网：https://redis.io/

推荐使用Linux服务器学习。

windows版本的Redis已经停更很久了…

Windows安装

GitHub – dmajkic/redis: Redis key-value store (Win32 / Win64 port)

1. 解压安装包

1. 开启redis-server.exe

2. 启动redis-cli.exe测试

Linux安装

1. 下载安装包！redis-5.0.8.tar.gz
2. 解压Redis的安装包！程序一般放在 /opt 目录下

     3. 进入解压后的文件,可以看到我们redis的配置文件

     4. 基本的环境安装命令

yum install gcc-c++
# 然后进入redis目录下执行
make
# 然后执行
make install

 这里make命令输入之后, 需要等待一段时间.

       5. redis默认安装路径 /usr/local/bin

      6. 将redis的配置文件复制到 程序安装目录 /usr/local/bin/kconfig下(这个kconfig文件夹可以自己创建一个)

       7. redis默认不是后台启动的，需要修改配置文件！

       8. 通过制定的配置文件启动redis服务

       9. 使用redis-cli连接指定的端口号测试，Redis的默认端口6379

      10. 查看redis进程是否开启

      11. 关闭Redis服务 shutdown 

     12. 再次查看进程是否存在

     13. 后面我们会使用单机多Redis启动集群测试

测试性能

redis-benchmark： Redis官方提供的性能测试工具，参数选项如下：

简单测试：

# 测试：100个并发连接 100000请求
redis-benchmark -h localhost -p 6379 -c 100 -n 100000

如何查看这些分析呢？

基础知识

redis默认有16个数据库

默认使用的第0个;

16个数据库为：DB 0~DB 15
默认使用DB 0 ，可以使用select n切换到DB n，dbsize可以查看当前数据库的大小，与key数量相关。

127.0.0.1:6379> config get databases # 命令行查看数据库数量databases
1) "databases"
2) "16"127.0.0.1:6379> select 8 # 切换数据库 DB 8
OK
127.0.0.1:6379[8]> dbsize # 查看数据库大小
(integer) 0# 不同数据库之间 数据是不能互通的，并且dbsize 是根据库中key的个数。
127.0.0.1:6379> set name sakura 
OK
127.0.0.1:6379> SELECT 8
OK
127.0.0.1:6379[8]> get name # db8中并不能获取db0中的键值对。
(nil)
127.0.0.1:6379[8]> DBSIZE
(integer) 0
127.0.0.1:6379[8]> SELECT 0
OK
127.0.0.1:6379> keys *
1) "counter:__rand_int__"
2) "mylist"
3) "name"
4) "key:__rand_int__"
5) "myset:__rand_int__"
127.0.0.1:6379> DBSIZE # size和key个数相关
(integer) 5

keys * ：查看当前数据库中所有的key。

flushdb：清空当前数据库中的键值对。

flushall：清空所有数据库的键值对。

Redis是单线程的，Redis是基于内存操作的。

所以Redis的性能瓶颈不是CPU,而是机器内存和网络带宽。

那么为什么Redis的速度如此快呢，性能这么高呢？QPS达到10W+

Redis为什么单线程还这么快？

• 误区1：高性能的服务器一定是多线程的？
• 误区2：多线程（CPU上下文会切换！）一定比单线程效率高！

核心：Redis是将所有的数据放在内存中的，所以说使用单线程去操作效率就是最高的，多线程（CPU上下文会切换：耗时的操作！）， 对于内存系统来说，如果没有上下文切换效率就是最高的，多次读写都是在一个CPU上的， 在内存存储数据情况下，单线程就是最佳的方案。 

三、五大数据类型

Redis是一个开源（BSD许可），内存存储的数据结构服务器，可用作数据库，高速缓存和消息队列代理。它支持字符串、哈希表、列表、集合、有序集合，位图，hyperloglogs等数据类型。内置复制、Lua脚本、LRU收回、事务以及不同级别磁盘持久化功能，同时通过Redis Sentinel提供高可用，通过Redis Cluster提供自动分区。

Redis-key

在redis中无论什么数据类型，在数据库中都是以key-value形式保存，通过进行对Redis-key的操作，来完成对数据库中数据的操作。

下面学习的命令：

• exists key：判断键是否存在
• del key：删除键值对
• move key db：将键值对移动到指定数据库
• expire key second：设置键值对的过期时间
• type key：查看value的数据类型

127.0.0.1:6379> keys * # 查看当前数据库所有key
(empty list or set)
127.0.0.1:6379> set name qinjiang # set key
OK
127.0.0.1:6379> set age 20
OK
127.0.0.1:6379> keys *
1) "age"
2) "name"
127.0.0.1:6379> move age 1 # 将键值对移动到指定数据库
(integer) 1
127.0.0.1:6379> EXISTS age # 判断键是否存在
(integer) 0 # 不存在
127.0.0.1:6379> EXISTS name
(integer) 1 # 存在
127.0.0.1:6379> SELECT 1
OK
127.0.0.1:6379[1]> keys *
1) "age"
127.0.0.1:6379[1]> del age # 删除键值对
(integer) 1 # 删除个数127.0.0.1:6379> set age 20
OK
127.0.0.1:6379> EXPIRE age 15 # 设置键值对的过期时间(integer) 1 # 设置成功 开始计数
127.0.0.1:6379> ttl age # 查看key的过期剩余时间
(integer) 13
127.0.0.1:6379> ttl age
(integer) 11
127.0.0.1:6379> ttl age
(integer) 9
127.0.0.1:6379> ttl age
(integer) -2 # -2 表示key过期，-1表示key未设置过期时间127.0.0.1:6379> get age # 过期的key 会被自动delete
(nil)
127.0.0.1:6379> keys *
1) "name"127.0.0.1:6379> type name # 查看value的数据类型
string

关于TTL命令

Redis的key，通过TTL命令返回key的过期时间，一般来说有3种：

1. 当前key没有设置过期时间，所以会返回-1.
2. 当前key有设置过期时间，而且key已经过期，所以会返回-2.
3. 当前key有设置过期时间，且key还没有过期，故会返回key的正常剩余时间.

关于重命名RENAME和RENAMENX 

• RENAME key newkey修改 key 的名称
• RENAMENX key newkey仅当 newkey 不存在时，将 key 改名为 newkey 。

更多命令学习：https://www.redis.net.cn/order/

String(字符串)

90%的程序员使用redis只会使用string类型。 api调用工程师

127.0.0.1:6379> set views 0     
OK
127.0.0.1:6379> get views
"0"
127.0.0.1:6379> incr views    #自增一
(integer) 1
127.0.0.1:6379> incr views
(integer) 2
127.0.0.1:6379> get views
"2"
127.0.0.1:6379> decr views     #自减一
(integer) 1
127.0.0.1:6379> incrby views 10   #自增10
(integer) 11
127.0.0.1:6379> incrby views 10
(integer) 21
127.0.0.1:6379> decrby views 5   #自减5
(integer) 16
127.0.0.1:6379> 

 截取字符串getrange

127.0.0.1:6379> clear
127.0.0.1:6379> keys *
(empty array)
127.0.0.1:6379> set key1 "fang"
OK
127.0.0.1:6379> get key1
"fang"
127.0.0.1:6379> getrange key1 0 2  #截取012字符串
"fan"
127.0.0.1:6379> getrange key1 0 -1 #获取全部字符串
"fang"
127.0.0.1:6379> 

 替换setrange

127.0.0.1:6379> set key2 abcdefg
OK
127.0.0.1:6379> get key2
"abcdefg"
127.0.0.1:6379> setrange key2 1 xx
(integer) 7
127.0.0.1:6379> get key2
"axxdefg"
127.0.0.1:6379> 

setex #设置过期时间
setnx #不存在在设置 （在分布式锁中常常使用）

127.0.0.1:6379> clear
127.0.0.1:6379> setex key3 30 "hello"
OK
127.0.0.1:6379> ttl key3
(integer) 17
127.0.0.1:6379> setnx mykey "redsi" #如果mykey存在，则创建失败。
(integer) 1
127.0.0.1:6379> keys *
1) "mykey"
2) "key2"
3) "key1"
127.0.0.1:6379> ttl key3
(integer) -2
127.0.0.1:6379> setnx mykey "MongDB"
(integer) 0
127.0.0.1:6379> get mykey
"redsi"

一次性获取，设置多个值：mset，mget

127.0.0.1:6379> clear
127.0.0.1:6379> keys *
(empty array)
127.0.0.1:6379> mset k1 v1 k2 v2 k3 v3     #同时设置多个值
OK
127.0.0.1:6379> keys *
1) "k3"
2) "k2"
3) "k1"
127.0.0.1:6379> mget k1 k2 k3    #同时获取多个值
1) "v1"
2) "v2"
3) "v3"
127.0.0.1:6379> msetnx k1 v1 k4 v4   #msetnx是一个原子性操作，要么成功要么失败
(integer) 0
127.0.0.1:6379> get k4
(nil)
127.0.0.1:6379> 

对象
mset user：1{name：zhangsan，age：3}#设置一个user：1对象 值为json字符来保存一个对象

127.0.0.1:6379> mset user:1:name fang user:1:age 2
OK
127.0.0.1:6379> mget user:1:name user:1:age
1) "fang"
2) "2"
127.0.0.1:6379> 

先get在set——-getset

127.0.0.1:6379> clear
127.0.0.1:6379> getset db redis
(nil)
127.0.0.1:6379> get db
"redis"
127.0.0.1:6379> getset db mongodb
"redis"
127.0.0.1:6379> get db
"mongodb"

string类型的使用场景：value除了字符串还可以是数字

1.计数器
2.统计多单位数量
3.粉丝数
4.对象缓存存储！

List(列表)

Redis列表是简单的字符串列表，按照插入顺序排序。你可以添加一个元素到列表的头部（左边）或者尾部（右边）一个列表最多可以包含 2^32 – 1 个元素 (4294967295, 每个列表超过40亿个元素)。

首先我们列表，可以经过规则定义将其变为队列、栈、双端队列等

正如图 Redis中List是可以进行双端操作的，所以命令也就分为了LXXX和RLLL两类， 有时候L也表示List例如LLEN

在redis里面可以把list完成栈，队列，阻塞队列！
所有的list命令以l开头

---------------------------LPUSH---RPUSH---LRANGE--------------------------------127.0.0.1:6379> LPUSH mylist k1 # LPUSH mylist=>{k2}
(integer) 1
127.0.0.1:6379> LPUSH mylist k2 # LPUSH mylist=>{k2,k1}
(integer) 2
127.0.0.1:6379> RPUSH mylist k3 # RPUSH mylist=>{k2,k1,k3}
(integer) 3
127.0.0.1:6379> get mylist # 普通的get是无法获取list值的
(error) WRONGTYPE Operation against a key holding the wrong kind of value
127.0.0.1:6379> LRANGE mylist 0 4 # LRANGE 获取起止位置范围内的元素
1) "k2"
2) "k1"
3) "k3"
127.0.0.1:6379> LRANGE mylist 0 2
1) "k2"
2) "k1"
3) "k3"
127.0.0.1:6379> LRANGE mylist 0 1
1) "k2"
2) "k1"
127.0.0.1:6379> LRANGE mylist 0 -1 # 获取全部元素
1) "k2"
2) "k1"
3) "k3"---------------------------LPUSHX---RPUSHX-----------------------------------127.0.0.1:6379> LPUSHX list v1 # list不存在 LPUSHX失败
(integer) 0
127.0.0.1:6379> LPUSHX list v1 v2  
(integer) 0
127.0.0.1:6379> LPUSHX mylist k4 k5 # 向mylist中 左边 PUSH k4 k5
(integer) 5
127.0.0.1:6379> LRANGE mylist 0 -1
1) "k5"
2) "k4"
3) "k2"
4) "k1"
5) "k3"---------------------------LINSERT--LLEN--LINDEX--LSET----------------------------127.0.0.1:6379> LINSERT mylist after k2 ins_key1 # 在k2元素后 插入ins_key1
(integer) 6
127.0.0.1:6379> LRANGE mylist 0 -1
1) "k5"
2) "k4"
3) "k2"
4) "ins_key1"
5) "k1"
6) "k3"
127.0.0.1:6379> LLEN mylist # 查看mylist的长度
(integer) 6
127.0.0.1:6379> LINDEX mylist 3 # 获取下标为3的元素
"ins_key1"
127.0.0.1:6379> LINDEX mylist 0
"k5"
127.0.0.1:6379> LSET mylist 3 k6 # 将下标3的元素 set值为k6
OK
127.0.0.1:6379> LRANGE mylist 0 -1
1) "k5"
2) "k4"
3) "k2"
4) "k6"
5) "k1"
6) "k3"---------------------------LPOP--RPOP--------------------------127.0.0.1:6379> LPOP mylist # 左侧(头部)弹出
"k5"
127.0.0.1:6379> RPOP mylist # 右侧(尾部)弹出
"k3"---------------------------RPOPLPUSH--------------------------127.0.0.1:6379> LRANGE mylist 0 -1
1) "k4"
2) "k2"
3) "k6"
4) "k1"
127.0.0.1:6379> RPOPLPUSH mylist newlist # 将mylist的最后一个值(k1)弹出，加入到newlist的头部
"k1"
127.0.0.1:6379> LRANGE newlist 0 -1
1) "k1"
127.0.0.1:6379> LRANGE mylist 0 -1
1) "k4"
2) "k2"
3) "k6"---------------------------LTRIM--------------------------127.0.0.1:6379> LTRIM mylist 0 1 # 截取mylist中的 0~1部分
OK
127.0.0.1:6379> LRANGE mylist 0 -1
1) "k4"
2) "k2"# 初始 mylist: k2,k2,k2,k2,k2,k2,k4,k2,k2,k2,k2
---------------------------LREM--------------------------127.0.0.1:6379> LREM mylist 3 k2 # 从头部开始搜索 至多删除3个 k2
(integer) 3
# 删除后：mylist: k2,k2,k2,k4,k2,k2,k2,k2127.0.0.1:6379> LREM mylist -2 k2 #从尾部开始搜索 至多删除2个 k2
(integer) 2
# 删除后：mylist: k2,k2,k2,k4,k2,k2---------------------------BLPOP--BRPOP--------------------------mylist: k2,k2,k2,k4,k2,k2
newlist: k1127.0.0.1:6379> BLPOP newlist mylist 30 # 从newlist中弹出第一个值，mylist作为候选
1) "newlist" # 弹出
2) "k1"
127.0.0.1:6379> BLPOP newlist mylist 30
1) "mylist" # 由于newlist空了 从mylist中弹出
2) "k2"

小结 

• list实际上是一个链表，before Node after , left, right 都可以插入值
• 如果key不存在，则创建新的链表
• 如果key存在，新增内容
• 如果移除了所有值，空链表，也代表不存在
• 在两边插入或者改动值，效率最高！修改中间元素，效率相对较低

应用：

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

Set(集合)

Redis的Set是string类型的无序集合。集合成员是唯一的，这就意味着集合中不能出现重复的数据。

Redis 中 集合是通过哈希表实现的，所以添加，删除，查找的复杂度都是O(1)。

集合中最大的成员数为 2^32 – 1 (4294967295, 每个集合可存储40多亿个成员)。

---------------SADD--SCARD--SMEMBERS--SISMEMBER--------------------127.0.0.1:6379> SADD myset m1 m2 m3 m4 # 向myset中增加成员 m1~m4
(integer) 4
127.0.0.1:6379> SCARD myset # 获取集合的成员数目
(integer) 4
127.0.0.1:6379> smembers myset # 获取集合中所有成员
1) "m4"
2) "m3"
3) "m2"
4) "m1"
127.0.0.1:6379> SISMEMBER myset m5 # 查询m5是否是myset的成员
(integer) 0 # 不是，返回0
127.0.0.1:6379> SISMEMBER myset m2
(integer) 1 # 是，返回1
127.0.0.1:6379> SISMEMBER myset m3
(integer) 1---------------------SRANDMEMBER--SPOP----------------------------------127.0.0.1:6379> SRANDMEMBER myset 3 # 随机返回3个成员
1) "m2"
2) "m3"
3) "m4"
127.0.0.1:6379> SRANDMEMBER myset # 随机返回1个成员
"m3"
127.0.0.1:6379> SPOP myset 2 # 随机移除并返回2个成员
1) "m1"
2) "m4"
# 将set还原到{m1,m2,m3,m4}---------------------SMOVE--SREM----------------------------------------127.0.0.1:6379> SMOVE myset newset m3 # 将myset中m3成员移动到newset集合
(integer) 1
127.0.0.1:6379> SMEMBERS myset
1) "m4"
2) "m2"
3) "m1"
127.0.0.1:6379> SMEMBERS newset
1) "m3"
127.0.0.1:6379> SREM newset m3 # 从newset中移除m3元素
(integer) 1
127.0.0.1:6379> SMEMBERS newset
(empty list or set)# 下面开始是多集合操作,多集合操作中若只有一个参数默认和自身进行运算
# setx=>{m1,m2,m4,m6}, sety=>{m2,m5,m6}, setz=>{m1,m3,m6}-----------------------------SDIFF------------------------------------127.0.0.1:6379> SDIFF setx sety setz # 等价于setx-sety-setz
1) "m4"
127.0.0.1:6379> SDIFF setx sety # setx - sety
1) "m4"
2) "m1"
127.0.0.1:6379> SDIFF sety setx # sety - setx
1) "m5"-------------------------SINTER---------------------------------------
# 共同关注（交集）127.0.0.1:6379> SINTER setx sety setz # 求 setx、sety、setx的交集
1) "m6"
127.0.0.1:6379> SINTER setx sety # 求setx sety的交集
1) "m2"
2) "m6"-------------------------SUNION---------------------------------------127.0.0.1:6379> SUNION setx sety setz # setx sety setz的并集
1) "m4"
2) "m6"
3) "m3"
4) "m2"
5) "m1"
6) "m5"
127.0.0.1:6379> SUNION setx sety # setx sety 并集
1) "m4"
2) "m6"
3) "m2"
4) "m1"
5) "m5"

应用场景：

微博，b站共同关注（并集）
数字集合类：
差集，交集，并集

微博，将a用户所有关注的人放在一个集合中，将他的粉丝也放在一个集合中！
共同关注，共同爱好

Hash（哈希）

Redis hash 是一个string类型的field和value的映射表，hash特别适合用于存储对象。

Set就是一种简化的Hash,只变动key,而value使用默认值填充。可以将一个Hash表作为一个对象进行存储，表中存放对象的信息。

------------------------HSET--HMSET--HSETNX----------------
127.0.0.1:6379> HSET studentx name sakura # 将studentx哈希表作为一个对象，设置name为sakura
(integer) 1
127.0.0.1:6379> HSET studentx name gyc # 重复设置field进行覆盖，并返回0
(integer) 0
127.0.0.1:6379> HSET studentx age 20 # 设置studentx的age为20
(integer) 1
127.0.0.1:6379> HMSET studentx sex 1 tel 15623667886 # 设置sex为1，tel为15623667886
OK
127.0.0.1:6379> HSETNX studentx name gyc # HSETNX 设置已存在的field
(integer) 0 # 失败
127.0.0.1:6379> HSETNX studentx email 12345@qq.com
(integer) 1 # 成功----------------------HEXISTS--------------------------------
127.0.0.1:6379> HEXISTS studentx name # name字段在studentx中是否存在
(integer) 1 # 存在
127.0.0.1:6379> HEXISTS studentx addr
(integer) 0 # 不存在-------------------HGET--HMGET--HGETALL-----------
127.0.0.1:6379> HGET studentx name # 获取studentx中name字段的value
"gyc"
127.0.0.1:6379> HMGET studentx name age tel # 获取studentx中name、age、tel字段的value
1) "gyc"
2) "20"
3) "15623667886"
127.0.0.1:6379> HGETALL studentx # 获取studentx中所有的field及其value1) "name"2) "gyc"3) "age"4) "20"5) "sex"6) "1"7) "tel"8) "15623667886"9) "email"
10) "12345@qq.com"--------------------HKEYS--HLEN--HVALS--------------
127.0.0.1:6379> HKEYS studentx # 查看studentx中所有的field
1) "name"
2) "age"
3) "sex"
4) "tel"
5) "email"
127.0.0.1:6379> HLEN studentx # 查看studentx中的字段数量
(integer) 5
127.0.0.1:6379> HVALS studentx # 查看studentx中所有的value
1) "gyc"
2) "20"
3) "1"
4) "15623667886"
5) "12345@qq.com"-------------------------HDEL--------------------------
127.0.0.1:6379> HDEL studentx sex tel # 删除studentx 中的sex、tel字段
(integer) 2
127.0.0.1:6379> HKEYS studentx
1) "name"
2) "age"
3) "email"-------------HINCRBY--HINCRBYFLOAT------------------------
127.0.0.1:6379> HINCRBY studentx age 1 # studentx的age字段数值+1
(integer) 21
127.0.0.1:6379> HINCRBY studentx name 1 # 非整数字型字段不可用
(error) ERR hash value is not an integer
127.0.0.1:6379> HINCRBYFLOAT studentx weight 0.6 # weight字段增加0.6
"90.8"

Hash变更的数据user name age，尤其是用户信息之类的，经常变动的信息！Hash更适合于对象的存储，Sring更加适合字符串存储！

四、三种特殊数据类型

Geospatial(地理位置)

使用经纬度定位地理坐标并用一个有序集合zset保存，所以zset命令也可以使用

有效经纬度

• 有效的经度从-180度到180度。
• 有效的纬度从-85.05112878度到85.05112878度。

指定单位的参数 unit 必须是以下单位的其中一个：

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

应用场景:

朋友定位，附近的人，打车距离技术，这个功能可以推算地理位置信息，两地之间的距离，方圆几里的人！
可以查询一些测试数据

只有六个命令！

geoadd(添加数据)

规则：两级无法添加，我们一般会下载城市数据，直接通过java程序一次性导入！

参数（经度 纬度 名称）

超过有效的经度纬度就会报错

127.0.0.1:6379> geoadd china:city 116.40 39.90 beijin
(integer) 1
127.0.0.1:6379> geoadd china:city 121.47 31.23 shanghai
(integer) 1
127.0.0.1:6379> geoadd china:city 106.50 29.53 chongqi
(integer) 1
127.0.0.1:6379> geoadd china:city 114.05 22.52 shengzhen
(integer) 1
127.0.0.1:6379> geoadd china:city 120.16 30.24 hangzhou 118.96 34.26 xian
(integer) 2
127.0.0.1:6379> 

geopos(获取当前定位：一定是一个坐标值)

127.0.0.1:6379> geopos china:city beijin
1) 1) "116.39999896287918091"2) "39.90000009167092543"
127.0.0.1:6379> geopos china:city beijin chongqi
1) 1) "116.39999896287918091"2) "39.90000009167092543"
2) 1) "106.49999767541885376"2) "29.52999957900659211"
127.0.0.1:6379> 

geodist(两人之间的距离)

127.0.0.1:6379> geodist china:city beijin shanghai
"1067378.7564"   #北京上海的==直线距离，默认单位为米==
127.0.0.1:6379> geodist china:city beijin shanghai km
"1067.3788"  

georadius(已给定的经纬度为中心，找某一半径内的元素)

我附近的人？（获得所有附近的人的地址，定位！）通过半径来查询

127.0.0.1:6379> georadius china:city 110 30 1000 km #以100经度30纬度为中心1000km为半径的圆内城市
1) "chongqi"
2) "shengzhen"
3) "hangzhou"
4) "xian"
127.0.0.1:6379> georadius china:city 110 30 500 km
1) "chongqi"
127.0.0.1:6379> georadius china:city 110 30 500 km withdist #显示到中心位置的距离
1) 1) "chongqi"2) "341.9374"
127.0.0.1:6379> georadius china:city 110 30 500 km withcoord #显示他人的定位信息
1) 1) "chongqi"2) 1) "106.49999767541885376"2) "29.52999957900659211"
127.0.0.1:6379> georadius china:city 110 30 500 km withdist withcoord count 1 #筛选出指定数量的结果
1) 1) "chongqi"2) "341.9374"3) 1) "106.49999767541885376"2) "29.52999957900659211"
127.0.0.1:6379> georadius china:city 110 30 500 km withdist withcoord count 2
1) 1) "chongqi"2) "341.9374"3) 1) "106.49999767541885376"2) "29.52999957900659211"
127.0.0.1:6379> georadius china:city 110 30 500 km withdist withcoord count 3
1) 1) "chongqi"2) "341.9374"3) 1) "106.49999767541885376"2) "29.52999957900659211"
127.0.0.1:6379> 

关于GEORADIUS的参数

通过georadius就可以完成 附近的人功能

withcoord:带上坐标

withdist:带上距离，单位与半径单位相同

COUNT n : 只显示前n个(按距离递增排序)

----------------georadius---------------------
127.0.0.1:6379> GEORADIUS china:city 120 30 500 km withcoord withdist # 查询经纬度(120,30)坐标500km半径内的成员
1) 1) "hangzhou"2) "29.4151"3) 1) "120.20000249147415"2) "30.199999888333501"
2) 1) "shanghai"2) "205.3611"3) 1) "121.40000134706497"2) "31.400000253193539"------------geohash---------------------------
127.0.0.1:6379> geohash china:city yichang shanghai # 获取成员经纬坐标的geohash表示
1) "wmrjwbr5250"
2) "wtw6ds0y300"

georadiusbymember(找出指定城市周围的位置)

127.0.0.1:6379> georadiusbymember china:city beijin 1000 km
1) "xian"
2) "beijin"
127.0.0.1:6379> 

geohash(该命令返回长度为11的字符串)

127.0.0.1:6379> geohash china:city beijin chongqi
1) "wx4fbxxfke0"    #将二维的经纬度转换为一维的字符串，如果字符串月接近，那么距离则越近
2) "wm5xzrybty0"

geo底层实现原理

原理其实就是zset，我们可以使用zset命令操作geo！

127.0.0.1:6379> zrange china:city 0 -1 ##查看元素
1) "chongqi"
2) "shengzhen"
3) "hangzhou"
4) "shanghai"
5) "xian"
6) "beijin"
127.0.0.1:6379> zrem china:city beijin
(integer) 1
127.0.0.1:6379> zrange china:city 0 -1
1) "chongqi"
2) "shengzhen"
3) "hangzhou"
4) "shanghai"
5) "xian"
127.0.0.1:6379> 

Hyperloglog(基数统计)

a{1，3，5，7，8，7}
b{1，2，5，7，8}
基数（不重复的元素）=5，可以接受误差
基数统计算法
优点：占用内存是固定的，2^64不同元素计数，只需要费12kb的内存，如果要从内存角度比较hyperloglog首选
网页的uv（一个人访问一个网站多次，但是还是算作一个人）
传统的方式，set保护用户的id，其元素不重复，然后就可以统计set中元素的数量作为标准判断
这个方式如果保持大量用户id就会比较麻烦，我们的目的就是为了计数，而不是保存用户id；
0.81%的错误率，可以接受

127.0.0.1:6379> pfadd mykey a b c d e f g h i j 
(integer) 1
127.0.0.1:6379> pfcount mykey #统计mykey元素基数数量
(integer) 10
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 #合并两组到mykey3
OK
127.0.0.1:6379> pfcount mykey3
(integer) 15
127.0.0.1:6379> 

如果允许容错统计数量，可以使用hyperloglog
如果不允许使用set或者其他数据类型

BitMaps(位图)

位存储
统计用户信息，活跃，不活跃，登入，为登入，打卡，365打卡！两个状态的都可以使用bitmaps
bitmaps位图，数据结构，都是操作二进制位来进行记录的，就只有0，1两个状态
365=365bit 1字节=8bit 46字节左右！
测试
记录周一到周日打卡
周一：1，周二：0，周三：0.。。。

127.0.0.1:6379> setbit sign 0 0
(integer) 0
127.0.0.1:6379> setbit sign 1 0
(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 0
(integer) 0
127.0.0.1:6379> setbit sign 6 0
(integer) 0
127.0.0.1:6379> 

查看某一天是否打卡

127.0.0.1:6379> getbit sign 3
(integer) 0
127.0.0.1:6379> getbit sign 4
(integer) 1
127.0.0.1:6379> 

统计打卡天数，可以看到是否有缺勤

127.0.0.1:6379> bitcount sign
(integer) 1

五、事务

redis事务本质：一组命令的集合!一个事务中的所有命令都会被序列化，执行过程中按照顺序执行
一次性，顺序性，排他性，执行一些列的命令

redis单条命令是保持原子性，但是事务不保证原子性
redis事务没有隔离级别的概念
所有的命令在事务中，并没有被直接执行，只有发起执行命令才会被执行!execRedis

事务操作过程

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

127.0.0.1:6379> multi # 开启事务
OK
127.0.0.1:6379> set k1 v1 # 命令入队
QUEUED
127.0.0.1:6379> set k2 v2 # ..
QUEUED
127.0.0.1:6379> get k1
QUEUED
127.0.0.1:6379> set k3 v3
QUEUED
127.0.0.1:6379> keys *
QUEUED
127.0.0.1:6379> exec # 事务执行
1) OK
2) OK
3) "v1"
4) OK
5) 1) "k3"2) "k2"3) "k1"

 取消事务(discurd)

127.0.0.1:6379> multi
OK
127.0.0.1:6379> set k1 v1
QUEUED
127.0.0.1:6379> set k2 v2
QUEUED
127.0.0.1:6379> DISCARD # 放弃事务
OK
127.0.0.1:6379> EXEC 
(error) ERR EXEC without MULTI # 当前未开启事务
127.0.0.1:6379> get k1 # 被放弃事务中命令并未执行
(nil)

事务错误

代码语法错误（编译时异常）所有的命令都不执行

127.0.0.1:6379> multi
OK
127.0.0.1:6379> set k1 v1
QUEUED
127.0.0.1:6379> set k2 v2
QUEUED
127.0.0.1:6379> error k1 # 这是一条语法错误命令
(error) ERR unknown command `error`, with args beginning with: `k1`, # 会报错但是不影响后续命令入队 
127.0.0.1:6379> get k2
QUEUED
127.0.0.1:6379> EXEC
(error) EXECABORT Transaction discarded because of previous errors. # 执行报错
127.0.0.1:6379> get k1 
(nil) # 其他命令并没有被执行

 代码逻辑错误 (运行时异常) **其他命令可以正常执行 ** >>> 所以不保证事务原子性

127.0.0.1:6379> multi
OK
127.0.0.1:6379> set k1 v1
QUEUED
127.0.0.1:6379> set k2 v2
QUEUED
127.0.0.1:6379> INCR k1 # 这条命令逻辑错误（对字符串进行增量）
QUEUED
127.0.0.1:6379> get k2
QUEUED
127.0.0.1:6379> exec
1) OK
2) OK
3) (error) ERR value is not an integer or out of range # 运行时报错
4) "v2" # 其他命令正常执行# 虽然中间有一条命令报错了，但是后面的指令依旧正常执行成功了。
# 所以说Redis单条指令保证原子性，但是Redis事务不能保证原子性。

监控

悲观锁：

• 很悲观，认为什么时候都会出现问题，无论做什么都会加锁

乐观锁：

• 很乐观，认为什么时候都不会出现问题，所以不会上锁！更新数据的时候去判断一下，在此期间是否有人修改过这个数据
• 获取version
• 更新的时候比较version

127.0.0.1:6379> set money 100 # 设置余额:100
OK
127.0.0.1:6379> set use 0 # 支出使用:0
OK
127.0.0.1:6379> watch money # 监视money (上锁)
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379> DECRBY money 20
QUEUED
127.0.0.1:6379> INCRBY use 20
QUEUED
127.0.0.1:6379> exec # 监视值没有被中途修改，事务正常执行
1) (integer) 80
2) (integer) 20

 测试多线程修改值，使用watch可以当做redis的乐观锁操作（相当于getversion）

我们启动另外一个客户端模拟插队线程。

线程1：

127.0.0.1:6379> watch money # money上锁
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379> DECRBY money 20
QUEUED
127.0.0.1:6379> INCRBY use 20
QUEUED
127.0.0.1:6379> 	# 此时事务并没有执行

 模拟线程插队，线程2：

127.0.0.1:6379> INCRBY money 500 # 修改了线程一中监视的money
(integer) 600

回到线程1，执行事务：

127.0.0.1:6379> EXEC # 执行之前，另一个线程修改了我们的值，这个时候就会导致事务执行失败
(nil) # 没有结果，说明事务执行失败127.0.0.1:6379> get money # 线程2 修改生效
"600"
127.0.0.1:6379> get use # 线程1事务执行失败，数值没有被修改
"0"

解锁获取最新值，然后再加锁进行事务。

unwatch进行解锁。

注意：每次提交执行exec后都会自动释放锁，不管是否成功 

六、Jedis

使用Java来操作Redis，Jedis是Redis官方推荐使用的Java连接redis的客户端。

1. 导入依赖

<!--导入jredis的包-->
<dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>3.2.0</version>
</dependency>
<!--fastjson-->
<dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.70</version>
</dependency>

编码测试

a.连接数据库
b.操作命令
c.断开连接

连接数据库

package com.softeem.kuang;import redis.clients.jedis.Jedis;/*** Created by 苍狼* Time on 2022-07-07*/
public class redisTest01 {public static void main(String[] args) {Jedis jedis = new Jedis("192.168.101.102", 6379);String ping = jedis.ping();System.out.println("测试结果为"+ping);System.out.println("jedis.get");}
}

String测试 

package com.softeem.kuang;import redis.clients.jedis.Jedis;import java.util.concurrent.TimeUnit;/*** Created by 苍狼* Time on 2022-07-08*/
public class TestString {public static void main(String[] args) {Jedis jedis = new Jedis("192.168.101.102", 6379);jedis.flushDB();System.out.println("===========增加数据===========");System.out.println(jedis.set("key1","value1"));System.out.println(jedis.set("key2","value2"));System.out.println(jedis.set("key3", "value3"));System.out.println("删除键key2:"+jedis.del("key2"));System.out.println("获取键key2:"+jedis.get("key2"));System.out.println("修改key1:"+jedis.set("key1", "value1Changed"));System.out.println("获取key1的值："+jedis.get("key1"));System.out.println("在key3后面加入值："+jedis.append("key3", "End"));System.out.println("key3的值："+jedis.get("key3"));System.out.println("增加多个键值对："+jedis.mset("key01","value01","key02","value02","key03","value03"));System.out.println("获取多个键值对："+jedis.mget("key01","key02","key03"));System.out.println("获取多个键值对："+jedis.mget("key01","key02","key03","key04"));System.out.println("删除多个键值对："+jedis.del("key01","key02"));System.out.println("获取多个键值对："+jedis.mget("key01","key02","key03"));jedis.flushDB();System.out.println("===========新增键值对防止覆盖原先值==============");System.out.println(jedis.setnx("key1", "value1"));System.out.println(jedis.setnx("key2", "value2"));System.out.println(jedis.setnx("key2", "value2-new"));System.out.println(jedis.get("key1"));System.out.println(jedis.get("key2"));System.out.println("===========新增键值对并设置有效时间=============");System.out.println(jedis.setex("key3", 2, "value3"));System.out.println(jedis.get("key3"));try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(jedis.get("key3"));System.out.println("===========获取原值，更新为新值==========");System.out.println(jedis.getSet("key2", "key2GetSet"));System.out.println(jedis.get("key2"));System.out.println("获得key2的值的字串："+jedis.getrange("key2", 2, 4));}
}

List测试

package com.softeem.kuang;import redis.clients.jedis.Jedis;/*** Created by 苍狼* Time on 2022-07-08*/
public class TestList {public static void main(String[] args) {Jedis jedis = new Jedis("192.168.101.102", 6379);jedis.flushDB();System.out.println("===========添加一个list===========");jedis.lpush("collections", "ArrayList", "Vector", "Stack", "HashMap", "WeakHashMap", "LinkedHashMap");jedis.lpush("collections", "HashSet");jedis.lpush("collections", "TreeSet");jedis.lpush("collections", "TreeMap");System.out.println("collections的内容："+jedis.lrange("collections", 0, -1));//-1代表倒数第一个元素，-2代表倒数第二个元素,end为-1表示查询全部System.out.println("collections区间0-3的元素："+jedis.lrange("collections",0,3));System.out.println("===============================");// 删除列表指定的值 ，第二个参数为删除的个数（有重复时），后add进去的值先被删，类似于出栈System.out.println("删除指定元素个数："+jedis.lrem("collections", 2, "HashMap"));System.out.println("collections的内容："+jedis.lrange("collections", 0, -1));System.out.println("删除下表0-3区间之外的元素："+jedis.ltrim("collections", 0, 3));System.out.println("collections的内容："+jedis.lrange("collections", 0, -1));System.out.println("collections列表出栈（左端）："+jedis.lpop("collections"));System.out.println("collections的内容："+jedis.lrange("collections", 0, -1));System.out.println("collections添加元素，从列表右端，与lpush相对应："+jedis.rpush("collections", "EnumMap"));System.out.println("collections的内容："+jedis.lrange("collections", 0, -1));System.out.println("collections列表出栈（右端）："+jedis.rpop("collections"));System.out.println("collections的内容："+jedis.lrange("collections", 0, -1));System.out.println("修改collections指定下标1的内容："+jedis.lset("collections", 1, "LinkedArrayList"));System.out.println("collections的内容："+jedis.lrange("collections", 0, -1));System.out.println("===============================");System.out.println("collections的长度："+jedis.llen("collections"));System.out.println("获取collections下标为2的元素："+jedis.lindex("collections", 2));System.out.println("===============================");jedis.lpush("sortedList", "3","6","2","0","7","4");System.out.println("sortedList排序前："+jedis.lrange("sortedList", 0, -1));System.out.println(jedis.sort("sortedList"));System.out.println("sortedList排序后："+jedis.lrange("sortedList", 0, -1));}
}

Hash测试

package com.softeem.kuang;import redis.clients.jedis.Jedis;import java.util.HashMap;
import java.util.Map;/*** Created by 苍狼* Time on 2022-07-08*/
public class TestHash {public static void main(String[] args) {Jedis jedis = new Jedis("192.168.101.102", 6379);jedis.flushDB();Map<String,String> map = new HashMap<String,String>();map.put("key1","value1");map.put("key2","value2");map.put("key3","value3");map.put("key4","value4");//添加名称为hash（key）的hash元素jedis.hmset("hash",map);//向名称为hash的hash中添加key为key5，value为value5元素jedis.hset("hash", "key5", "value5");System.out.println("散列hash的所有键值对为："+jedis.hgetAll("hash"));//return Map<String,String>System.out.println("散列hash的所有键为："+jedis.hkeys("hash"));//return Set<String>System.out.println("散列hash的所有值为："+jedis.hvals("hash"));//return List<String>System.out.println("将key6保存的值加上一个整数，如果key6不存在则添加key6："+jedis.hincrBy("hash", "key6", 6));System.out.println("散列hash的所有键值对为："+jedis.hgetAll("hash"));System.out.println("将key6保存的值加上一个整数，如果key6不存在则添加key6："+jedis.hincrBy("hash", "key6", 3));System.out.println("散列hash的所有键值对为："+jedis.hgetAll("hash"));System.out.println("删除一个或者多个键值对："+jedis.hdel("hash", "key2"));System.out.println("散列hash的所有键值对为："+jedis.hgetAll("hash"));System.out.println("散列hash中键值对的个数："+jedis.hlen("hash"));System.out.println("判断hash中是否存在key2："+jedis.hexists("hash","key2"));System.out.println("判断hash中是否存在key3："+jedis.hexists("hash","key3"));System.out.println("获取hash中的值："+jedis.hmget("hash","key3"));System.out.println("获取hash中的值："+jedis.hmget("hash","key3","key4"));}
}

事务

public class Test01 {public static void main(String[] args) {Jedis jedis = new Jedis("39.99.xxx.xx", 6379);JSONObject jsonObject = new JSONObject();jsonObject.put("hello", "world");jsonObject.put("name", "kuangshen");// 开启事务Transaction multi = jedis.multi();String result = jsonObject.toJSONString();// jedis.watch(result)try {multi.set("user1", result);multi.set("user2", result);// 执行事务multi.exec();}catch (Exception e){// 放弃事务multi.discard();} finally {// 关闭连接System.out.println(jedis.get("user1"));System.out.println(jedis.get("user2"));jedis.close();}}
}

七、SpringBoot整合

springboot操作数据库：spring-data jpa jdbc mongodb redis
springData也是和springboot其名的项目

1. 导入依赖

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

springboot 2.x后 ，原来使用的 Jedis 被 lettuce 替换。

jedis：采用的直连，多个线程操作的话，是不安全的。如果要避免不安全，使用jedis pool连接池！更像BIO模式

lettuce：采用netty，实例可以在多个线程中共享，不存在线程不安全的情况！可以减少线程数据了，更像NIO模式

我们在学习SpringBoot自动配置的原理时，整合一个组件并进行配置一定会有一个自动配置类xxxAutoConfiguration,并且在spring.factories中也一定能找到这个类的完全限定名。Redis也不例外。

ctrl+f 搜索redis 

那么就一定还存在一个RedisProperties类

    @ConditionalOnSingleCandidate(RedisConnectionFactory.class)//我们可以自己定义一个redisTemplate来替换这个默认的！public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {//默认的RedisTemplate没有过多的设置，redis对象都需要序列化！//两个泛型都是object类型，我们使用需要强制转换<String,Object>RedisTemplate<Object, Object> template = new RedisTemplate();template.setConnectionFactory(redisConnectionFactory);return template;}@Bean@ConditionalOnMissingBean@ConditionalOnSingleCandidate(RedisConnectionFactory.class)//由于String是Redis中最常使用的类型，所有单独提出来一个beanpublic StringRedisTemplate stringRedisTemplate(RedisConnectionFactory redisConnectionFactory) {StringRedisTemplate template = new StringRedisTemplate();template.setConnectionFactory(redisConnectionFactory);return template;}

之前我们说SpringBoot2.x后默认使用Lettuce来替换Jedis，现在我们就能来验证了。

先看Jedis:

@ConditionalOnClass({aaa.class, bbb.class, ccc.class}): 代表需要引入这三个类这个bean才会生效.而这里是没有GenericObjectPool.class, Jedis.class这两个类的(下面有波浪线)

然后再看Lettuce：

需要RedisClient.class类才生效, 满足条件, 完美生效. 

现在我们回到RedisAutoConfiguratio

只有两个简单的Bean

• RedisTemplate
• StringRedisTemplate

当看到xxTemplate时可以对比RestTemplate、SqlSessionTemplate,通过使用这些Template来间接操作组件。那么这俩也不会例外。分别用于操作Redis和Redis中的String数据类型。

在RedisTemplate上也有一个条件注解，说明我们是可以对其进行定制化的 

就是这个@ConditionalOnMissingBean注解

@ConditionalOnMissingBean 不满足条件时候进行组件的注入

例如在配置类中标入@ConditionalOnMissingBean(User.class)意为：组件中有User类，我们配置类中的下面的注入不会生效，当组件中没有User类时,我们配置类中的下面的注入才会生效 

所以在这里如果你不自己手写一个RedisTemplate则系统提供的这个RedisTemplate就会生效. 但是我们一般都会重新手写一个RedisTemplate类.

说完这些，我们需要知道如何编写配置文件然后连接Redis，就需要阅读RedisProperties

还有一些连接池相关的配置。注意使用时一定使用Lettuce的连接池。

        2. 编写配置文件

# 配置redis
spring.redis.host=39.99.xxx.xx
spring.redis.port=6379

        3.使用RedisTemplate

@SpringBootTest
class RedisSpringbootApplicationTests {@Autowiredprivate RedisTemplate redisTemplate;@Testvoid contextLoads() {// redisTemplate 操作不同的数据类型，api和我们的指令是一样的// opsForValue 操作字符串 类似String// opsForList 操作List 类似List// opsForHah// 除了基本的操作，我们常用的方法都可以直接通过redisTemplate操作，比如事务和基本的CRUD// 获取连接对象//RedisConnection connection = redisTemplate.getConnectionFactory().getConnection();//connection.flushDb();//connection.flushAll();redisTemplate.opsForValue().set("mykey","kuangshen");System.out.println(redisTemplate.opsForValue().get("mykey"));}
}

        4. 测试结果

此时我们回到Redis查看数据时候，惊奇发现全是乱码，可是程序中可以正常输出：

这时候就关系到存储对象的序列化问题，在网络中传输的对象也是一样需要序列化，否者就全是乱码。

我们转到看那个默认的RedisTemplate内部什么样子：

        5. 定制RedisTemplate的模板：

我们创建一个Bean加入容器，就会触发RedisTemplate上的条件注解使默认的RedisTemplate失效。

@Configuration
public class RedisConfig {//编写自己的redsiTemplate@Bean@SuppressWarnings("all")public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();template.setConnectionFactory(factory);// 序列化配置 解析任意对象Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);// json序列化利用ObjectMapper进行转义ObjectMapper om = new ObjectMapper();om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);jackson2JsonRedisSerializer.setObjectMapper(om);// 2.序列化String类型StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();// key采用String的序列化方式template.setKeySerializer(stringRedisSerializer);// hash的key也采用String的序列化方式template.setHashKeySerializer(stringRedisSerializer);// value序列化方式采用jacksontemplate.setValueSerializer(jackson2JsonRedisSerializer);// hash的value序列化方式采用jacksontemplate.setHashValueSerializer(jackson2JsonRedisSerializer);template.afterPropertiesSet();return template;}}

 这样一来，只要实体类进行了序列化，我们存什么都不会有乱码的担忧了。

八、自定义Redis工具类

使用RedisTemplate需要频繁调用.opForxxx然后才能进行对应的操作，这样使用起来代码效率低下，工作中一般不会这样使用，而是将这些常用的公共API抽取出来封装成为一个工具类，然后直接使用工具类来间接操作Redis,不但效率高并且易用。

这里是我封装的一个redis工具类

package com.softeem.springbootredis02.utils;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;//在我们真实的开发中, 或者你们在公司, 一般都可以看到一个公司自己封装redis的工具
@Component
public final class RedisUtil {@Autowiredprivate RedisTemplate<String, Object> redisTemplate;// =============================common============================/*** 指定缓存失效时间* @param key  键* @param time 时间(秒)*/public boolean expire(String key, long time) {try {if (time > 0) {redisTemplate.expire(key, time, TimeUnit.SECONDS);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 根据key 获取过期时间* @param key 键 不能为null* @return 时间(秒) 返回0代表为永久有效*/public long getExpire(String key) {return redisTemplate.getExpire(key, TimeUnit.SECONDS);}/*** 判断key是否存在* @param key 键* @return true 存在 false不存在*/public boolean hasKey(String key) {try {return redisTemplate.hasKey(key);} catch (Exception e) {e.printStackTrace();return false;}}/*** 删除缓存* @param key 可以传一个值 或多个*/
//    @SuppressWarnings("unchecked")public void del(String... key) {if (key != null && key.length > 0) {if (key.length == 1) {redisTemplate.delete(key[0]);} else {redisTemplate.delete((Collection<String>) CollectionUtils.arrayToList(key));}}}// ============================String=============================/*** 普通缓存获取* @param key 键* @return 值*/public Object get(String key) {return key == null ? null : redisTemplate.opsForValue().get(key);}/*** 普通缓存放入* @param key   键* @param value 值* @return true成功 false失败*/public boolean set(String key, Object value) {try {redisTemplate.opsForValue().set(key, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 普通缓存放入并设置时间* @param key   键* @param value 值* @param time  时间(秒) time要大于0 如果time小于等于0 将设置无限期* @return true成功 false 失败*/public boolean set(String key, Object value, long time) {try {if (time > 0) {redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);} else {set(key, value);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 递增* @param key   键* @param delta 要增加几(大于0)*/public long incr(String key, long delta) {if (delta < 0) {throw new RuntimeException("递增因子必须大于0");}return redisTemplate.opsForValue().increment(key, delta);}/*** 递减* @param key   键* @param delta 要减少几(小于0)*/public long decr(String key, long delta) {if (delta < 0) {throw new RuntimeException("递减因子必须大于0");}return redisTemplate.opsForValue().increment(key, -delta);}// ================================Map=================================/*** HashGet* @param key  键 不能为null* @param item 项 不能为null*/public Object hget(String key, String item) {return redisTemplate.opsForHash().get(key, item);}/*** 获取hashKey对应的所有键值* @param key 键* @return 对应的多个键值*/public Map<Object, Object> hmget(String key) {return redisTemplate.opsForHash().entries(key);}/*** HashSet* @param key 键* @param map 对应多个键值*/public boolean hmset(String key, Map<String, Object> map) {try {redisTemplate.opsForHash().putAll(key, map);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** HashSet 并设置时间* @param key  键* @param map  对应多个键值* @param time 时间(秒)* @return true成功 false失败*/public boolean hmset(String key, Map<String, Object> map, long time) {try {redisTemplate.opsForHash().putAll(key, map);if (time > 0) {expire(key, time);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 向一张hash表中放入数据,如果不存在将创建** @param key   键* @param item  项* @param value 值* @return true 成功 false失败*/public boolean hset(String key, String item, Object value) {try {redisTemplate.opsForHash().put(key, item, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 向一张hash表中放入数据,如果不存在将创建** @param key   键* @param item  项* @param value 值* @param time  时间(秒) 注意:如果已存在的hash表有时间,这里将会替换原有的时间* @return true 成功 false失败*/public boolean hset(String key, String item, Object value, long time) {try {redisTemplate.opsForHash().put(key, item, value);if (time > 0) {expire(key, time);}return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 删除hash表中的值** @param key  键 不能为null* @param item 项 可以使多个 不能为null*/public void hdel(String key, Object... item) {redisTemplate.opsForHash().delete(key, item);}/*** 判断hash表中是否有该项的值** @param key  键 不能为null* @param item 项 不能为null* @return true 存在 false不存在*/public boolean hHasKey(String key, String item) {return redisTemplate.opsForHash().hasKey(key, item);}/*** hash递增 如果不存在,就会创建一个 并把新增后的值返回** @param key  键* @param item 项* @param by   要增加几(大于0)*/public double hincr(String key, String item, double by) {return redisTemplate.opsForHash().increment(key, item, by);}/*** hash递减** @param key  键* @param item 项* @param by   要减少记(小于0)*/public double hdecr(String key, String item, double by) {return redisTemplate.opsForHash().increment(key, item, -by);}// ============================set=============================/*** 根据key获取Set中的所有值* @param key 键*/public Set<Object> sGet(String key) {try {return redisTemplate.opsForSet().members(key);} catch (Exception e) {e.printStackTrace();return null;}}/*** 根据value从一个set中查询,是否存在** @param key   键* @param value 值* @return true 存在 false不存在*/public boolean sHasKey(String key, Object value) {try {return redisTemplate.opsForSet().isMember(key, value);} catch (Exception e) {e.printStackTrace();return false;}}/*** 将数据放入set缓存** @param key    键* @param values 值 可以是多个* @return 成功个数*/public long sSet(String key, Object... values) {try {return redisTemplate.opsForSet().add(key, values);} catch (Exception e) {e.printStackTrace();return 0;}}/*** 将set数据放入缓存** @param key    键* @param time   时间(秒)* @param values 值 可以是多个* @return 成功个数*/public long sSetAndTime(String key, long time, Object... values) {try {Long count = redisTemplate.opsForSet().add(key, values);if (time > 0)expire(key, time);return count;} catch (Exception e) {e.printStackTrace();return 0;}}/*** 获取set缓存的长度** @param key 键*/public long sGetSetSize(String key) {try {return redisTemplate.opsForSet().size(key);} catch (Exception e) {e.printStackTrace();return 0;}}/*** 移除值为value的** @param key    键* @param values 值 可以是多个* @return 移除的个数*/public long setRemove(String key, Object... values) {try {Long count = redisTemplate.opsForSet().remove(key, values);return count;} catch (Exception e) {e.printStackTrace();return 0;}}// ===============================list=================================/*** 获取list缓存的内容** @param key   键* @param start 开始* @param end   结束 0 到 -1代表所有值*/public List<Object> lGet(String key, long start, long end) {try {return redisTemplate.opsForList().range(key, start, end);} catch (Exception e) {e.printStackTrace();return null;}}/*** 获取list缓存的长度** @param key 键*/public long lGetListSize(String key) {try {return redisTemplate.opsForList().size(key);} catch (Exception e) {e.printStackTrace();return 0;}}/*** 通过索引 获取list中的值** @param key   键* @param index 索引 index>=0时， 0 表头，1 第二个元素，依次类推；index<0时，-1，表尾，-2倒数第二个元素，依次类推*/public Object lGetIndex(String key, long index) {try {return redisTemplate.opsForList().index(key, index);} catch (Exception e) {e.printStackTrace();return null;}}/*** 将list放入缓存** @param key   键* @param value 值*/public boolean lSet(String key, Object value) {try {redisTemplate.opsForList().rightPush(key, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 将list放入缓存* @param key   键* @param value 值* @param time  时间(秒)*/public boolean lSet(String key, Object value, long time) {try {redisTemplate.opsForList().rightPush(key, value);if (time > 0)expire(key, time);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 将list放入缓存** @param key   键* @param value 值* @return*/public boolean lSet(String key, List<Object> value) {try {redisTemplate.opsForList().rightPushAll(key, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 将list放入缓存** @param key   键* @param value 值* @param time  时间(秒)* @return*/public boolean lSet(String key, List<Object> value, long time) {try {redisTemplate.opsForList().rightPushAll(key, value);if (time > 0)expire(key, time);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 根据索引修改list中的某条数据** @param key   键* @param index 索引* @param value 值* @return*/public boolean lUpdateIndex(String key, long index, Object value) {try {redisTemplate.opsForList().set(key, index, value);return true;} catch (Exception e) {e.printStackTrace();return false;}}/*** 移除N个值为value** @param key   键* @param count 移除多少个* @param value 值* @return 移除的个数*/public long lRemove(String key, long count, Object value) {try {Long remove = redisTemplate.opsForList().remove(key, count, value);return remove;} catch (Exception e) {e.printStackTrace();return 0;}}}

九、Redis.conf

单位

1. 配置文件unit单位对大小不敏感

包含

就是好比我们学习spring, include, import. 可以使用 include 组合多个配置问题

网络

下面这个bind 127.0.0.1 代表绑定的ip地址, 如果不把它注释掉代表只能在本机中访问. 

protected-mode: 是否受保护的(默认是开启的, 我这里改成了no)

port: 设置的端口号(redis默认都是6379)

通用配置(GENERAL)

daemonize yes: 以守护进程的方式去运行(默认是no, 要我们自己手动开启)

pidfile /var/run/redis_6379.pid: 如果以后台的方式运行, 我们就需要一个pid文件

loglevel notice: 日志的等级通知(有debug, verbose, notice, warning) 

logfile" ": 日志的文件名字

databases 16 : 数据库的数量, 默认是16个数据库

always-show-logo yes : 是否总是显示logo 

快照

在规定的时间内, 执行了多少次操作, 则会持久化到文件的.rdb , .aof 

redis是内存数据库, 如果没有持久化, 那么数据断电及失.

save 900 1 : 在900s内, 如果至少有一个key进行了修改, 我们即进行持久化操作

save 300 10 : 如果至少10key进行了修改, 我们即进行持久化操作

stop-writes-on-bgsave-error yes: 持久化如果出错, 是否需要继续工作

rdbcompression yes: 是否压缩rdb文件, 需要消耗一些cpu的资源

rdbchecksum yes: 保存rdb文件的时候进行错误的校验, 检查.

dir ./: rdb文件保存的目录.

REPLICATION 复制, 我们后面讲解主从复制 

SECURITY(安全)

可以在这里设置redis的密码， 默认是没有密码的。

当然我们一般也不会在文件中直接设置, 一般都是在控制台用命令来设置. 

127.0.0.1:6379> ping 
PONG
127.0.0.1:6379> config get requirepass   # 获取redis的密码， 默认是没有密码的
1) "requirepass"
2) ""
127.0.0.1:6379> config set requirepass "123456"  # 设置redis的密码
OK
127.0.0.1:6379> config get requirepass
1) "requirepass"
2) "123456"
127.0.0.1:6379> ping
PONG
127.0.0.1:6379> exit    # 退出重新登陆
[root@CentOS-7-64 bin]# redis-cli -p 6379
127.0.0.1:6379> ping    # 发现不能进行直接使用命令
(error) NOAUTH Authentication required.
127.0.0.1:6379> auth 123456   #d登录密码
OK
127.0.0.1:6379> ping   #密码登录成功， 命令都可以正常使用了。
PONG

限制客户端 (CLIENTS)

maxclients : 设置能连接连接上redis的最大客户端的数量

maxmemory <bytes> : redis配置最大的内存容量

maxmemory-policy noeviction： 内存到达上限之后的处理策略

1、volatile-lru：只对设置了过期时间的key进行删除（默认值）

2、allkeys-lru ： 删除lru算法的key

3、volatile-random：随机删除即将过期key

4、allkeys-random：随机删除

5、volatile-ttl ： 删除即将过期的

6、noeviction ： 永不过期，返回错误

 APPEND ONLY MODE(aof配置)

appendonly no : 默认是不开启aof模式的， 默认是使用rdb方式持久化的， 在大部分所有的情况下， rdb完全够用。

appendfilename "appendonly.aof" : 持久化的文件名字。

appendfsync always : 每次修改都会sync. 消耗性能

appendfsync everysec: 每秒执行一次 sync, 可能会丢失这1s的性能。

appendfsync no : 不执行sync， 这个时候操作系统自己同步数据

具体的配置， 我们在Redis持久化中去给大家详细讲解

十、持久化—RDB

RDB：Redis Databases

什么是RDB?

在指定时间间隔后，将内存中的数据集快照写入数据库 ；在恢复时候，直接读取快照文件，进行数据的恢复；

面试和工作的重点

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

原理是redis会单独创建（fork）一个与当前线程一模一样的子进程来进行持久化，这个子线程的所有数据（变量。环境变量，程序程序计数器等）都和原进程一模一样，会先将数据写入到一个临时文件中，待持久化结束了，再用这个临时文件替换上次持久化好的文件，整个过程中，主进程不进行任何的io操作，这就确保了极高的性能。

我们默认的就是RDB，一般情况下不需要修改这个配置
如果需要进行大规模数据恢复，且对于数据的完整性回复不是非常的敏感，那么RDB方式比AOF的方式更加高效。RDB的缺点是最后一次持久化的数据可能会丢失

rdb保存的文件是dump.rdb

搜索snapshot

删除dump.rdb文件

一分钟内设置5次key，自动生成dump.rdb文件 

触发机制

1. save的规则满足的情况下，会自动触发rdb原则
2. 执行flushall命令，也会触发我们的rdb原则
3. 退出redis，也会自动产生rdb文件

如何恢复rdb文件

1. 只需要将rdb文件放在redis启动目录就可以，redis启动时就会自动检测dump.rdb文件，恢复其中数据。
2. rdb文件存放的位置一般就放在 usr/local/bin目录下

优缺点

优点：

1. 适合大规模的数据恢复
2. 对数据的完整性要求不高

缺点：

1. 需要一定的时间间隔进行操作，如果redis意外宕机了，这个最后一次修改的数据就没有了。
2. fork进程的时候，会占用一定的内容空间。

十一、持久化—AOF

Append Only File

追加文件
将我们所有的命令都记录下来，恢复的时候将这个文件全部执行一遍

以日志的形式来记录每一个写操作，将Redis执行过的所有写指令记录下来（读操作不记录），只许追加文件不许改写文件，redis重启之处会会读取该文件重新构建数据，inother words，如果redis重启就会根据日志文件的内容将写指令从前到后执行一次以完成数据的恢复工作。

什么是AOF?

快照功能（RDB）并不是非常耐久（durable）： 如果 Redis 因为某些原因而造成故障停机， 那么服务器将丢失最近写入、以及未保存到快照中的那些数据。 从 1.1 版本开始， Redis 增加了一种完全耐久的持久化方式： AOF 持久化。

aof保存的文件是appendonly.aof

默认为no需要手动进行开启（系统默认采用的是rdb）

重启redis后读取.aof文件

如果这个aof文件有错位，这时候redis是启动不起来的，我需要修改这个aof文件

redis给我们提供了一个工具 redis-check-aof --fix

如果文件正常，重启就可以直接恢复了。

appendonly yes  # 默认是不开启aof模式的，默认是使用rdb方式持久化的，在大部分的情况下，rdb完全够用
appendfilename "appendonly.aof"# appendfsync always # 每次修改都会sync 消耗性能
appendfsync everysec # 每秒执行一次 sync 可能会丢失这一秒的数据
# appendfsync no # 不执行 sync ,这时候操作系统自己同步数据，速度最快

优点和缺点

优点

1. 每一次修改都会同步，文件的完整性会更加好
2. 没秒同步一次，可能会丢失一秒的数据
3. 从不同步，效率最高

缺点

1. 相对于数据文件来说，aof远远大于rdb，修复速度比rdb慢！
2. Aof运行效率也要比rdb慢，所以我们redis默认的配置就是rdb持久化

aof默认是文件无限追加，文件会越来越大

如果aof文件大于64m，fork一个新进程将我们的文件进行重写

注： 在主从复制当中， 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文件就可以了，代价一是带来了持续的lO，二是AOF rewrite的最后将rewrite过程中产生的新数据写到新文件造成的阻塞几乎是不可避免的。只要硬盘许可，应该尽量减少AOF rewrite的频率，AOF重写的基础大小默认值64M太小了，可以设到5G以上，默认超过原大小100%大小重写可以改到适当的数值。
• ·如果不Enable AOF，仅靠Master-Slave Repllcation实现高可用性也可以，能省掉一大笔IO，也减少了rewrite时带来的系统波动。代价是如果Master/Slave同时倒掉，会丢失十几分钟的数据，启动脚本也要比较Master/Slave 中的两个RDB文件，载入较新的那个. 微博就是这种架构。

十二、RDB和AOP选择 

RDB 和 AOF 对比

如何选择使用哪种持久化方式？

1. 一般来说， 如果想达到足以媲美 PostgreSQL 的数据安全性， 你应该同时使用两种持久化功能。
2. 如果你非常关心你的数据， 但仍然可以承受数分钟以内的数据丢失， 那么你可以只使用 RDB 持久化。
3. 有很多用户都只使用 AOF 持久化， 但并不推荐这种方式： 因为定时生成 RDB 快照（snapshot）非常便于进行数据库备份， 并且 RDB 恢复数据集的速度也要比 AOF 恢复的速度要快。

十三、Redis发布与订阅

redis发布订阅（pub/sub）是一种消息通信模式：发送者（pub）发送消息，订阅者（sub）接受消息。微信，微博，关注系统

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

订阅/发布消息图

第一个：消息发送者
第二个：频道
第三个：消息订阅者 

------------订阅端----------------------
127.0.0.1:6379> SUBSCRIBE sakura # 订阅sakura频道
Reading messages... (press Ctrl-C to quit) # 等待接收消息
1) "subscribe" # 订阅成功的消息
2) "sakura"
3) (integer) 1
1) "message" # 接收到来自sakura频道的消息 "hello world"
2) "sakura"
3) "hello world"
1) "message" # 接收到来自sakura频道的消息 "hello i am sakura"
2) "sakura"
3) "hello i am sakura"--------------消息发布端-------------------
127.0.0.1:6379> PUBLISH sakura "hello world" # 发布消息到sakura频道
(integer) 1
127.0.0.1:6379> PUBLISH sakura "hello i am sakura" # 发布消息
(integer) 1-----------------查看活跃的频道------------
127.0.0.1:6379> PUBSUB channels
1) "sakura"

原理

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. 多人在线聊天室。

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

十四、Redis主从复制

概念

主从复制，是指将一台Redis服务器的数据，复制到其他的Redis服务器。前者称为主节点（Master/Leader）,后者称为从节点（Slave/Follower）， 数据的复制是单向的！只能由主节点复制到从节点（主节点以写为主、从节点以读为主）。

默认情况下，每台Redis服务器都是主节点，一个主节点可以有0个或者多个从节点，但每个从节点只能由一个主节点。

作用

1. 数据冗余：主从复制实现了数据的热备份，是持久化之外的一种数据冗余的方式。
2. 故障恢复：当主节点故障时，从节点可以暂时替代主节点提供服务，是一种服务冗余的方式
3. 负载均衡：在主从复制的基础上，配合读写分离，由主节点进行写操作，从节点进行读操作，分担服务器的负载；尤其是在多读少写的场景下，通过多个从节点分担负载，提高并发量。
4. 高可用基石：主从复制还是哨兵和集群能够实施的基础。

为什么使用集群

1. 单台服务器难以负载大量的请求
2. 单台服务器故障率高，系统崩坏概率大
3. 单台服务器内存容量有限。

环境配置

我们在讲解配置文件的时候，注意到有一个replication模块 (见Redis.conf中第8条)

查看当前库的信息：info replication

127.0.0.1:6379> info replication
# Replication
role:master # 角色
connected_slaves:0 # 从机数量
master_replid:3b54deef5b7b7b7f7dd8acefa23be48879b4fcff
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

既然需要启动多个服务，就需要多个配置文件。每个配置文件对应修改以下信息：

• 端口号
• pid文件名
• 日志文件名
• rdb文件名

启动单机多服务集群：

一主二从配置 

默认情况下，每台Redis服务器都是主节点； 我们一般情况下只用配置从机就好了！

认老大！一主（79）二从（80，81）

使用SLAVEOF host port就可以为从机配置主机了。

然后主机上也能看到从机的状态：

我们这里是使用命令搭建，是暂时的， 真实开发中应该在从机的配置文件中进行配置， 这样的话是永久的。 

使用规则

        1. 从机只能读，不能写，主机可读可写但是多用于写。

127.0.0.1:6381> set name sakura # 从机6381写入失败
(error) READONLY You can't write against a read only replica.127.0.0.1:6380> set name sakura # 从机6380写入失败
(error) READONLY You can't write against a read only replica.127.0.0.1:6379> set name sakura
OK
127.0.0.1:6379> get name
"sakura"

测试: 主机断开连接, 从机依旧连接到主机的, 但是没有写操作, 这个时候, 主机要回来了, 从机依旧可以直接获取主机上写的信息.

如果使用命令行, 来配置的主从,这个时候如果重启了, 就会变回主机! 只要变回从机立马就会从主机中获取值!

复制原理

Slave启动成功连接到 master后会发送一个sync同步命令

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

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

增量复制:Master继续将新的所有收集到的修改命令依次传给slave，完成同步

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

主从复制的另外一种模式

 层层链路

这个时候也可以完成我们的主从复制

如果没有老大了, 这个时候能不能选择一个老大出来呢? 手动! 

谋权篡位

如果主机断开连接, 我们可以使用SLAVEOF no one, 让自己变成主机!  其它的节点就可以手动连接到最新的这个主节点(手动)! 如果这个时候老大修复了, 那就重新连接! 

总结: 

当从机断电宕机后，若不是使用配置文件配置的从机，再次启动后作为主机是无法获取之前主机的数据的，若此时重新配置称为从机，又可以获取到主机的所有数据。这里就要提到一个同步原理。

当主机断电宕机后，默认情况下从机的角色不会发生变化 ，集群中只是失去了写操作，当主机恢复以后，又会连接上从机恢复原状。

第二条中提到，默认情况下，主机故障后，不会出现新的主机，有两种方式可以产生新的主机：

从机手动执行命令slaveof no one,这样执行以后从机会独立出来成为一个主机
使用哨兵模式（自动选举）

十五、哨兵模式

主从切换的方法是：当主服务器宕机后，需要手动把一台服务器切换为主服务器，需要人工干预，费时费力，还会造成一段时间内服务器不可用。更多的时候我们考虑的是哨兵模式，redis从2.8开始正式提供了sentinel架构来解决这个问题。

​谋朝篡位的自动版，能够后台监控主机是否故障，如果故障根据投票数自动将从库转换为主库。

​哨兵模式是一种特殊的模式，首先redis提供了哨兵命令，哨兵是一个独立的进程，作为进程，他会独立运行，其原理是哨兵通过发送命令，等待redis服务器响应，从而监控运行的多个redis实例

哨兵的作用

• 通过发送命令，让redis返回其运行状态，包括主服务器和从服务器
• 当哨兵检测到master宕机，会自动将slave切换成master，然后通过发布订阅模式通知其他从服务器修改配置文件，让他们切换为主机

然而一个哨兵进程对redis服务器进行监控，可能会出现问题，为此，我们可以使用多个哨兵进行监控，各个哨兵相互监控，这样就形成了多哨兵模式。

假设主服务器宕机，哨兵1先检测到这个结果，系统并不会立马进行failover过程，仅仅是哨兵1主观的认为主服务器不可用，这个现象称为主观下线。当后面的哨兵也检测到主服务器不可用，并且达到一定数量，那么哨兵之间会进行一次投票，投票的结果由一个哨兵发起，进行failover（故障转移操作）。q切换成功后，通过发布订阅模式，让各个哨兵把自己的监控的从服务器实现切换主机，这个过称为客观下线

测试

我们目前的状态是一主二次!

1. 配置哨兵配置文件sentinel.conf 

# sentinel monitor 被监控的名称 host port 1

sentinel monitor myredis 127.0.0.1 6379 1 

后面的这个数字1, 代表主机挂了, slave投票看让谁接替成为主机, 票数最多的, 就会成为主机! (哨兵对该redis服务的投票数)

2. 启动哨兵

[root@CentOS-7-64 bin]# redis-sentinel config/sentinel.conf
6796:X 12 Jul 2022 10:30:38.997 # oO0OoO0OoO0Oo Redis is starting oO0OoO0OoO0Oo
6796:X 12 Jul 2022 10:30:38.997 # Redis version=7.0.2, bits=64, commit=00000000, modified=0, pid=6796, just started
6796:X 12 Jul 2022 10:30:38.997 # Configuration loaded
6796:X 12 Jul 2022 10:30:38.998 * Increased maximum number of open files to 10032 (it was originally set to 1024).
6796:X 12 Jul 2022 10:30:38.998 * monotonic clock: POSIX clock_gettime_._                                                  _.-``__ ''-._                                             _.-``    `.  `_.  ''-._           Redis 7.0.2 (00000000/0) 64 bit.-`` .-```.  ```\/    _.,_ ''-._                                  (    '      ,       .-`  | `,    )     Running in sentinel mode|`-._`-...-` __...-.``-._|'` _.-'|     Port: 26379|    `-._   `._    /     _.-'    |     PID: 6796`-._    `-._  `-./  _.-'    _.-'                                   |`-._`-._    `-.__.-'    _.-'_.-'|                                  |    `-._`-._        _.-'_.-'    |           https://redis.io       `-._    `-._`-.__.-'_.-'    _.-'                                   |`-._`-._    `-.__.-'    _.-'_.-'|                                  |    `-._`-._        _.-'_.-'    |                                  `-._    `-._`-.__.-'_.-'    _.-'                                   `-._    `-.__.-'    _.-'                                       `-._        _.-'                                           `-.__.-'                                               6796:X 12 Jul 2022 10:30:39.111 # WARNING: The TCP backlog setting of 511 cannot be enforced because /proc/sys/net/core/somaxconn is set to the lower value of 128.
6796:X 12 Jul 2022 10:30:39.120 * Sentinel new configuration saved on disk
6796:X 12 Jul 2022 10:30:39.120 # Sentinel ID is 74d47ee6e0c32a059c58f7225c18bac27a1d89b8
6796:X 12 Jul 2022 10:30:39.120 # +monitor master myredis 127.0.0.1 6379 quorum 1
6796:X 12 Jul 2022 10:31:09.035 # +sdown master myredis 127.0.0.1 6379
6796:X 12 Jul 2022 10:31:09.035 # +odown master myredis 127.0.0.1 6379 #quorum 1/1
6796:X 12 Jul 2022 10:31:09.035 # +new-epoch 1
6796:X 12 Jul 2022 10:31:09.035 # +try-failover master myredis 127.0.0.1 6379
6796:X 12 Jul 2022 10:31:09.038 * Sentinel new configuration saved on disk
6796:X 12 Jul 2022 10:31:09.038 # +vote-for-leader 74d47ee6e0c32a059c58f7225c18bac27a1d89b8 1
6796:X 12 Jul 2022 10:31:09.038 # +elected-leader master myredis 127.0.0.1 6379
6796:X 12 Jul 2022 10:31:09.038 # +failover-state-select-slave master myredis 127.0.0.1 6379
6796:X 12 Jul 2022 10:31:09.102 # -failover-abort-no-good-slave master myredis 127.0.0.1 6379
6796:X 12 Jul 2022 10:31:09.186 # Next failover delay: I will not start a failover before Tue Jul 12 10:37:09 2022
6796:X 12 Jul 2022 10:31:52.252 * +slave slave 127.0.0.1:6380 127.0.0.1 6380 @ myredis 127.0.0.1 6379
6796:X 12 Jul 2022 10:31:52.253 * Sentinel new configuration saved on disk
6796:X 12 Jul 2022 10:31:52.343 # -sdown master myredis 127.0.0.1 6379
6796:X 12 Jul 2022 10:31:52.343 # -odown master myredis 127.0.0.1 6379

如果Master节点断开, 这个时候就会从从机中随机选择一个服务器! (这里面有一个投票算法)

哨兵日志

如果主机此时回来了, 只能归并到新的主机下, 当做从机, 这就是哨兵模式的规则!  

哨兵模式

优点：

• 哨兵集群，基于主从复制模式，所有主从复制的优点，它都有
• 主从可以切换，故障可以转移，系统的可用性更好
• 哨兵模式是主从模式的升级，手动到自动，更加健壮

缺点：

• Redis不好在线扩容，集群容量一旦达到上限，在线扩容就十分麻烦
• 实现哨兵模式的配置其实是很麻烦的，里面有很多配置项
   

哨兵模式的全部配置 

#Example sentine1.conf#哨兵sentine1实例运行的端口默认26379
port 26379#哨兵sentine1的工作目录dir 
/tmp#哨兵sentinel监控的redis主节点的ip port
# master-name可以自己命名的主节点名字只能由字母A-z、数字0-9、这三个字符".-_"组成。
# quorum配置多少个sentine1哨兵统一认为master主节点失联那么这时客观上认为主节点失联了
# sentinel monitor <master-name> <ip> <redis-port> <quorum>sentine1 monitor mymaster 127.0.0.1 6379 2#当在Redis实例中开启了requirepass foobared 授权密码这样所有连接Redis实例的客户端都要提供密码
#设置哨兵sentinel连接主从的密码注意必须为主从设置一样的验证密码
# sentine1 auth-pass <master-name> <password> I
sentinel auth-pass mymaster MySUPER--secret-0123passwOrd#指定多少毫秒之后主节点没有应答哨兵sentinel此时哨兵主观上认为主节点下线默认30秒
# sentine1 down-after-milliseconds <master-name> <mi77iseconds>
sentinel down-after-mi77iseconds mymaster 30000#这个配置项指定了在发生failover主备切换时最多可以有多少个slave同时对新的master进行同步，这个数字越小，完成failover所需的时间就越长，
但是如果这个数字越大，就意味着越多的s1ave因为rep1ication而不可用。
可以通过将这个值设为1来保证每次只有一个s1ave 处于不能处理命令请求的状态。
#sentinel para1le7-syncs <master-name> <nums1aves>
sentine1 para1le1-syncs mymaster 1#故障转移的超时时间failover-timeout可以用在以下这些方面:
#1．同一个sentine1对同一个master两次failover之间的间隔时间。
#2．当一个slave从一个错误的master那里同步数据开始计算时间。直到slave被纠正为向正确的master那里同步数据时。
#3.当想要取消一个正在进行的failover所需要的时间。
#4.当进行failoveri时，配置所有slaves指向新的master所需的最大时间。不过，即使过了这个超时, slaves依然会被正确配置为指向master，但是就不按para7le1-syncs所配置的规则来了
#默认三分钟
#sentinel failover-timeout <master-name> <mi17iseconds>sentinel failover-timeout mymaster 180000# SCRIPTS EXECUTION#配置当某一事件发生时所需要执行的脚本，可以通过脚本来通知管理员，例如当系统运行不正常时发邮件通知相关人员。#对于脚本的运行结果有以下规则;
#若脚本执行后返回1，那么该脚本稍后将会被再次执行，重复次数目前默认为10
#若脚本执行后返回2，或者比2更高的一个返回值，脚本将不会重复执行。
#如P作本在执行过程中由于收到系统中断信号被终止了，则同返回值为1时的行为相同。
#个卿本的取火执行时间为60s，如果超过这个时间，脚本将会被一个SIGKILL信号终止，之后重新执行。#通知型脚本:当sentine7有任何警告级别的事件发生时(比如说redis实例的主观失效和客观失效等等)，将会去调用这个脚本，这时这个脚本应该通过邮件，SMS等方式去通知系统管理员关于系统不正常运行的信息。调用该脚本时，将传给脚本两个参数，一个是事件的类型，一个是事件的描述。如果sentine1.conf配置文件中配置了这个脚本路径，那么必须保证这个脚本存在于这个路径，并且是可执行的，否则sentine1无法正常启动成功。
#通知脚本
# sentinel notification-script <master-name> <script-path>sentine1 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(即旧的s1ave)通信的#这个脚本应该是通用的，能被多次调用，不是针对性的。
# sentinel client-reconfig-script <master-name> <script-path>sentinel client-reconfig-script mymaster /var/redis/reconfig.sh

十六、缓存穿透与雪崩

redis缓存穿透和雪崩（面试高频，工作常用）

缓存穿透

概念

缓存穿透是指查询一个一定不存在的数据，由于缓存是不命中时需要从数据库查询，查不到数据则不写入缓存，这将导致这个不存在的数据每次请求都要到数据库去查询，进而给数据库带来压力。

解决方案

布隆过滤器

对所有可能查询的参数以Hash的形式存储，以便快速确定是否存在这个值，在控制层先进行拦截校验，校验不通过直接打回，减轻了存储系统的压力。

缓存空对象

一次请求若在缓存和数据库中都没找到，就在缓存中方一个空对象用于处理后续这个请求。

这样做有一个缺陷：存储空对象也需要空间，大量的空对象会耗费一定的空间，存储效率并不高。解决这个缺陷的方式就是设置较短过期时间

即使对空值设置了过期时间，还是会存在缓存层和存储层的数据会有一段时间窗口的不一致，这对于需要保持一致性的业务会有影响。

缓存击穿（量太大，缓存过期）

概念

相较于缓存穿透，缓存击穿的目的性更强，一个存在的key，在缓存过期的一刻，同时有大量的请求，这些请求都会击穿到DB，造成瞬时DB请求量大、压力骤增。这就是缓存被击穿，只是针对其中某个key的缓存不可用而导致击穿，但是其他的key依然可以使用缓存响应。

比如热搜排行上，一个热点新闻被同时大量访问就可能导致缓存击穿。

解决方案

1. 设置热点数据永不过期

   这样就不会出现热点数据过期的情况，但是当Redis内存空间满的时候也会清理部分数据，而且此种方案会占用空间，一旦热点数据多了起来，就会占用部分空间。

2. 加互斥锁(分布式锁)

在访问key之前，采用SETNX（set if not exists）来设置另一个短期key来锁住当前key的访问，访问结束再删除该短期key。保证同时刻只有一个线程访问。这样对锁的要求就十分高。

缓存雪崩

概念 

大量的key设置了相同的过期时间，导致在缓存在同一时刻全部失效，造成瞬时DB请求量大、压力骤增，引起雪崩。

解决方案 

• redis高可用

这个思想的含义是，既然redis有可能挂掉，那我多增设几台redis，这样一台挂掉之后其他的还可以继续工作，其实就是搭建的集群

• 限流降级

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

• 数据预热

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