NoSQL概述

为什么要用Nosql

我们现在是处于大数据时代；

1，单机MySQL

在以前，一个网站的访问量一般不会太大，所以单个数据库就足够了。而且都是静态页面，服务器无压力。

这种情况下：整个网站的瓶颈是什么？

1. 数据量如果太大，一个机器放不下
2. 数据的索引（B+ Tree），一个机器内存放不下
3. 访问量（读写混合），一个服务器承受不了

出现以上三种情况之一，那么就必须晋级

2，Memcached（缓存）+ MySQL + 垂直分离（读写分离）

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

发展过程：优化数据结构和索引–》文件缓存（IO）–》Memcached（当时最热门的技术）

3，分库分表+水平拆分

本质：数据库（读，写）

早些年MyISAM：表锁，十分影响效率。对人 的要求也越来越高

转战到Innodb：行锁

慢慢使用分库分表来解决写的压力！

MySQL的集群，就是很好的解决了当时的需求。

到现在，MySQL等关系型数据库已经不够用了！数据很大，变化快

MySQL有的使用他来存取一些很大的文件，博客，图片！数据库表很大，效率低！所以要有一种专门处理这些数据的数据库

目前互联网加过的基本模型

为什么要用NoSQL

用户的个人信息，社交网络，地理位置，用户自己产生的数据，用户日志等等爆发式的增长

这个时候NoSQL就是一个很好的数据库，可以更好的处理以上情况

什么是NoSQL

NoSQL

Nosql = Not Only SQL

泛指非关系型数据库的，随着web2.0互联网的诞生！传统的关系型数据库很难对付web2.0时代！尤其是超大规模的高并发的社区！

就会暴露出很过难以克服的问题，NoSQL在当今数据库环境下发展的很快，Redis是发展最快的，而且是我们当下必须要去掌握的

很多的数据类型的存储不需要固定的格式，不需要多余的操作就可以横行扩展的！Map<Stirng,Object>使用键值对来控制！

Nosql

解耦

1. 方便扩展（数据之间没有关系，很好拓展）

2. 大数据量高性能（Redis一秒写8万次，读取11万次，NoSQL的缓存记录级，是一种细粒度的缓存，性能会比较高）

3. 数据类型是多样的！（不需要实现设计数据库！随取随用！如果是数据量很大的表，很多人就无法设计了！）

4. 传统RDBMS和NoSQL

   传统的RDBMS
   -结构化组织
   -SQL
   -数据和关系都存在单独的表中
   -操作对象，数据定义语言
   -严格的一致性
   -基础的事务服务
   -。。。
   1
   2
   3
   4
   5
   6
   7
   8

   Nosql
   —不仅仅是数据库
   -没有固定的查询语言
   -键值对存储，列存储，文档存储，图形数据库（社交关系）
   -最终一致性
   -CAP定理和BASE（异地多活）
   —高可用，高性能，高扩展
   1
   2
   3
   4
   5
   6
   7

了解：3V+3高

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

• 海量Volume
• 多样Variety
• 实时Velocity

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

• 高并发
• 高可扩（随时水平拆分）
• 高性能（保证用户体验）

真正在公司中的实践：NoSQL+RDBMS一起使用才是最强的，阿里巴巴的架构演进！

阿里巴巴演进分析

思考问题：就是这么多东西都是用一个数据库来存储的吗？

开源才是技术的王道！

任何一家互联网公司，都不可能只是简简单单让用户能用就好了。

架构师：没有什么是加一层解决不了的。

# 1,商品的基本信息
    名称，价格，商家信息：
    关系型数据库就可以解决了！MySql/Oracle  推荐文章：阿里云的这群疯子
    淘宝内部的MySQL 不是大家用的MySQL
    
# 2，商品的描述，评论（文字比较多的）
	文档型数据库中，MongoDB
	
# 3，图片
	分布式文件系统 FastDFS
	-淘宝自己的TFS
	-Gooale的 GFS
	-Hadoop   HDFS
	-阿里云的  oss
	
# 4，商品的关键字
	-搜索引擎 solr elasticsearch
	-ISerach：多隆 
	所有牛逼的人都有一段苦逼的岁月！但是你只要想SB一样去坚持，终将牛逼

# 5，商品热门的波段信息
	-内存数据库
	-Redis，Tair，Menache
	
# 6，商品的交易，外部支付接口
	-三方应用
	
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

大型互联网应用的问题：

• 数据类型太多了
• 数据源繁多，经常重构
• 数据要改造，大面积改造

解决问题：

NoSQL的四大分类

KV键值对：

• 新浪：Redis
• 美团：Redis+Tair
• 阿里，百度：Reids+memecache

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

• MongoDB（必须要掌握）
  • 是一个基于分布式文件存储的数据库，C++编写，主要是用来处理大量的文档！
  • 是一个介于关系型数据库和非关系型数据库中中间的产品！MongoDB是非关系型数剧库中功能最丰富，最像关系型数据库的！
• ConthDE

列存储数据库

• HBase
• 分布式文件系统

图关系数据库：

• 他不是存图形的，而是放的是关系，比如朋友圈社交网络，广告推荐
• Neo4j，InfiGrid

四者对比

1.Redis入门

概述

Redis是什么

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

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

redis会周期性的把更新的数据写入磁盘或者把修改操作写入追加的记录文件，并且在此基础上实现了master-slave(主从)同步。

人们称之为结构化数据库

Redis能干什么

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

特性

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

。。。

学习中需要用到的东西

1. 官网：https://redis.io/

2. 中文网：http://www.redis.cn/

3. 下载地址通过官网下载即可

注意：Windows在GitHub上下载（停更很久了）

Redis推荐都是在Linux服务器上搭建的，我们是基于Linux学习的！

WIndows下载

1. 下载安装包：https://github.com/MicrosoftArchive/redis/releases

2. 下载完毕得到安装包：

   

3. 解压，十分小

4. 开启Redis，双击运行即可

   

5. 使用redis客户端来连接redis

   [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-sa8MOnhv-1652690888850)(Redis.assets/image-20211030181020489.png)]

   Windows使用确实简单，但是Redis推荐在Linux下使用

   

Linux安装

1. 下载安装包

2. 解压Redis的安装包！储蓄一般放到opt包下

   mv redis-6.2.6.tar.gz /opt
   1

3. 解压文件

   tar -zxvf redis-6.2.6.tar.gz
   1

   

4. 进入解压后的文件可以看到他的配置文件
   

5. 基本的环境安装

   yum install gcc-c++
   make
   make
   make install
   1
   2
   3
   4

   [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-sXgo6kOa-1652690888851)(Redis.assets/image-20220311220031479.png)]

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

   

7. 将redis配置文件复制一份到我们的当前目录下

   mkdir myconfig
   cp /opt/redis-6.2.6/redis.conf myconfig
   1
   2

   

8. Redis默认不是后台启动的，我们要修改配置文件

   

9. 启动Redis服务

redis-server myconfig/redis.conf
# 在六点多的版本中没有启动提示
1
2

11. 使用Redis-cli

    redis-cli -p 6379 -a 自己的密码（如果是服务器，这里一定一定要设置密码，不然肯定被黑）
    1

    

12. 查看redis的进程是否被启用ps -ef|grep redis

    

13. 关闭Redis服务shotdown

    

性能测试

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

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

redis-benchmark命令参数！

我们来简单测试一下：

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

如何查看这些分析呢？

基础知识

redis默认有16个数据库

默认使用的是第0个数据库

可以使用select来切换数据库

127.0.0.1:6379> select 3 #切换数据库
OK
127.0.0.1:6379[3]> DBSIZE # 查看数据库大小
(integer) 0
127.0.0.1:6379[3]> set name zhao
OK
127.0.0.1:6379[3]> get name 
"zhao"
127.0.0.1:6379[3]> keys * # 查看所有的key
1) "name"
1
2
3
4
5
6
7
8
9
10

清除当前数据库flushdb

127.0.0.1:6379[3]> flushdb  # 清除当前数据库
OK
1
2

清除全部数据库FLUSHALL

127.0.0.1:6379[3]> FLUSHALL
OK
1
2

思考：为什么redis的端口号是6379！粉丝效应！（了解即可！）

Redis是单线程的！（+6版本之前，在6版本之后添加了多线程，为了解决I/O问题）

明白Redis是很快的，官方表示，Redis是基于内存操作的，CPU不是Redis的瓶颈，Redis的瓶颈是机器的内存和网络的带宽，既然可以用品那个单线程来表示就使用单线程了

Redis是C语言写的，官方提供的数据为100000+的QPS，这个不比Memecache差！

Redis为啥单线程还很快呢？

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

先去CPU>内存>硬盘的速度要有所了解！

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

2.五大数据类型

官网文档

全段翻译

Redis 是一种开源（BSD 许可）、内存中数据结构存储，用作数据库、缓存和消息代理。Redis 提供了诸如字符串、散列、列表、集合、带范围查询的排序集合、位图、超级日志、地理空间索引和流等数据结构。Redis 内置复制、Lua 脚本、LRU 驱逐、事务和不同级别的磁盘持久化，并通过 Redis Sentinel 和 Redis Cluster 自动分区提供高可用性。

现在讲的所有的命令都是十分重要的，后面我们使用SpringBoot，Jedis，所有的方法就是用这些命令！

单点登录

Redis-Key

127.0.0.1:6379> keys *  # 查看所有的key
1) "name"
127.0.0.1:6379> set age 1 # set key
OK
127.0.0.1:6379> EXISTS name  # 判断当前的key是否存在
(integer) 1
127.0.0.1:6379> EXISTS name1 # 1代表存在，0代表不存在
(integer) 0
127.0.0.1:6379> move name1 
(error) ERR wrong number of arguments for 'move' command
127.0.0.1:6379> move name 1  # 移动当前key到一号数据库
(integer) 1
127.0.0.1:6379> keys *
1) "age"
127.0.0.1:6379> set name zhao
OK
127.0.0.1:6379> keys *
1) "name"
2) "age"
127.0.0.1:6379> get name
"zhao"
127.0.0.1:6379> EXPIRE name 10  # 设置key的过期时间，单位是杪
(integer) 1
127.0.0.1:6379> ttl
(error) ERR wrong number of arguments for 'ttl' command
127.0.0.1:6379> ttl name # 查看当前key的剩余时间
(integer) 4
127.0.0.1:6379> ttl name # 如果是负数就是过期了
(integer) -2
127.0.0.1:6379> get name
(nil)
127.0.0.1:6379> type name  # 查看当前key的类型
string
127.0.0.1:6379> type age
string

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

后面要是有不会的命令可以去看官方的把帮助文档

总结

keys *查看当前库所有key    (匹配：keys *1)
exists key判断某个key是否存在
type key 查看你的key是什么类型
del key       删除指定的key数据
unlink key   根据value选择非阻塞删除
仅将keys从keyspace元数据中删除，真正的删除会在后续异步操作。
expire key 10   10秒钟：为给定的key设置过期时间
ttl key 查看还有多少秒过期，-1表示永不过期，-2表示已过期

select命令切换数据库
dbsize查看当前数据库的key的数量
flushdb清空当前库
flushall通杀全部库
1
2
3
4
5
6
7
8
9
10
11
12
13

String（字符串）

90%的程序员使用redis只会使用String类型。

127.0.0.1:6379> set key1 v1 # 设置值
OK
127.0.0.1:6379> get key1  # 获得值
"v1"
127.0.0.1:6379> EXISTS key1  # 判断是否存在该字段
(integer) 1
127.0.0.1:6379> APPEND key1 "hello"  # 追加字符串，如果当前key不存在，就相当于setkey
(integer) 7
127.0.0.1:6379> get key!
(nil)
127.0.0.1:6379> get key1
"v1hello"
127.0.0.1:6379> APPEND key1 ",zhao"
(integer) 12
127.0.0.1:6379> STRLEN key1  # 获取字符串长度
(integer) 12
127.0.0.1:6379> get key1
"v1hello,zhao"

################################################################
# i++
# 步长 i+=
127.0.0.1:6379> set views 0  # 设置初始量为0
OK
127.0.0.1:6379> get vieds
(nil)
127.0.0.1:6379> get views
"0"
127.0.0.1:6379> incr views  # 自增1  浏览器加一
(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  # 自减1 浏览量减一
(integer) 1
127.0.0.1:6379> decr views
(integer) 0
127.0.0.1:6379> get views
"0"
127.0.0.1:6379> INCRBY views 10 # 设置步长，指定增量
(integer) 10
127.0.0.1:6379> DECRBY views 10
(integer) 0

###########################################################################
# 字符串范围  range
127.0.0.1:6379> set key1 "hello,zhao" # 设置key1的值
OK
127.0.0.1:6379> GETRANGE key1 0 3 # 截取字符串【0,3】
"hell"
127.0.0.1:6379> GETRANGE key1 0 -1 # 获取全部字符串和get key是一样的
"hello,zhao"



# 替换
127.0.0.1:6379> get key2
"abdcerf"
127.0.0.1:6379> SETRANGE key2 1 xx  # 替换指定位置开始的字符串！
(integer) 7
127.0.0.1:6379> get key2
"axxcerf"
127.0.0.1:6379> GETRANGE name 2 5
"ngzb"
###############################################################################
# setex（set with expire） 设置过期时间
# setnx（set if not exist） 不存在在设置（在发布式锁中常常使用）
127.0.0.1:6379> setex key3 30 "hello" # 设置key3的值为hello 过期时间30
OK
127.0.0.1:6379> ttl key3
(integer) 23
127.0.0.1:6379> get key3
"hello"
127.0.0.1:6379> setnx mykey "redis" # 如果mykey 不存在，创建mykey
(integer) 1
127.0.0.1:6379> keys *\
(empty array)
127.0.0.1:6379> keys * # 我们发现key3没有了
1) "key2"
2) "key1"
3) "mykey"
127.0.0.1:6379> ttl key3 # 结果发现过期了
(integer) -2
127.0.0.1:6379> setnx mykey "MongoDB" # 如果mykey存在，创建失败
(integer) 0
127.0.0.1:6379> get mykey
"redis"
#################################################
mset
mget

127.0.0.1:6379> mset k1 v1 k2 v2 k3 v3 # 同时设置多个值
OK
127.0.0.1:6379> keys *
1) "k1"
2) "k2"
3) "k3"
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)


# 对象
set user:1 {name:zhangsan,age:3} # 设置一个user：1对象 值为json字符串来保存一个对象！

# 这里的key是一个巧妙的设计： user:{id}:{file},如此设计在Redis中是完全OK的（我觉得这样子设计会更容易操作对象的属性）
127.0.0.1:6379> mset user:1:name zhangsan user:1:age 12
OK
127.0.0.1:6379> mget user:1:name user:1:age
1) "zhangsan"
2) "12"

#######################################################
getset #先get后set
127.0.0.1:6379> getset db redis # 如果不存在值，则返回nil
(nil)
127.0.0.1:6379> get db
"redis"
127.0.0.1:6379> getset db  mongdb # 如果存在值，获取到原来的值，并设置新的值
"redis"
127.0.0.1:6379> get db
"mongdb"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128

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

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

总结

set   <key><value>添加键值对
	*NX：当数据库中key不存在时，可以将key-value添加数据库
	*XX：当数据库中key存在时，可以将key-value添加数据库，与NX参数互斥
	*EX：key的超时秒数
	*PX：key的超时毫秒数，与EX互斥

get   <key>查询对应键值
append  <key><value>将给定的<value> 追加到原值的末尾
strlen  <key>获得值的长度
setnx  <key><value>只有在 key 不存在时    设置 key 的值

incr  <key>
将 key 中储存的数字值增1
只能对数字值操作，如果为空，新增值为1
decr  <key>
将 key 中储存的数字值减1
只能对数字值操作，如果为空，新增值为-1
incrby / decrby  <key><步长>将 key 中储存的数字值增减。自定义步长。
mset  <key1><value1><key2><value2>  ..... 
同时设置一个或多个 key-value对  
mget  <key1><key2><key3> .....
同时获取一个或多个 value  
msetnx <key1><value1><key2><value2>  ..... 
同时设置一个或多个 key-value 对，当且仅当所有给定 key 都不存在。
getrange  <key><起始位置><结束位置>
获得值的范围，类似java中的substring，前包，后包
setrange  <key><起始位置><value>
用 <value>  覆写<key>所储存的字符串值，从<起始位置>开始(索引从0开始)。

setex  <key><过期时间><value>
设置键值的同时，设置过期时间，单位秒。
getset <key><value>
以新换旧，设置了新值同时获得旧值。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

List

基本的数据类型，列表

在redis里面。我们可以把list玩成栈，队列，阻塞队列！

所有的list命令都是以s开头的

redis不区分大小写

###################################################################
lpush左边插入
rpush右边插入
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 # 获取list中的值
1) "three"
2) "two"
3) "one"
127.0.0.1:6379> LRANGE list 0 1 # 通过区间获取具体的值
1) "three"
2) "two"
127.0.0.1:6379> RPUSH list foure # 将一个值或者多个值放在右边，他的左右是由L或者R控制的
(integer) 4
127.0.0.1:6379> lrange list 0 -1
1) "three"
2) "two"
3) "one"
4) "foure"
##################################################
LPOP 左边移除
RPOP 右边移除

127.0.0.1:6379> 
127.0.0.1:6379> lpop list # 移除list的第一个元素
"three"
127.0.0.1:6379> Rpop list # 移除list的第三个元素
"foure"
127.0.0.1:6379> lrange list 0 -1
1) "two"
##############################################################
Lindex 通过下标获取一个值
 
127.0.0.1:6379> lindex list 0 # 通过下标获取list中的某一个值！
"two"
127.0.0.1:6379> lindex list 1
"one"
###################################################################
Llen 返回列表的长度

127.0.0.1:6379> flushdb # 清空数据库
OK
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> llen list  # 返回列表的长度
(integer) 3
###################################################################
移除指定的值！
取关 uid

Lrem 从左边删除n个value(从左到右)

127.0.0.1:6379> lrange list 0 -1
1) "three"
2) "three"
3) "two"
127.0.0.1:6379> lrem list 1 three # 移除指定的值 list后的数字表示移除几个值
(integer) 1
127.0.0.1:6379> lrange list 0 -1
1) "three"
2) "two"
127.0.0.1:6379> Lpush list three
(integer) 3
127.0.0.1:6379> lrem list 2 three
(integer) 2
127.0.0.1:6379> lrange list 0 -1
1) "two"


####################################################################
trim修剪

127.0.0.1:6379> rpush mylist "hello2"
(integer) 2
127.0.0.1:6379> rpush mylist "hello3"
(integer) 3
127.0.0.1:6379> rpush mylist "hello"
(integer) 4
127.0.0.1:6379> ltrim mylist 1 2 # 通过下标截取指定的长度，这个list已经被改变了，截断了只剩下截取的元素
OK
127.0.0.1:6379> lrange mylist 0 -1
1) "hello2"
2) "hello3"
######################################################################
rpoplpush # 移除列表的最后一个元素并将它一移到新的列表中去
127.0.0.1:6379> rpush mylist "hello"
(integer) 1
127.0.0.1:6379> rpush mylist "hello1"
(integer) 2
127.0.0.1:6379> rpush mylist "hello12"
(integer) 3
127.0.0.1:6379> rpoplpush mylist myotherlist # 移除列表的最后一个元素并将它一移到新的列表中去
"hello12"
127.0.0.1:6379> lrange mylist 0 -1
1) "hello"
2) "hello1"
127.0.0.1:6379> lrange myotherlist 0 -1 # 查看目标列表中是否有该值
1) "hello12"
#############################################################################
lset 将列表中指定下标的值替换为另外一个值，更新操作
127.0.0.1:6379> exists list  # 先判断这个list是否存在
(integer) 0
127.0.0.1:6379> lset list 0 item # 如果不存在列表我们取更新就会报错
(error) ERR no such key
127.0.0.1:6379> lpush list value1
(integer) 1
127.0.0.1:6379> lrange list 0 0 
1) "value1"
127.0.0.1:6379> lset list 0 item # 如果存在，跟新当前下标的值，如果不存在也会报错
OK
127.0.0.1:6379> lrange list 0 0
1) "item"

######################################################################
linsert # 将某个具体的value插入到列表中的某个元素的前后
127.0.0.1:6379> linsert mylist befort "world"
(error) ERR wrong number of arguments for 'linsert' command
127.0.0.1:6379> linsert mylist befort "hello" "world" 
(error) ERR syntax error
127.0.0.1:6379> lrange mylist 0 -1
1) "hello"
2) "hello1"
127.0.0.1:6379> linsert mylist before "hello1" "hello2"
(integer) 3
127.0.0.1:6379> linsert mylist after "hello1" "hello2"
(integer) 4
127.0.0.1:6379> lrange mylist 0 -1
1) "hello"
2) "hello2"
3) "hello1"
4) "hello2"

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140

小结

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

消息排队！消息队列！（左进右拿） 栈（左进左拿）

·总结常见的命令

lpush/rpush  <key><value1><value2><value3> .... 从左边/右边插入一个或多个值。
lpop/rpop  <key>从左边/右边吐出一个值。值在键在，值光键亡。

rpoplpush  <key1><key2>从<key1>列表右边吐出一个值，插到<key2>列表左边。

lrange <key><start><stop>
按照索引下标获得元素(从左到右)
lrange mylist 0 -1   0左边第一个，-1右边第一个，（0-1表示获取所有）
lindex <key><index>按照索引下标获得元素(从左到右)
llen <key>获得列表长度 

linsert <key>  before <value><newvalue>在<value>的后面插入<newvalue>插入值
lrem <key><n><value>从左边删除n个value(从左到右)
lset<key><index><value>将列表key下标为index的值替换成value

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

Set（集合）

set中的值是不能重读的！

##########################################################
127.0.0.1:6379> sadd myset "hello" # set集合中添加元素
(integer) 1
127.0.0.1:6379> sadd myset "zhao"
(integer) 1
127.0.0.1:6379> sadd myset "lovezhao"
(integer) 1
127.0.0.1:6379> SMEMBERS myset # 查看set的所有值
1) "zhao"
2) "hello"
3) "lovezhao"
127.0.0.1:6379> SISMEMBER myset hello  # 判断某一个值是不是在set集合中
(integer) 1
127.0.0.1:6379> SISMEMBER myset zhao1
(integer) 0
##############################################################
127.0.0.1:6379> scard myset  # 获取set集合中元素的个数
(integer) 3
##############################################################
rem
127.0.0.1:6379> srem myset hello  # 移除set中的指定元素
(integer) 1
127.0.0.1:6379> scard myset
(integer) 2
127.0.0.1:6379> smembers myset
1) "zhao"
2) "lovezhao"
################################################################
set 无序不重复集合，抽随机
127.0.0.1:6379> SRANDMEMBER myset # 随机抽选出一个元素
"zhao" 
127.0.0.1:6379> SRANDMEMBER myset
"lovezhao"
127.0.0.1:6379> SRANDMEMBER myset 2 # 随机抽选出指定个数的元素
1) "zhao"
2) "lovezhao"
#####################################################################
移除指定的key，随机删除key

127.0.0.1:6379> spop myset # 随机删除一个指定的元素
"lovezhao"
127.0.0.1:6379> SMEMBERS myset
1) "zhao"
######################################################################
将一个指定的值，移动到另外一个set集合中！
127.0.0.1:6379> sadd myset "hello"
(integer) 1
127.0.0.1:6379> sadd myset "world"
(integer) 1
127.0.0.1:6379> sadd myset "zhao"
(integer) 1
127.0.0.1:6379> sadd myset "set2"
(integer) 1
127.0.0.1:6379> asdd myset2 "set2"
(error) ERR unknown command `asdd`, with args beginning with: `myset2`, `set2`, 
127.0.0.1:6379> sadd myset2 "set2"
(integer) 1
127.0.0.1:6379> smove myset myset2 "zhao"  # 将一个指定的值，移动到另外一个set集合中
(integer) 1
127.0.0.1:6379> SMEMBERS myset2
1) "set2"
2) "zhao"
127.0.0.1:6379> SMEMBERS myset
1) "set2"
2) "world"
3) "hello"
############################################################################
微博，B站，共同关注（交集）
数字集合类：	
-差集
-交集
-并集
127.0.0.1:6379> sadd key1 a
(integer) 1
127.0.0.1:6379> sadd key1 b
(integer) 1
127.0.0.1:6379> sadd key1 c
(integer) 1
127.0.0.1:6379> sadd key2 c
(integer) 1
127.0.0.1:6379> sadd key2 d
(integer) 1
127.0.0.1:6379> sadd key2 e
(integer) 1
127.0.0.1:6379> SDIFF key1 key2 # 差集 difficult
1) "b"
2) "a"
127.0.0.1:6379> SINTER key1 key2 # 交集 intersect
1) "c"
127.0.0.1:6379> SUNION key1 key2 # 并集 union：联合
1) "e"
2) "a"
3) "c"
4) "d"
5) "b"

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

微博，A用户将所有关注得人放在一个set集合中，将它的粉丝也放在一个集合中！

共同关注，共同爱好，二度好友

总结常用的命令

lpush/rpush  <key><value1><value2><value3> .... 从左边/右边插入一个或多个值。
lpop/rpop  <key>从左边/右边吐出一个值。值在键在，值光键亡。

rpoplpush  <key1><key2>从<key1>列表右边吐出一个值，插到<key2>列表左边。

lrange <key><start><stop>
按照索引下标获得元素(从左到右)
lrange mylist 0 -1   0左边第一个，-1右边第一个，（0-1表示获取所有）
lindex <key><index>按照索引下标获得元素(从左到右)
llen <key>获得列表长度 

linsert <key>  before <value><newvalue>在<value>的后面插入<newvalue>插入值
lrem <key><n><value>从左边删除n个value(从左到右)
lset<key><index><value>将列表key下标为index的值替换成value
lpush/rpush  <key><value1><value2><value3> .... 从左边/右边插入一个或多个值。
lpop/rpop  <key>从左边/右边吐出一个值。值在键在，值光键亡。

rpoplpush  <key1><key2>从<key1>列表右边吐出一个值，插到<key2>列表左边。

lrange <key><start><stop>
按照索引下标获得元素(从左到右)
lrange mylist 0 -1   0左边第一个，-1右边第一个，（0-1表示获取所有）
lindex <key><index>按照索引下标获得元素(从左到右)
llen <key>获得列表长度 

linsert <key>  before <value><newvalue>在<value>的后面插入<newvalue>插入值
lrem <key><n><value>从左边删除n个value(从左到右)
lset<key><index><value>将列表key下标为index的值替换成value

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

Hash（哈希）

Map集合，key-map！这时候这个值是一个map集合！

就是一个map里面套map,本质和String类型是没有太大区别的，还是一个简单的key - value！

127.0.0.1:6379> hset myhash filed1 zhao # set一个具体的key-value
(integer) 1
127.0.0.1:6379> hget myhash filed1 # 获取一个字段值
"zhao"
127.0.0.1:6379> hmset myhash filed1 hello filed2 world # 获取多个key-value
OK
127.0.0.1:6379> hmget myhash filed1 filed2 # 得到多个字段值
1) "hello"
2) "world"
127.0.0.1:6379> hgetall myhash # 获取全部数据
1) "filed1"
2) "hello"
3) "filed2"
4) "world"
127.0.0.1:6379> hdel myhash filed1 # 删除hash指定的key字段！对应的value值也就没有了
(integer) 1
127.0.0.1:6379> hgetall myhash
1) "filed2"
2) "world"
127.0.0.1:6
##########################################################################
hlen

127.0.0.1:6379> hlen myhash # 获取哈希的字段数量
(integer) 1
127.0.0.1:6379> hset myhash filed1 world
(integer) 1
127.0.0.1:6379> hlen myhash
(integer) 2
##########################################################################
HEXISTS

127.0.0.1:6379> HEXISTS myhash filed3 # 判断哈希中指定字段是否存在
(integer) 0
##########################################################################
只获取所有的key
只获取所有的value

127.0.0.1:6379> hkeys myhash
1) "filed2"
2) "filed1"
127.0.0.1:6379> HVALS myhash
1) "world"
2) "world"
##########################################################################
incr decr
 
127.0.0.1:6379> hset myhash field3 5 
(integer) 1
127.0.0.1:6379> HINCRBY myhash field3 1 # 指定增量
(integer) 6
127.0.0.1:6379> HINCRBY myhash field3 -1
(integer) 5
127.0.0.1:6379> HSETNX myhash field4 hello # 如果不存在则可以使用
(integer) 1
127.0.0.1:6379> HSETNX myhash field4 world #如果存在则不能设置
(integer) 0

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

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

总结常用命令

hset <key><field><value>给<key>集合中的  <field>键赋值<value>
hget <key1><field>从<key1>集合<field>取出 value 
hmset <key1><field1><value1><field2><value2>... 批量设置hash的值
hexists<key1><field>查看哈希表 key 中，给定域 field 是否存在。 
hkeys <key>列出该hash集合的所有field
hvals <key>列出该hash集合的所有value
hincrby <key><field><increment>为哈希表 key 中的域 field 的值加上增量 1   -1
hsetnx <key><field><value>将哈希表 key 中的域 field 的值设置为 value ，当且仅当域 field 不存在 .

1
2
3
4
5
6
7
8
9

Zset（有序集合）

在set的基础上增加了一个值，set k1 v1 zset k1 score1 v1

127.0.0.1:6379> zadd myset 1 one # 添加一个值也可以添加多个值 签名的数字是用于排序时确定优先级的
(integer) 1
127.0.0.1:6379> zadd myset 2 two
(integer) 1
127.0.0.1:6379> zadd myset 3 three
(integer) 1
127.0.0.1:6379> ZRANGE myset 0 -1 # 遍历所有元素
1) "one"
2) "two"
3) "three"
##########################################################################
排序
ZRANGEBYSCORE 拆分为 zrange查询所有的集合 by通过 score 

127.0.0.1:6379> zadd salary 2000 xiaoming # 添加三个用户
(integer) 1
127.0.0.1:6379> zadd salary 5000 zhangsan
(integer) 1
127.0.0.1:6379> zadd salary 1 xueshuai
(integer) 1
# 他这里第一个为最小值，第二个为最大值
127.0.0.1:6379> ZRANGEBYSCORE salary -inf +inf # 显示全部用户 从小到大
1) "xueshuai"
2) "xiaoming"
3) "zhangsan"
127.0.0.1:6379> ZRANGEBYSCORE salary -inf +inf withscores # 显示全部用户，并且附带成绩
1) "xueshuai"
2) "1"
3) "xiaoming"
4) "2000"
5) "zhangsan"
6) "5000"
127.0.0.1:6379> ZRANGEBYSCORE salary -inf 2500 withscores # 显示工资小与两千五的员工，升序
1) "xueshuai"
2) "1"
3) "xiaoming"
4) "2000"
127.0.0.1:6379> ZREVRANGEBYSCORE salary +inf -inf withscores # 降序是这个，剩下的操作同上
1) "zhangsan"
2) "5000"
3) "xiaoming"
4) "2000"
5) "xueshuai"
6) "1"
##########################################################################
移除rem中的元素
127.0.0.1:6379> ZRANGE salary 0 -1
1) "xueshuai"
2) "xiaoming"
3) "zhangsan" 
127.0.0.1:6379> zrem salary xiaoming # 移除有序集合中指定元素
(integer) 1
127.0.0.1:6379> zrange salary 0 -1
1) "xueshuai"
2) "zhangsan"
##########################################################################
获取有序集合中的个数
127.0.0.1:6379> ZCARD salary # 获取有序集合中的个数
(integer) 2
##########################################################################
127.0.0.1:6379> zadd myset 1 hello
(integer) 1
127.0.0.1:6379> zadd myset 2 world 3 zhao
(integer) 2
127.0.0.1:6379> ZCOUNT myset 1 3 # 获取指定区间的成员数量！闭区间
(integer) 3
127.0.0.1:6379> ZCOUNT myset 1 2
(integer) 2

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

其余的api，剩下的去官方文档去查

案例思路：set 排序 存储班级成绩！

普通消息 1，重要消息 2 带权重进行判断

排行榜应用实现，取top n

总结常用的命令

zadd  <key><score1><value1><score2><value2>…
将一个或多个 member 元素及其 score 值加入到有序集 key 当中。
zrange <key><start><stop>  [WITHSCORES]   
返回有序集 key 中，下标在<start><stop>之间的元素，带WITHSCORES，可以让分数一起和值返回到结果集。
zrangebyscore key minmax [withscores] [limit offset count]
返回有序集 key 中，所有 score 值介于 min 和 max 之间(包括等于 min 或 max )的成员。有序集成员按 score 值递增(从小到大)次序排列。 
zrevrangebyscore key maxmin [withscores] [limit offset count]               
同上，改为从大到小排列。 
zadd  <key><score1><value1><score2><value2>…
将一个或多个 member 元素及其 score 值加入到有序集 key 当中。
zrange <key><start><stop>  [WITHSCORES]   
返回有序集 key 中，下标在<start><stop>之间的元素
带WITHSCORES，可以让分数一起和值返回到结果集。
zrangebyscore key minmax [withscores] [limit offset count]
返回有序集 key 中，所有 score 值介于 min 和 max 之间(包括等于 min 或 max )的成员。有序集成员按 score 值递增(从小到大)次序排列。 
zrevrangebyscore key maxmin [withscores] [limit offset count]               
同上，改为从大到小排列。 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

3.三种特殊数据类型

geospatoal地理位置

朋友的单位，附近的人？

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

查询测试数据：http://www.jsons.cn/lngcode/

只有6个命令

getadd

# getadd 添加地理位置
# 规则：两极是无法直接添加的，我们一般会通过下载城市数据，直接通过java程序一次性导入
# 参数key 值 （经度、纬度、名称）
# 有效经度为 -180 到 180 度。
# 有效纬度是从 -85.05112878 到 85.05112878 度。
127.0.0.1:6379> geoadd china:city 121.472644 31.231706 shanghai
(integer) 1
127.0.0.1:6379> geoadd china:city 112.549248 37.857014 taiyuan 114.085947 22.547 shenzhen
(integer) 2
127.0.0.1:6379> geoadd china:city 120.153576 30.287459 hangzhou 125.14904 42.927 xian
(integer) 2

1
2
3
4
5
6
7
8
9
10
11
12

getpos

127.0.0.1:6379> GEOPOS china:city beijing # 获取指定城市的经度纬度
1) 1) "116.40528291463851929"
   2) "39.9049884229125027"
1
2
3

getdist

两人之间的距离

单位：

• m为米。
• km为公里。
• mi数英里。
• ft为英尺。

127.0.0.1:6379> GEODIST china:city beijing taiyuan # 查询两个城市的直线距离
"404111.9580"
127.0.0.1:6379> GEODIST china:city beijing taiyuan km
"404.1120"
1
2
3
4

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

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

127.0.0.1:6379> GEORADIUS china:city 110 30 1000 km # 以110 30为中心，半径为1000km范围所包含的所有的城市
1) "shenzhen"
2) "hangzhou"
3) "taiyuan"
127.0.0.1:6379> GEORADIUS china:city 110 30 1000 km withdist# 以110 30为中心，半径为1000km范围所包含的所有的城市，加上距离
1) 1) "shenzhen"
   2) "923.4929"
2) 1) "hangzhou"
   2) "976.8197"
3) 1) "taiyuan"
   2) "904.9063"
127.0.0.1:6379> GEORADIUS china:city 110 30 1000 km withdist withcoord count 1 # 以110 30为中心，半径为1000km范围所包含的所有的城市，加上距离，经纬度和允许查询的个数
1) 1) "taiyuan"
   2) "904.9063"
   3) 1) "112.54924803972244263"
      2) "37.85701483724372451"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

GEORADIUSBYMEMBER

# 用于找出指定元素周围的其他元素！
127.0.0.1:6379> GEORADIUSBYMEMBER china:city beijing 1000 km
1) "taiyuan"
2) "beijing"
3) "xian"
1
2
3
4
5

• GEOHASH

该命令返回 11 个字符的 Geohash 字符串

# 将二维的经纬度转化为一维的字符串，如果俩字符串越接近，距离越近
127.0.0.1:6379> GEOHASH china:city beijing taiyuan 
1) "wx4g0b7xrt0"
2) "ww8p35ev5e0"
1
2
3
4

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

127.0.0.1:6379> zrange china:city 0 -1 # 查看地图中全部的元素
1) "shenzhen"
2) "hangzhou"
3) "shanghai"
4) "taiyuan"
5) "beijing"
6) "xian"
127.0.0.1:6379> zrem china:city beijing # 移除指定的地理位置
(integer) 1
127.0.0.1:6379> zrange china:city 0 -1
1) "shenzhen"
2) "hangzhou"
3) "shanghai"
4) "taiyuan"
5) "xian"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

Hyperloglog

什么是基数

A{1,2,4,5,6,7,7}

B{1,4,6,5,8}

基数（就是统计一批数据中不重复的元素）= 7，可以接受误差

简介

Redis2.8.9版本更新了Hyperloglog数据结构！

Redis Hyperloglog基数统计的算法

优点：占用的内存是固定的，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 a b c d e f l z x c
(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) 13

1
2
3
4
5
6
7
8
9
10
11
12
13

如果允许容错，那么一定可以使用Hyperloglog！

如果不允许容错，那么就是用set或者自己定义的结构

Bitmap

位存储

统计用户信息，活跃，不活跃！登录，未登录！打卡，未打卡！两个状态的都可以使用Bitmaps！

Bitmap位图，数据结构！都是操作二进制来记录，就只有0和1两个状态！

365天 = 365bit 1字节 = 8bit 46个字节左右！

测试

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

周一：1 周二：0

127.0.0.1:6379> setbit sign 0 1 # 第一个是位置 第二个是状态
(integer) 0
127.0.0.1:6379> setbit sign 1 0
(integer) 0
127.0.0.1:6379> setbit sign 2 1
(integer) 0
127.0.0.1:6379> setbit sign 3 1
(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 0
(integer) 0

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

查看某天的打卡情况

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

统计操作：打卡的天数

127.0.0.1:6379> BITCOUNT sign  # 统计这周的发卡记录
(integer) 5
1
2

4.Redis.conf详解

启动的时候，通过配置文件来启动的

行家有没有，出手就知道了

单位

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

包含

就是好比我们学习Spring，improt，include

bind 127.0.0.1 -::1  # 绑定的ip
protected-mode yes # 保护模式
port 6379 # 端口设置
1
2
3

通用GENERAL

daemonize yes # 以守护进程的方式运行，默认是no，我们需他自己开启

pidfile /var/run/redis_6379.pid # 如果是以后台方式运行，我们就需要指定一个pid文件

# 日志
# Specify the server verbosity level.
# This can be one of:
# debug (a lot of information, useful for development/testing)
# verbose (many rarely useful info, but not a mess like the debug level)
# notice (moderately verbose, what you want in production probably) 生产环境
# warning (only very important / critical messages are logged)
loglevel notice 

logfile "" # 日志文件位置名

databases 16 # 数据库的个数

always-show-logo no # 是否总是显示logo

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

快照 SNAPSHOTTING

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

redis是内存数据库，如果没有持久化，那么数据就是断电即失的！

# 如果900s内，如果至少有一个1 key进行了修改，我们及进入持久化操作
save 3600 1
# 如果300s内，如果至少有一个100 key进行了修改，我们及进入持久化操作
save 300 100
# 如果60s内，如果至少有一个10000 key进行了修改，我们及进入持久化操作
save 60 10000
# 之后我们会自己定义这个测试

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

rdbcompression yes # 是否压缩rdb文件，需要消耗一些cpu

rdbchecksum yes # 保存rdb文件，进行错误的检查校验！

dir ./ # rdb文件保存的目录
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

REPLICATION 复制，后面主从复制的时候在写

SECURITY 安全

config get requirepass  #获取密码
config set requirepass "123456"  #设置密码
auth "123456" # 登录
1
2
3

限制CLIENTS

maxclients 10000 # 是指能连接上redis的最大客户端的数量

maxmemory <bytes> # 设置最大容量
 
maxmemory-policy noeviction # 内存到达上限之后的处理策略
1、volatile-lru：只对设置了过期时间的key进行LRU（默认值） 

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

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

4、allkeys-random：随机删除   

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

6、noeviction ： 永不过期，返回错
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

APPEND ONLY MODE模式 aof配置

appendonly no # 默认是不开启aof模式的，默认是使用的rdb方式持久化的，在大部分情况下，rdb是足够用的

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

# appendfsync always # ，每次修改都会sync 消耗性能
appendfsync everysec # 美妙执行一次sync，可能会丢失1s的数据
# appendfsync no     # 不执行sync 这个时候操作系统自己同步数据，速度最快

1
2
3
4
5
6
7
8

5.Redis发布订阅

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

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

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

订阅/发布消息图：

命令

这些命令被广泛用于构建即时通信应用，比如网络聊天室（chatroom）和实时广播哦，实时提醒等

测试

订阅端：

127.0.0.1:6379> SUBSCRIBE zhao # 订阅一个频道zhao
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "zhao"
3) (integer) 1
# 等待读取推送信息
1) "message" # 消息
2) "zhao" # 哪个频道的消息
3) "hello world" # 具体内容
1
2
3
4
5
6
7
8
9

发送端：

127.0.0.1:6379> PUBLISH zhao "hello world" # 发布者发布消息到频道
(integer) 1
1
2

原理

Redis是使用C实现的,通过分析Redis源码里的pubsub.c文件,了解发布和订阅机制的底层实现,籍此加深对Redis的理解。
Redis通过PUBLISH、SUBSCRIBE 和PSUBSCRIBE等命令实现发布和订阅功能。

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

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

Pub/Sub从字面.上理解就是发布( Publish)与订阅( Subscribe ) , 在Redis中,你可以设定对某一个key值进行消息发布及消息订
阅，当一个key值上进行了消息发布后,所有订阅它的客户端都会收到相应的消息。这一功能最明显的用法就是 用作实时消息系
统,比如普通的即时聊天,群聊等功能。

使用场景:

• 实时消息系统！
• 实时聊天系统（相当于相互关注，将信息回显给所有的人）
• 订阅，关注系统都是可以的！

稍微复杂一点的可以用中间插件来做MQ

6.Jedis

我们要用Java来操纵Redis

什么是Jedis是Redis官方推荐的java连接工具，使用Java操作的Redis中间件！如果你要使用Java操作redis，那么一定要对Jedis十分熟悉

• 导入对应的依赖

  <!--导入Jedis的包-->
      <dependencies>
          <!-- https://mvnrepository.com/artifact/redis.clients/jedis -->
          <dependency>
              <groupId>redis.clients</groupId>
              <artifactId>jedis</artifactId>
              <version>3.2.0</version>
          </dependency>
          <dependency>
              <groupId>com.alibaba</groupId>
              <artifactId>fastjson</artifactId>
              <version>1.2.62</version>
          </dependency>
      </dependencies>
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14

• 编码测试：

  • 连接数据库
  • 操作命令
  • 断开连接

  public class TestPing {
      public static void main(String[] args) {
          //创建一个Jedis对象
          Jedis jedis = new Jedis("127.0.0.1",6379);
          //有密码的输入这个
        	jedis.auth("你的密码");
          //Jedis 所有的命令就是我们之前学习的所有指令！
          System.out.println(jedis.ping());
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10

  输出：

  

常用的API

String

List

Set

Hash

Zset

所有的api命令，就是我们对应上面学习的指令，一个都没有变化！

写一个案例来体验一下

这里呢只实现后端的逻辑代码

package com.ming.redis;

import redis.clients.jedis.Jedis;

import java.util.Random;

/**
 * @version v1.0
 * @ProjectName: jedis_redis
 * @ClassName: PhoneCode
 * @Description: 请描述该类的功能
 * @Author: ming
 * @Date: 2022/5/17 0:45
 */
public class PhoneCode {
    public static void main(String[] args) {
        //模拟验证码发送
        verifyCode("13678765435");

        //模拟验证码校验
        //getRedisCode("13678765435","4444");
    }

    //3 验证码校验
    public static void getRedisCode(String phone,String code) {
        //从redis获取验证码
        Jedis jedis = new Jedis("47.98.128.119",6379);
        //有密码的输入这个
        jedis.auth("*****");
        //验证码key
        String codeKey = "VerifyCode"+phone+":code";
        String redisCode = jedis.get(codeKey);
        //判断
        if(redisCode.equals(code)) {
            System.out.println("成功");
        }else {
            System.out.println("失败");
        }
        jedis.close();
    }

    //2 每个手机每天只能发送三次，验证码放到redis中，设置过期时间120
    public static void verifyCode(String phone) {
        //连接redis
        Jedis jedis = new Jedis("47.98.128.119",6379);
        //有密码的输入这个
        jedis.auth("*****");

        //拼接key
        //手机发送次数key
        String countKey = "VerifyCode"+phone+":count";
        //验证码key
        String codeKey = "VerifyCode"+phone+":code";

        //每个手机每天只能发送三次
        String count = jedis.get(countKey);
        if(count == null) {
            //没有发送次数，第一次发送
            //设置发送次数是1
            jedis.setex(countKey,24*60*60,"1");
        } else if(Integer.parseInt(count)<=2) {
            //发送次数+1
            jedis.incr(countKey);
        } else if(Integer.parseInt(count)>2) {
            //发送三次，不能再发送
            System.out.println("今天发送次数已经超过三次");
            jedis.close();
        }

        //发送验证码放到redis里面
        String vcode = getCode();
        jedis.setex(codeKey,120,vcode);
        jedis.close();
    }

    //1 生成6位数字验证码
    public static String getCode() {
        Random random = new Random();
        String code = "";
        for(int i=0;i<6;i++) {
            int rand = random.nextInt(10);
            code += rand;
        }
        return code;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

7.SpringBoot整合

SpringBoot操作数据层：spring-data jpa jdbc mongdb redis

SpringData也是和SpringBoot齐名的项目

说明：在SpringBoot2.X的版本以后，原来使用的Jedis被替换成了lettuce

<dependencies>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter</artifactId>
    <version>2.5.6</version>
    <scope>compile</scope>
  </dependency>
  <dependency>
    <groupId>org.springframework.data</groupId>
    <artifactId>spring-data-redis</artifactId>
    <version>2.5.6</version>
    <scope>compile</scope>
  </dependency>
  <dependency>
    <groupId>io.lettuce</groupId>
    <artifactId>lettuce-core</artifactId>
    <version>6.1.5.RELEASE</version>
    <scope>compile</scope>
  </dependency>
</dependencies>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

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

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

源码分析

@Bean
@ConditionalOnMissingBean(name = "redisTemplate") // 我们可以自己定义一个redisTemplate来替换这个默认值
@ConditionalOnSingleCandidate(RedisConnectionFactory.class)
public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
   //默认的RedisTempleate没有过多的设置，redis对象都是需要序列化的
   //两个泛型都是Object的类型，后面我们使用是需要强转的<String,Object>
   RedisTemplate<Object, Object> template = new RedisTemplate<>();
   template.setConnectionFactory(redisConnectionFactory);
   return template;
}

@Bean
@ConditionalOnMissingBean //由于String类型是我们最常使用的类型4所以直接使用就行了
@ConditionalOnSingleCandidate(RedisConnectionFactory.class)
public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory redisConnectionFactory) {
   StringRedisTemplate template = new StringRedisTemplate();
   template.setConnectionFactory(redisConnectionFactory);
   return template;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

测试类

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.cache.CacheProperties;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisTemplate;

@SpringBootTest
class Redis02SpringbootApplicationTests {

    @Autowired
    private RedisTemplate redisTemplate;
    @Test
    void contextLoads() {
        //redisTemplate 操作不同的数据类型
        //opsForValue 操作字符串
        //opsForList 操作list
        //opsForGeo
        //opsForSet
        //opsForHash
        //opsForZset
        //opsForHyperLogLog

        //除了基本的操作都可以直接通过redisTemplate操作，比如事务和基本的CRUD
        //获取连接对象
//        RedisConnection connection = redisTemplate.getConnectionFactory().getConnection();
//        connection.flushDb();
//        connection.flushAll();
        redisTemplate.opsForValue().set("mykey","zhao");
        System.out.println(redisTemplate.opsForValue().get("mykey"));
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

关于对象的保存

我们自己编写一个redisTemplate模板

//编写我们自己的redisTemplate
//这个是一个固定模板
@Bean
@SuppressWarnings("all") //作用：告诉编译器忽略指定的警告，不用在编译完成后出现警告信息。
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory)
        throws UnknownHostException {
    RedisTemplate<String, Object> template = new RedisTemplate<>();
    template.setConnectionFactory(redisConnectionFactory);

    //json序列化配置
    Jackson2JsonRedisSerializer<Object> objectJackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(Object.class);
    ObjectMapper objectMapper = new ObjectMapper();
    objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
    objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
    objectJackson2JsonRedisSerializer.setObjectMapper(objectMapper);

    //String的序列化
    StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
    //String的序列化
    template.setKeySerializer(stringRedisSerializer);
    //hash的序列化
    template.setHashKeySerializer(stringRedisSerializer);
    //value的序列化
    template.setValueSerializer(objectJackson2JsonRedisSerializer);
    //HashValue的序列化
    template.setHashValueSerializer(objectJackson2JsonRedisSerializer);
    template.afterPropertiesSet();

    return template;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

redis工具类模板

package com.zhao.util;

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;
@Component
public class RedisUtils {
    @Autowired
    private 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;
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499

所有的Redis操作其实对于java开发人员来说十分简单，重要的是去理解redis的思想和每一种数据结构的用处

8.事务

要么同时成功，要么同时失败，原子性！

Redis单体命令式保证原子性的，但是事务不保证原子性！

Redis本质：一组命令的集合！一个事务中的所有命令都会被序列化，在事务执行过程中，会按照顺序执行！

一次性，顺序性，排他性！执行一些命令！

--------队列 set set set 执行--------
1

Redis事务没有隔离级别的概念！

所有的命令在事务中，并没有直接被执行！只有发起执行的命令的时候才会被执行！Exec

Redis事务：

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

正常执行事务

127.0.0.1:6379> multi # 开启事务
OK
# 命令入队
127.0.0.1:6379(TX)> set k1 v1
QUEUED
127.0.0.1:6379(TX)> set k2 v2
QUEUED
127.0.0.1:6379(TX)> get k2
QUEUED
127.0.0.1:6379(TX)> set k3 v3
QUEUED
127.0.0.1:6379(TX)> EXEC # 执行事务
1) OK
2) OK
3) "v2"
4) OK
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

放弃事务

127.0.0.1:6379> MULTI # 开启事务
OK
127.0.0.1:6379(TX)> set k1 v1
QUEUED
127.0.0.1:6379(TX)> set k2 v2
QUEUED
127.0.0.1:6379(TX)> set k3 v3
QUEUED
127.0.0.1:6379(TX)> DISCARD # 取消事务
OK
127.0.0.1:6379> get k4 # 事务中的队列都不会被执行
(nil)
1
2
3
4
5
6
7
8
9
10
11
12

编译型异常（代码有错！命令有错！），事务中所有的命令都不会执行！

127.0.0.1:6379> MULTI
OK
127.0.0.1:6379(TX)> set k1 v1 
QUEUED
127.0.0.1:6379(TX)> getset k2
(error) ERR wrong number of arguments for 'getset' command
127.0.0.1:6379(TX)> set k3 v3
QUEUED
127.0.0.1:6379(TX)> exec # 执行事务报错
(error) EXECABORT Transaction discarded because of previous errors.
127.0.0.1:6379> get k3
(nil)
127.0.0.1:6379> get k1
(nil)
1
2
3
4
5
6
7
8
9
10
11
12
13
14

命令错误，所有的操作都失效

运行时异常（1/0），如果事务队列中存在语法性错误，其他命令是可以正常执行的，错误命令会抛出异常

127.0.0.1:6379> MULTI
OK
127.0.0.1:6379(TX)> set k1 "v1"
QUEUED
127.0.0.1:6379(TX)> exec
1) OK
127.0.0.1:6379> get k1
"v1"
127.0.0.1:6379> MULTI
OK
127.0.0.1:6379(TX)> incr k1
QUEUED
127.0.0.1:6379(TX)> set k2 v2
QUEUED
127.0.0.1:6379(TX)> set k3 v3
QUEUED
127.0.0.1:6379(TX)> exec
1) (error) ERR value is not an integer or out of range # 虽然第一条命令报错了但是依旧正常执行成功，所有事务不保证原子性
2) OK
3) OK
127.0.0.1:6379> get k2
"v2"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

总结：

只要不再事务内报错，后面的命令就可以继续执行

监控！Watch（面试常问）

悲观锁：

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

乐观锁：

• 很乐观，认为什么时候都没问题，无论什么时候都不会加锁！更新数据的时候判断一下，再次期间是否有人修改过这个数据
• 获取version
• 更新的时候比较version

Redis监事测试

正常执行成功！

127.0.0.1:6379> set money 100
OK
127.0.0.1:6379> set out 0
OK
127.0.0.1:6379> watch money # 监视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 out 20
QUEUED
127.0.0.1:6379(TX)> exec
1) (integer) 80
2) (integer) 20
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

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

127.0.0.1:6379> watch money # 监视money
OK
127.0.0.1:6379> MULTI
OK
127.0.0.1:6379(TX)> DECRBY money 10
QUEUED
127.0.0.1:6379(TX)> INCRBY out 10
QUEUED
127.0.0.1:6379(TX)> exec # 执行之前，，另外一个线程修改了我们的值，这个时候导致事务失败
(nil)

1
2
3
4
5
6
7
8
9
10
11

如果修改失败获取最新的数据即可

WATCH key [key …]
在执行multi之前，先执行watch key1 [key2],可以监视一个(或多个) key ，如果在事务执行之前这个(或这些) key 被其他命令所改动，那么事务将被打断。

unwatch
取消 WATCH 命令对所有 key 的监视。
如果在执行 WATCH 命令之后，EXEC 命令或DISCARD 命令先被执行了的话，那么就不需要再执行UNWATCH 了。
,

小结

Reids事务的三大特性

• 单独的隔离操作
  • 事务中的所有命令都会序列化，按照顺序地执行。事务再执行过程中，不会被其他客户端发送来的命令请求打断。
• 没有隔离级别的概念
  • 队列中命令没有提交前都不会实际被执行，因为事务提交如何指令都不会被实际执行
• 不保证原子性
  • 事务中如果由一条命令执行失败，其后的命令仍然会被执行，没有回滚。

9.Redis_事务_秒杀案例

9.1.解决计数器和人员记录的事务操作

代码实现：

JedisPoolUtil

package com.atguigu;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

public class JedisPoolUtil {
	private static volatile JedisPool jedisPool = null;

	private JedisPoolUtil() {
	}

	public static JedisPool getJedisPoolInstance() {
		if (null == jedisPool) {
			synchronized (JedisPoolUtil.class) {
				if (null == jedisPool) {
					JedisPoolConfig poolConfig = new JedisPoolConfig();
					poolConfig.setMaxTotal(200);
					poolConfig.setMaxIdle(32);
					poolConfig.setMaxWaitMillis(100*1000);
					poolConfig.setBlockWhenExhausted(true);
					poolConfig.setTestOnBorrow(true);  // ping  PONG
				 
					jedisPool = new JedisPool(poolConfig, "192.168.44.168", 6379, 60000 );
				}
			}
		}
		return jedisPool;
	}

	public static void release(JedisPool jedisPool, Jedis jedis) {
		if (null != jedis) {
			jedisPool.returnResource(jedis);
		}
	}

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

SecKill_redis（核心代码）

package com.atguigu;

import java.io.IOException;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.slf4j.LoggerFactory;

import ch.qos.logback.core.rolling.helper.IntegerTokenConverter;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.ShardedJedisPool;
import redis.clients.jedis.Transaction;

/**
 *
 */
public class SecKill_redis {

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

	//秒杀过程
	public static boolean doSecKill(String uid,String prodid) throws IOException {
		//1 uid和prodid非空判断
		if(uid == null || prodid == null) {
			return false;
		}

		//2 连接redis
		//Jedis jedis = new Jedis("192.168.44.168",6379);
		//通过连接池得到jedis对象 解决连接超时的问题
		JedisPool jedisPoolInstance = JedisPoolUtil.getJedisPoolInstance();
		Jedis jedis = jedisPoolInstance.getResource();

		//3 拼接key
		// 3.1 库存key
		String kcKey = "sk:"+prodid+":qt";
		// 3.2 秒杀成功用户key
		String userKey = "sk:"+prodid+":user";

		//监视库存  通过乐观锁解决超卖问题
		jedis.watch(kcKey);

		//4 获取库存，如果库存null，秒杀还没有开始
		String kc = jedis.get(kcKey);
		if(kc == null) {
			System.out.println("秒杀还没有开始，请等待");
			jedis.close();
			return false;
		}

		// 5 判断用户是否重复秒杀操作
		if(jedis.sismember(userKey, uid)) {
			System.out.println("已经秒杀成功了，不能重复秒杀");
			jedis.close();
			return false;
		}

		//6 判断如果商品数量，库存数量小于1，秒杀结束
		if(Integer.parseInt(kc)<=0) {
			System.out.println("秒杀已经结束了");
			jedis.close();
			return false;
		}

		//7 秒杀过程
		//使用事务
		Transaction multi = jedis.multi();

		//组队操作
		multi.decr(kcKey);
		multi.sadd(userKey,uid);

		//执行
		List<Object> results = multi.exec();

		if(results == null || results.size()==0) {
			System.out.println("秒杀失败了....");
			jedis.close();
			return false;
		}

		//7.1 库存-1
		//jedis.decr(kcKey);
		//7.2 把秒杀成功用户添加清单里面
		//jedis.sadd(userKey,uid);

		System.out.println("秒杀成功了..");
		jedis.close();
		return true;
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

SecKill_redisByScript（为了解决因乐观锁产生商品存在但是不能卖的情况，从而用LUA脚本语言实现）

package com.atguigu;

import java.io.IOException;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.slf4j.LoggerFactory;

import ch.qos.logback.core.joran.conditional.ElseAction;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.ShardedJedisPool;
import redis.clients.jedis.Transaction;

public class SecKill_redisByScript {
	
	private static final  org.slf4j.Logger logger =LoggerFactory.getLogger(SecKill_redisByScript.class) ;

	public static void main(String[] args) {
		JedisPool jedispool =  JedisPoolUtil.getJedisPoolInstance();
 
		Jedis jedis=jedispool.getResource();
		System.out.println(jedis.ping());
		
		Set<HostAndPort> set=new HashSet<HostAndPort>();

	//	doSecKill("201","sk:0101");
	}
	
	static String secKillScript ="local userid=KEYS[1];\r\n" + 
			"local prodid=KEYS[2];\r\n" + 
			"local qtkey='sk:'..prodid..\":qt\";\r\n" + 
			"local usersKey='sk:'..prodid..\":usr\";\r\n" + 
			"local userExists=redis.call(\"sismember\",usersKey,userid);\r\n" + 
			"if tonumber(userExists)==1 then \r\n" + 
			"   return 2;\r\n" + 
			"end\r\n" + 
			"local num= redis.call(\"get\" ,qtkey);\r\n" + 
			"if tonumber(num)<=0 then \r\n" + 
			"   return 0;\r\n" + 
			"else \r\n" + 
			"   redis.call(\"decr\",qtkey);\r\n" + 
			"   redis.call(\"sadd\",usersKey,userid);\r\n" + 
			"end\r\n" + 
			"return 1" ;
			 
	static String secKillScript2 = 
			"local userExists=redis.call(\"sismember\",\"{sk}:0101:usr\",userid);\r\n" +
			" return 1";

	public static boolean doSecKill(String uid,String prodid) throws IOException {

		JedisPool jedispool =  JedisPoolUtil.getJedisPoolInstance();
		Jedis jedis=jedispool.getResource();

		 //String sha1=  .secKillScript;
		String sha1=  jedis.scriptLoad(secKillScript);
		Object result= jedis.evalsha(sha1, 2, uid,prodid);

		  String reString=String.valueOf(result);
		if ("0".equals( reString )  ) {
			System.err.println("已抢空！！");
		}else if("1".equals( reString )  )  {
			System.out.println("抢购成功！！！！");
		}else if("2".equals( reString )  )  {
			System.err.println("该用户已抢过！！");
		}else{
			System.err.println("抢购异常！！");
		}
		jedis.close();
		return true;
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

SecKillServlet

public class SecKillServlet extends HttpServlet {
	private static final long serialVersionUID = 1L;

    public SecKillServlet() {
        super();
    }

	protected void doPost(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {

		String userid = new Random().nextInt(50000) +"" ;
		String prodid =request.getParameter("prodid");
		
		//boolean isSuccess=SecKill_redis.doSecKill(userid,prodid);
		boolean isSuccess= SecKill_redisByScript.doSecKill(userid,prodid);
		response.getWriter().print(isSuccess);
	}

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

index.jsp

<%@ page language="java" contentType="text/html; charset=UTF-8"
    pageEncoding="UTF-8"%>
<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
<html>
<head>
<meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
<title>Insert title here</title>
</head>
<body>
<h1>iPhone 13 Pro !!!  1元秒杀！！！
</h1>


<form id="msform" action="${pageContext.request.contextPath}/doseckill" enctype="application/x-www-form-urlencoded">
	<input type="hidden" id="prodid" name="prodid" value="0101">
	<input type="button"  id="miaosha_btn" name="seckill_btn" value="秒杀点我"/>
</form>

</body>
<script  type="text/javascript" src="${pageContext.request.contextPath}/script/jquery/jquery-3.1.0.js"></script>
<script  type="text/javascript">
$(function(){
	$("#miaosha_btn").click(function(){	 
		var url=$("#msform").attr("action");
	     $.post(url,$("#msform").serialize(),function(data){
     		if(data=="false"){
    			alert("抢光了" );
    			$("#miaosha_btn").attr("disabled",true);
    		}
		} );    
	})
})
</script>
</html>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

10.Redis持久化（重点）

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

RDB（Redis DataBase）

什么是RDB

在主从复制中，rdb就是备用的！从机上面不占用内存

在指定的时间间隔将内存中的数据集快照写入磁盘，也就是行话讲的Snapshot快照，他恢复时是将快照文件直接读到内存里。
Redis会单独创建（fork）一个子进程进行持久化，会先将数据写入一个临时文件中，待持久化过程都结束了，再用这个临时文件替换上次持久化好的文件。整个过程中，主进程是不进行然后IO操作的。这就确保了极高的性能。如果需要进行大规模数据的恢复，且对于数据恢复的完整性不是非常敏感，那么RDB方式要比AOF方式更高效。RDB的缺点是最后一次持久化后的数据可能会丢失。我们默认的就是RDB，一般情况下不需要修改这个配置
有时候在生产环境我们会对这个文件进行备份

rdb保存的文件是dump.rdbn

触发机制

1. save的规则满足的时候，会自动触发rdb规则
2. 执行flushall命令也会触发rdb规则
3. 退出redis，也会生成一个rdb文件

备份就会自动生成一个dump.rdb文件

任何恢复rdb文件！

1. 只要将rdb文件放入我们redis启动目录就可以，redis在启动的时候就会自动检查dump.rdb恢复其中的数据！
2. 查看需要存放的位置

几乎就是他自己默认的配置就够用了

优点：

• 适合大规模的数据恢复！dump.rdb
• 对数据完整性要求不高

缺点：

• 需要一定的时间间隔进程操作！如果redis意外宕机，这个最后一次修改数据就没有了
• fork进程·会占用一定的内存空间

AOF（Append Only File）

将我们所有命令都记录下来，类似于history，恢复的时候就会把这个文件全部在执行一遍！

是什么

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

AOF保存的是appendonly.aof文件

append

默认是不开启的，我们需要手动去配置！我们只需要将appendonly改为yes即可开启aof

重启redis就可以生效了！
如果这个aof文件有错位，这个时候redis是启动不起来的，我们需要对这个文件进行修复

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

如果这个文件可以启动的话就是就证明了文件恢复了

重写规则

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

如果aof文件大于64吗，太大了！就会fork一个新的进程将我们的文件重写

优点和缺点

appendonly no # 默认是不开启aof模式的，默认是使用的rdb方式持久化的，在大部分情况下，rdb是足够用的

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

# appendfsync always # ，每次修改都会sync 消耗性能
appendfsync everysec # 美妙执行一次sync，可能会丢失1s的数据
# appendfsync no     # 不执行sync 这个时候操作系统自己同步数据，速度最快
# rewrite 重写
1
2
3
4
5
6
7
8

优点：

• 每一次修改都是同步的，文件的完整性会更好！
• 每一秒同步一次，可能会丢失一秒的数据
• 从不同步，效率最高

缺点：

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

扩展：

• RDB持久化方式能够在指定的时间间隔内对你的数据进行快照存储
• AOF持久化记录方式每次对服务器写的操作，当服务器重启的时候就会重新执行这些命令来恢复原始数据，AOF命令以Redis协议追加保存每次写的擦欧总到文件末尾，Redis还能对文件进行后台重写，是得AOF文件的体积不至于过大
• 只做缓存,如果你只希望你的数据在服务器运行的时候存在,你也可以不使用任何持久化
• 同时开启两种持久化方式
  • 在这种情况下,当redis重启的时候会优先载入AOF文件来恢复原始的数据,因为在通常情况下AOF文件保存的数据集要比RDB文件保存的数据集要完整。
  • RDB的数据不实时，同时使用两者时服务器重启也只会找AOF文件,那要不要只使用AOF呢?作者建议不要,因为RDB更适合用于备份数据库( AOF在不断变化不好备份) , 快速重启,而且不会有AOF可能潜在的Bug ,留着作为-一个万- -的手段。
• 性能建议
  • 因为RDB文件只用作后备用途,建议只在Slave,上持久化RDB文件,而且只要15分钟备份一 次就够了,只保留save 900 1这条规则。
  • 如果Enable AOF ,好处是在最恶劣情况下也只会丢失不超过两秒数据,启动脚本较简单只load自己的AOF文件就可以了,代价一是带来了持续的I0 ,二是AOF rewrite的最后将rewrite过程中产生的新数据写到新文件造成的阻塞几乎是不可避免的。只要硬盘许可,应该尽量减少AOF rewrite的频率, AOF重写的基础大小默认值64M太小了,可以设到5G以上,默认超过原大小100%大小重写可以改到适当的数值。
  • 如果不Enable AOF ,仅靠Master-Slave Repllcation（主从复制）实现高可用性也可以,能省掉一 大笔I0 ,也减少了rewrite时带来的系统波动。代价是如果Master/Slave 同时倒掉,会丢失十几分钟的数据,启动脚本也要比较两个Master/Slave 中的RDB文件,载入较新的那个,微博就是这种架构。

11.Redis主从复制

概念

主从复制,是指将一台Redis服务器的数据,复制到其他的Redis服务器。前者称为主节点(master/leader) ,后者称为从节点(slave/follower) ;数据的复制是单向的,只能由主节点到从节点。Master以写为主 , Slave以读为主。

默认情况下,每台Redis服务器都是主节点;且一个主节点可以有多个从节点(或没有从节点) ,但-一个从节点只能有一一个主节点。

主从复制的作用主要包括:
1、数据冗余:主从复制实现了数据的热备份,是持久化之外的一种数据冗余方式。

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

3、负载均衡:在主从复制的基础上,配合读写分离,可以由主节点提供写服务,由从节点提供读服务(即写Redis数据时应用连接主节点,读Redis数据时应用连接从节点) , 分担服务器负载;尤其是在写少读多的场景下,通过多个从节点分担读负载,可以大大提高Redis服务器的并发量。

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

一般来说，要将Redis运用于工程项目中,只使用一台Redis是万万不能（宕机）的,原因如下:

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

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

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

对于这中场景，我们可以使用以下结构：

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

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

环境配置

只配置从库

127.0.0.1:6379> info replication # 查看当前库的信息
# Replication
role:master # 角色 master
connected_slaves:0 # 没有丛机
master_failover_state:no-failover
master_replid:c6df7267ba665a502bf3d590f86f8b310c8ca987
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
1
2
3
4
5
6
7
8
9
10
11
12
13

复制三个配置文件，然后修改对应的信息：

1. 端口
2. pid名字
3. log文件的名字
4. dump.rdb名字
   

一主二从

默认情况下,每台Redis服务器都是主节点；一般情况下只要配从机就好了！

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

127.0.0.1:6380> SLAVEOF 127.0.0.1 6379 # SLAVEOF 127.0.0.1 6379 找谁当老大
OK
127.0.0.1:6380> info replication
# Replication
role:slave # 当前的角色是从机
master_host:127.0.0.1  # 这里可以看到主机的信息
master_port:6379
master_link_status:up
master_last_io_seconds_ago:3
master_sync_in_progress:0
slave_read_repl_offset:28
slave_repl_offset:28
slave_priority:100
slave_read_only:1
replica_announced:1
connected_slaves:0
master_failover_state:no-failover
master_replid:b76d6d0e8c4393504dd367038dc2bc92ca3f5556
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:28
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:28


# 在主机中看信息
127.0.0.1:6379> info replication
# Replication
role:master
connected_slaves:1 # 从机的信息
slave0:ip=127.0.0.1,port=6380,state=online,offset=56,lag=1
master_failover_state:no-failover
master_replid:b76d6d0e8c4393504dd367038dc2bc92ca3f5556
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:56
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:56

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

如果两个都配置完毕

真实的主从配置应该在配置文件中配置，这样的话是永久的，我们这里使用的是命令，是暂时的

细节

主机可以写，从机不能写只能读！主机中的所有信息和数据，都会自动被从机保存！

主机写：

从机只能读取，不能写

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

如果是使用命令行配置的主从，如果重启就会变回主机！只要变回从机，立马就会从主机中获取值！

复制原理

1. Slave启动成功连接到master后会发送一个sync命令。
2. Master接到命令启动后台的存盘进程，同时收集所有接收到的用于修改数据集命令， 在后台进程执行完毕之后，master将传送整个数据文件到slave,以完成一次完全同步。
3. 全量复制：而slave服务在接收到数据库文件数据后，将其存盘并加载到内存中。
4. 增量复制：Master继续将新的所有收集到的修改命令依次传给slave,完成同步。
5. 但是只要是重新连接master,一次完全同步（全量复制)将被自动执行。

层层链路

上一个M连接下一个s

层层链路是指如果从机太多，可以让第二个从机负责后面从机的主从复制。类似于一个公司最大的boss是老板，但是手下的人太多了，于是又分出了各种职位来继续领到下面的人。

如果没有老大，这个时候能不能选择一个老大出来呢？动手！

谋权篡位

如果主机断开了连接，我们可以使用SLAVEOF no one 让自己变成主机！其他节点就可以手动连接到这个最新的主节点（手动）！如果这个时候老大修复了，那就重新配置连接！

换句话说就是，只要之前主机断开连接，别的从机手动变为主机之后，哪怕之前的主机回归，他也没有从机了。

哨兵模式（自动选老大）

概述

主从切换技术的方法是:当主服务器宕机后,需要手动把一台从服务器切换为主服务器,这就需要人工干预,费事费力,还会造成段时间内服务不可用。这不是一种推荐的方式,更多时候,我们优先考虑哨兵模式。Redis从2.8开始正式提供 了Sentinel (哨兵)架构来解决这个问题。

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

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

这里的哨兵有两个作用

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

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

假设主服务器宕机,哨兵1先检测到这个结果,系统并不会马.上进行failover过程,仅仅是哨兵1主观的认为主服务器不可用,这个现象成为主观下线。

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

测试

我们目前的状态是一主二从

1. 哨兵配置文件sentinel.config

   # sentinel monitor 被监控的名称 host port
   sentinel monitor myredis 127.0.0.1 6379 1
   1
   2

   后面的数字1，表示这个主机挂了，slave投票看让谁成为主机，票数最多的成为主机。

2. 启动哨兵

   [root@zhao bin]# redis-sentinel myconfig/sentinel.conf
   7816:X 01 Nov 2021 23:37:31.854 # oO0OoO0OoO0Oo Redis is starting oO0OoO0OoO0Oo
   7816:X 01 Nov 2021 23:37:31.854 # Redis version=6.2.6, bits=64, commit=00000000, modified=0, pid=7816, just started
   7816:X 01 Nov 2021 23:37:31.854 # Configuration loaded
   7816:X 01 Nov 2021 23:37:31.855 * monotonic clock: POSIX clock_gettime
                   _._                                                  
              _.-``__ ''-._                                             
         _.-``    `.  `_.  ''-._           Redis 6.2.6 (00000000/0) 64 bit
     .-`` .-```.  ```\/    _.,_ ''-._                                  
    (    '      ,       .-`  | `,    )     Running in sentinel mode
    |`-._`-...-` __...-.``-._|'` _.-'|     Port: 26379
    |    `-._   `._    /     _.-'    |     PID: 7816
     `-._    `-._  `-./  _.-'    _.-'                                   
    |`-._`-._    `-.__.-'    _.-'_.-'|                                  
    |    `-._`-._        _.-'_.-'    |           https://redis.io       
     `-._    `-._`-.__.-'_.-'    _.-'                                   
    |`-._`-._    `-.__.-'    _.-'_.-'|                                  
    |    `-._`-._        _.-'_.-'    |                                  
     `-._    `-._`-.__.-'_.-'    _.-'                                   
         `-._    `-.__.-'    _.-'                                       
             `-._        _.-'                                           
                 `-.__.-'                                               
   
   7816:X 01 Nov 2021 23:37:31.863 # Sentinel ID is efa18f445818f0d3e5c16a5a9cf1cfab81e99f06
   7816:X 01 Nov 2021 23:37:31.863 # +monitor master myredis 127.0.0.1 6379 quorum 1
   7816:X 01 Nov 2021 23:37:31.864 * +slave slave 127.0.0.1:6380 127.0.0.1 6380 @ myredis 127.0.0.1 6379
   
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23
   24
   25
   26
   27

如果我们的Master节点断开了，这个时候就会从从机中随机选择一个服务器！（这里有一个投票算法）

如果此时前主机回来了，也只能当从机！

哨兵模式

优点：

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

缺点：

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

哨兵模式的全部配置

# 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 处于不能处理命令请求的状态。
# 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无法正常启动成功。

#通知脚本
# 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 # 一般都是由运维来配置！

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

12.Redis集群

问题

容量不够，redis如何进行扩容？
并发写操作， redis如何分摊？
另外，主从模式，薪火相传模式，主机宕机，导致ip地址发生变化，应用程序中配置需要修改对应的主机地址、端口等信息。78
之前通过代理主机来解决，但是redis3.0中提供了解决方案。就是无中心化集群配置。

什么是集群

Redis 集群实现了对Redis的水平扩容，即启动N个redis节点，将整个数据库分布存储在这N个节点中，每个节点存储总数据的1/N。

Redis 集群通过分区（partition）来提供一定程度的可用性（availability）： 即使集群中有一部分节点失效或者无法进行通讯， 集群也可以继续处理命令请求。

实例

1.删除持久化文件

将rdb,aof文件都删除掉。

2.制作6个实例，6379,6380,6381,6389,6390,6391

2.1.配置基本信息

开启daemonize yes
Pid文件名字
指定端口
Log文件名字
Dump.rdb名字
Appendonly 关掉或者换名字

2.2.Appendonly 关掉或者换名字

cluster-enabled yes 打开集群模式
cluster-config-file nodes-6379.conf 设定节点配置文件名
cluster-node-timeout 15000 设定节点失联时间，超过该时间（毫秒），集群自动进行主从切换。

include /home/bigdata/redis.conf
port 6379
pidfile "/var/run/redis_6379.pid"
dbfilename "dump6379.rdb"
dir "/home/bigdata/redis_cluster"
logfile "/home/bigdata/redis_cluster/redis_err_6379.log"
cluster-enabled yes
cluster-config-file nodes-6379.conf
cluster-node-timeout 15000
1
2
3
4
5
6
7
8
9

2.3.修改好redis6379.conf文件，拷贝多个

2.4.使用查找替换修改另外5个文件

例如：:%s/6379/6380

2.5.启动6个redis服务

3.将六个节点合成一个集群

组合之前，请确保所有redis实例启动后，nodes-xxxx.conf文件都生成正常。

合体：

cd  /opt/redis-6.2.1/src

redis-cli --cluster create --cluster-replicas 1 192.168.11.101:6379 192.168.11.101:6380 192.168.11.101:6381 192.168.11.101:6389 192.168.11.101:6390 192.168.11.101:6391
1
2
3

此处不要用127.0.0.1， 请用真实IP地址
–replicas 1 采用最简单的方式配置集群，一台主机，一台从机，正好三组。


普通方式登录
可能直接进入读主机，存储数据时，会出现MOVED重定向操作。所以，应该以集群方式登录。

4.-c 采用集群策略连接，设置数据会自动切换到相应的写主机

5.通过 cluster nodes 命令查看集群信息

6.redis cluster 如何分配这六个节点?

一个集群至少要有三个主节点。
选项 --cluster-replicas 1 表示我们希望为集群中的每个主节点创建一个从节点。
分配原则尽量保证每个主数据库运行在不同的IP地址，每个从库和主库不在一个IP地址上。

7.什么是slots

一个 Redis 集群包含 16384 个插槽（hash slot）， 数据库中的每个键都属于这 16384 个插槽的其中一个， 集群使用公式 CRC16(key) % 16384 来计算键 key 属于哪个槽， 其中 CRC16(key) 语句用于计算键 key 的 CRC16 校验和 。
集群中的每个节点负责处理一部分插槽。 举个例子， 如果一个集群可以有主节点， 其中：
节点 A 负责处理 0 号至 5460 号插槽。
节点 B 负责处理 5461 号至 10922 号插槽。
节点 C 负责处理 10923 号至 16383 号插槽。

8.在集群中录入值

在redis-cli每次录入、查询键值，redis都会计算出该key应该送往的插槽，如果不是该客户端对应服务器的插槽，redis会报错，并告知应前往的redis实例地址和端口。

redis-cli客户端提供了 –c 参数实现自动重定向。

如 redis-cli -c –p 6379 登入后，再录入、查询键值对可以自动重定向。

不在一个slot下的键值，是不能使用mget,mset等多键操作。

可以通过{}来定义组的概念，从而使key中{}内相同内容的键值对放到一个slot中去。

9.查询集群中的值

CLUSTER GETKEYSINSLOT 返回 count 个 slot 槽中的键。

10.故障恢复

如果主节点下线？从节点能否自动升为主节点？注意：15秒超时

主节点恢复后，主从关系会如何？主节点回来变成从机。

如果所有某一段插槽的主从节点都宕掉，redis服务是否还能继续?
如果某一段插槽的主从都挂掉，而cluster-require-full-coverage 为yes ，那么 ，整个集群都挂掉。
如果某一段插槽的主从都挂掉，而cluster-require-full-coverage 为no ，那么，该插槽数据全都不能使用，也无法存储。
redis.conf中的参数 cluster-require-full-coverage

11.集群的Jedis开发

即使连接的不是主机，集群会自动切换主机存储。主机写，从机读。
无中心化主从集群。无论从哪台主机写的数据，其他主机上都能读到数据。

public class JedisClusterTest {
  public static void main(String[] args) { 
     Set<HostAndPort>set =new HashSet<HostAndPort>();
     set.add(new HostAndPort("192.168.31.211",6379));
     JedisCluster jedisCluster=new JedisCluster(set);
     jedisCluster.set("k1", "v1");
     System.out.println(jedisCluster.get("k1"));
  }
}

1
2
3
4
5
6
7
8
9
10

12.Redis 集群提供了以下好处

• 实现扩容
• 分摊压力
• 无中心配置相对简单

13.Redis 集群的不足

• 多键操作是不被支持的
• 多键的Redis事务是不被支持的。lua脚本不被支持
• 由于集群方案出现较晚，很多公司已经采用了其他的集群方案，而代理或者客户端分片的方案想要迁移至redis cluster，需要整体迁移而不是逐步过渡，复杂度较大。

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

12.1.缓存穿透

1.问题描述

缓存穿透的概念很简单，用户想要查询一个数据，发现redis内存数据库没有，也就是缓存没有命中，于是向持久层数据库查询。发现还没有，于是本次查询失败。当用户很多的时候，缓存都没有命中（秒杀！），于是都去请求持久层数据库，这就会给持久层数据库带来很大压力，这个时候就是缓存穿透

2.解决方案

一个一定不存在缓存及查询不到的数据，由于缓存是不命中时被动写的，并且出于容错考虑，如果从存储层查不到数据则不写入缓存，这将导致这个不存在的数据每次请求都要到存储层去查询，失去了缓存的意义。

解决方案

（1） 对空值缓存：
如果一个查询返回的数据为空（不管是数据是否不存在），我们仍然把这个空结果（null）进行缓存，设置空结果的过期时间会很短，最长不超过五分钟

（2） 设置可访问的名单（白名单）：
使用bitmaps类型定义一个可以访问的名单，名单id作为bitmaps的偏移量，每次访问和bitmap里面的id进行比较，如果访问id不在bitmaps里面，进行拦截，不允许访问。
（3） 采用布隆过滤器：
(布隆过滤器（Bloom Filter）是1970年由布隆提出的。它实际上是一个很长的二进制向量(位图)和一系列随机映射函数（哈希函数）。
布隆过滤器可以用于检索一个元素是否在一个集合中。它的优点是空间效率和查询时间都远远超过一般的算法，缺点是有一定的误识别率和删除困难。)

（4） 进行实时监控：
当发现Redis的命中率开始急速降低，需要排查访问对象和访问的数据，和运维人员配合，可以设置黑名单限制服务。

12.2.缓存击穿

1.问题描述

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

2.解决方案

key可能会在某些时间点被超高并发地访问，是一种非常“热点”的数据。这个时候，需要考虑一个问题：缓存被“击穿”的问题。
（1）预先设置热门数据：
在redis高峰访问之前，把一些热门数据提前存入到redis里面，加大这些热门数据key的时长
（2）实时调整：
现场监控哪些数据热门，实时调整key的过期时长
（3）使用锁：
（1） 就是在缓存失效的时候（判断拿出来的值为空），不是立即去load db。
（2） 先使用缓存工具的某些带成功操作返回值的操作（比如Redis的SETNX）去set一个mutex key
（3） 当操作返回成功时，再进行load db的操作，并回设缓存,最后删除mutex key；
（4） 当操作返回失败，证明有线程在load db，当前线程睡眠一段时间再重试整个get缓存的方法。

12.3.缓存雪崩

1.问题描述

key对应的数据存在，但在redis中过期，此时若有大量并发请求过来，这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存，这个时候大并发的请求可能会瞬间把后端DB压垮。
缓存雪崩与缓存击穿的区别在于这里针对很多key缓存，前者则是某一个key正常访问

2.解决方案

（1） 构建多级缓存架构：
nginx缓存 + redis缓存 +其他缓存（ehcache等）
（2） 使用锁或队列：
用加锁或者队列的方式保证来保证不会有大量的线程对数据库一次性进行读写，从而避免失效时大量的并发请求落到底层存储系统上。不适用高并发情况
（3） 设置过期标志更新缓存：
记录缓存数据是否过期（设置提前量），如果过期会触发通知另外的线程在后台去更新实际key的缓存。
（4） 将缓存失效时间分散开：
比如我们可以在原有的失效时间基础上增加一个随机值，比如1-5分钟随机，这样每一个缓存的过期时间的重复率就会降低，就很难引发集体失效的事件。

13.分布式锁

1.问题描述

随着业务发展的需要，原单体单机部署的系统被演化成分布式集群系统后，由于分布式系统多线程、多进程并且分布在不同机器上，这将使原单机部署情况下的并发控制锁策略失效，单纯的Java API并不能提供分布式锁的能力。为了解决这个问题就需要一种跨JVM的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题！
分布式锁主流的实现方案：

1. 基于数据库实现分布式锁
2. 基于缓存（Redis等）
3. 基于Zookeeper

每一种分布式锁解决方案都有各自的优缺点：

1. 性能：redis最高
2. 可靠性：zookeeper最高
   这里，我们就基于redis实现分布式锁。

2.解决方案：使用redis实现分布式锁

redis:命令
set sku:1:info “OK” NX PX 10000
EX second ：设置键的过期时间为 second 秒。 SET key value EX second 效果等同于 SETEX key second value 。
PX millisecond ：设置键的过期时间为 millisecond 毫秒。 SET key value PX millisecond 效果等同于 PSETEX key millisecond value 。
NX ：只在键不存在时，才对键进行设置操作。 SET key value NX 效果等同于 SETNX key value 。
XX ：只在键已经存在时，才对键进行设置操作。

1. 多个客户端同时获取锁（setnx）
2. 获取成功，执行业务逻辑{从db获取数据，放入缓存}，执行完成释放锁（del）
3. 其他客户端等待重试

3.代码实现

@GetMapping("testLock")
public void testLock(){
    //1获取锁，setne
    Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", "111");
    //2获取锁成功、查询num的值
    if(lock){
        Object value = redisTemplate.opsForValue().get("num");
        //2.1判断num为空return
        if(StringUtils.isEmpty(value)){
            return;
        }
        //2.2有值就转成成int
        int num = Integer.parseInt(value+"");
        //2.3把redis的num加1
        redisTemplate.opsForValue().set("num", ++num);
        //2.4释放锁，del
        redisTemplate.delete("lock");

    }else{
        //3获取锁失败、每隔0.1秒再获取
        try {
            Thread.sleep(100);
            testLock();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

问题
场景：如果业务逻辑的执行时间是7s。执行流程如下

1. index1业务逻辑没执行完，3秒后锁被自动释放。
2. index2获取到锁，执行业务逻辑，3秒后锁被自动释放。
3. index3获取到锁，执行业务逻辑
4. index1业务逻辑执行完成，开始调用del释放锁，这时释放的是index3的锁，导致index3的业务只执行1s就被别人释放。
   最终等于没锁的情况。

解决：setnx获取锁时，设置一个指定的唯一值（例如：uuid）；释放前获取这个值，判断是否自己的锁

4.优化之UUID防误删

5.优化之LUA脚本保证删除的原子性

@GetMapping("testLockLua")
public void testLockLua() {
    //1 声明一个uuid ,将做为一个value 放入我们的key所对应的值中
    String uuid = UUID.randomUUID().toString();
    //2 定义一个锁：lua 脚本可以使用同一把锁，来实现删除！
    String skuId = "25"; // 访问skuId 为25号的商品 100008348542
    String locKey = "lock:" + skuId; // 锁住的是每个商品的数据

    // 3 获取锁
    Boolean lock = redisTemplate.opsForValue().setIfAbsent(locKey, uuid, 3, TimeUnit.SECONDS);

    // 第一种： lock 与过期时间中间不写任何的代码。
    // redisTemplate.expire("lock",10, TimeUnit.SECONDS);//设置过期时间
    // 如果true
    if (lock) {
        // 执行的业务逻辑开始
        // 获取缓存中的num 数据
        Object value = redisTemplate.opsForValue().get("num");
        // 如果是空直接返回
        if (StringUtils.isEmpty(value)) {
            return;
        }
        // 不是空 如果说在这出现了异常！ 那么delete 就删除失败！ 也就是说锁永远存在！
        int num = Integer.parseInt(value + "");
        // 使num 每次+1 放入缓存
        redisTemplate.opsForValue().set("num", String.valueOf(++num));
        /*使用lua脚本来锁*/
        // 定义lua 脚本
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        // 使用redis执行lua执行
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptText(script);
        // 设置一下返回值类型 为Long
        // 因为删除判断的时候，返回的0,给其封装为数据类型。如果不封装那么默认返回String 类型，
        // 那么返回字符串与0 会有发生错误。
        redisScript.setResultType(Long.class);
        // 第一个要是script 脚本 ，第二个需要判断的key，第三个就是key所对应的值。
        redisTemplate.execute(redisScript, Arrays.asList(locKey), uuid);
    } else {
        // 其他线程等待
        try {
            // 睡眠
            Thread.sleep(1000);
            // 睡醒了之后，调用方法。
            testLockLua();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

Lua 脚本详解：

项目中正确使用：

6.总结

1、加锁

// 1. 从redis中获取锁,set k1 v1 px 20000 nx
String uuid = UUID.randomUUID().toString();
Boolean lock = this.redisTemplate.opsForValue()
      .setIfAbsent("lock", uuid, 2, TimeUnit.SECONDS);

1
2
3
4
5

2、使用lua释放锁

// 2. 释放锁 del
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
// 设置lua脚本返回的数据类型
DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
// 设置lua脚本返回类型为Long
redisScript.setResultType(Long.class);
redisScript.setScriptText(script);
redisTemplate.execute(redisScript, Arrays.asList("lock"),uuid);

1
2
3
4
5
6
7
8
9

3、重试

Thread.sleep(500);
testLock();
1
2

为了确保分布式锁可用，我们至少要确保锁的实现同时满足以下四个条件：

• 互斥性。在任意时刻，只有一个客户端能持有锁。
• 不会发生死锁。即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁。
• 解铃还须系铃人。加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了。
• 加锁和解锁必须具有原子性。

14.Redis新特性

1.ACL

1.1.简介

Redis ACL是Access Control List（访问控制列表）的缩写，该功能允许根据可以执行的命令和可以访问的键来限制某些连接。
在Redis 5版本之前，Redis 安全规则只有密码控制 还有通过rename 来调整高危命令比如 flushdb ， KEYS* ， shutdown 等。Redis 6 则提供ACL的功能对用户进行更细粒度的权限控制 ：
（1）接入权限:用户名和密码
（2）可以执行的命令
（3）可以操作的 KEY

1.2.命令

1、使用acl list命令展现用户权限列表
（1）数据说明

2、使用acl cat命令
（1）查看添加权限指令类别

（2）加参数类型名可以查看类型下具体命令

3、使用acl whoami命令查看当前用户

4、使用aclsetuser命令创建和编辑用户ACL
（1）ACL规则
下面是有效ACL规则的列表。某些规则只是用于激活或删除标志，或对用户ACL执行给定更改的单个单词。其他规则是字符前缀，它们与命令或类别名称、键模式等连接在一起。

（2）通过命令创建新用户默认权限
acl setuser user1

在上面的示例中，我根本没有指定任何规则。如果用户不存在，这将使用just created的默认属性来创建用户。如果用户已经存在，则上面的命令将不执行任何操作。

（3）设置有用户名、密码、ACL权限、并启用的用户
acl setuser user2 on >password ~cached:* +get

(4)切换用户，验证权限

2.IO多线程

2.1.简介

Redis6终于支撑多线程了，告别单线程了吗？
IO多线程其实指客户端交互部分的网络IO交互处理模块多线程，而非执行命令多线程。Redis6执行命令依然是单线程。

2.2.原理架构

Redis 6 加入多线程,但跟 Memcached 这种从 IO处理到数据访问多线程的实现模式有些差异。Redis 的多线程部分只是用来处理网络数据的读写和协议解析，执行命令仍然是单线程。之所以这么设计是不想因为多线程而变得复杂，需要去控制 key、lua、事务，LPUSH/LPOP 等等的并发问题。整体的设计大体如下:

另外，多线程IO默认也是不开启的，需要再配置文件中配置
io-threads-do-reads yes
io-threads 4

3.其他新功能