Redis入门及简单原理分析

Redis

Redis中文官网

1.Mac环境下安装Redis

1.1.安装HomeBrew

/bin/zsh -c "$(curl -fsSL https://gitee.com/cunkai/HomebrewCN/raw/master/Homebrew.sh)"

1.2.使用HomeBrew安装Redis

brew install redis

会出现如下的命令行界面

==> Downloading https://mirrors.ustc.edu.cn/homebrew-bottles/bottles/redis-6.0.8
######################################################################## 100.0%
==> Pouring redis-6.0.8.high_sierra.bottle.tar.gz
==> Caveats
To have launchd start redis now and restart at login:
  brew services start redis
Or, if you don't want/need a background service you can just run:
  redis-server /usr/local/etc/redis.conf
==> Summary
🍺  /usr/local/Cellar/redis/6.0.8: 13 files, 3.8MB
==> `brew cleanup` has not been run in 30 days, running now...
  • Homebrew安装的软件会默认在/usr/local/Cellar/路径下
  • redis的配置文件redis.conf存放在/usr/local/etc路径下

1.3.修改redis.conf文件

修改redis.conf的目的是为了让redis可以后台运行

  • cd /usr/local/etc
  • vim redis.conf
  • /daemonize(vim的全文查找)
  • 修改为damoniz yes

1.4.启动Redis

redis-server 			#前台启动redis服务端
redis-server &		#后台启动redis服务端,停止可以使用jobs查看id,然后kill signal %id
redis-cli					#启动redis客户端
redis-server /usr/local/etc/redis.conf		#通过配置文件启动redis

启动以后的界面

2872:C 09 Feb 2021 14:41:00.551 # oO0OoO0OoO0Oo Redis is starting oO0OoO0OoO0Oo
2872:C 09 Feb 2021 14:41:00.551 # Redis version=6.0.8, bits=64, commit=00000000, modified=0, pid=2872, just started
2872:C 09 Feb 2021 14:41:00.551 # Warning: no config file specified, using the default config. In order to specify a config file use redis-server /path/to/redis.conf
2872:M 09 Feb 2021 14:41:00.552 * Increased maximum number of open files to 10032 (it was originally set to 256).
                _._                                                  
           _.-``__ ''-._                                             
      _.-``    `.  `_.  ''-._           Redis 6.0.8 (00000000/0) 64 bit
  .-`` .-```.  ```\/    _.,_ ''-._                                   
 (    '      ,       .-`  | `,    )     Running in standalone mode
 |`-._`-...-` __...-.``-._|'` _.-'|     Port: 6379
 |    `-._   `._    /     _.-'    |     PID: 2872
  `-._    `-._  `-./  _.-'    _.-'                                   
 |`-._`-._    `-.__.-'    _.-'_.-'|                                  
 |    `-._`-._        _.-'_.-'    |           http://redis.io        
  `-._    `-._`-.__.-'_.-'    _.-'                                   
 |`-._`-._    `-.__.-'    _.-'_.-'|                                  
 |    `-._`-._        _.-'_.-'    |                                  
  `-._    `-._`-.__.-'_.-'    _.-'                                   
      `-._    `-.__.-'    _.-'                                       
          `-._        _.-'                                           
              `-.__.-'                                               
2878:M 09 Feb 2021 14:44:05.191 # Server initialized
2878:M 09 Feb 2021 14:44:05.192 * Loading RDB produced by version 6.0.8
2878:M 09 Feb 2021 14:44:05.192 * RDB age 181 seconds
2878:M 09 Feb 2021 14:44:05.192 * RDB memory usage when created 0.95 Mb
2878:M 09 Feb 2021 14:44:05.193 * DB loaded from disk: 0.001 seconds
2878:M 09 Feb 2021 14:44:05.193 * Ready to accept connections

1.5.测试Redis可用性

命令行检测

(base) chennianzuisuideMacBook-Air:~ chennianzuisui$ ps -ef | grep redis
  501  2878   931   0  2:44下午 ttys000    0:00.36 redis-server *:6379		#redis服务端
  501  2899  2777   0  2:51下午 ttys001    0:00.00 grep redis

客户端检测

(base) chennianzuisuideMacBook-Air:~ chennianzuisui$ redis-cli
127.0.0.1:6379> ping
PONG
127.0.0.1:6379> set ch chenheng
OK
127.0.0.1:6379> get ch
"chenheng"
127.0.0.1:6379> 

1.6.redis-benchmark性能测试

序号选项描述默认值
1-h指定服务器主机名127.0.0.1
2-p指定服务器端口6379
3-s指定服务器 socket
4-c指定并发连接数50
5-n指定请求数10000
6-d以字节的形式指定 SET/GET 值的数据大小2
7-k1=keep alive 0=reconnect1
8-rSET/GET/INCR 使用随机 key, SADD 使用随机值
9-P通过管道传输 请求1
10-q强制退出 redis。仅显示 query/sec 值
11--csv以 CSV 格式输出
12-l生成循环,永久执行测试
13-t仅运行以逗号分隔的测试命令列表。
14-IIdle 模式。仅打开 N 个 idle 连接并等待。
redis-benchmark -c 100 -n 100000			#本机6379端口 100并发量,总数并发100000请求数,每个线程1000
#这边建议自己尝试一下,各个参数应该都是可以看得懂的,这里就不再解析了

1.7.参考文献

Mac安装Redis,原来就是这么简单

2.Redis基础知识

2.1.Redis基本命令

Redis有16个数据库

# Set the number of databases. The default database is DB 0, you can select
# a different one on a per-connection basis using SELECT <dbid> where
# dbid is a number between 0 and 'databases'-1
databases 16
127.0.0.1:6379> select 1					#选择1号数据库
OK
127.0.0.1:6379[1]> set name ch		#存储name-ch
OK
127.0.0.1:6379[1]> get name				#获取name
"ch"
127.0.0.1:6379[1]> keys *					#查看当前数据库的所有key-value
1) "name"
127.0.0.1:6379[1]> dbsize					#查看当前数据库的大小
(integer) 1
127.0.0.1:6379[1]> flushdb				#清空当前数据库
OK
127.0.0.1:6379[1]> keys *
(empty array)
127.0.0.1:6379[1]> flushall				#清空所有数据库
OK

2.2.Redis的6379端口号

6379在是手机按键上MERZ对应的号码,而MERZ取自意大利歌女Alessia Merz的名字。MERZ长期以来被Redis作者antirez及其朋友当作愚蠢的代名词。后来Redis作者在开发Redis时就选用了这个端口。

2.3.Redis单线程的性能优势

单线程指的是网络请求模块使用了一个线程(所以不需考虑并发安全性),其他模块仍用了多个线程。

Redis基于内存操作,CPU不是Redis的性能瓶颈,它的性能瓶颈是取决于机器内存和网络带宽

核心:Redis将数据放在内存中,所以说使用单线程去操作效率就是最高的,多线程(CPU上下文会切换,这属于耗时操作),对于内存系统来说,如果没有上下文切换效率会很高,多次读写都是在同一个CPU上,省去了CPU上下文切换的操作。

基于内存实现

  • 数据都存储在内存里,减少了一些不必要的 I/O 操作,操作速率很快。

高效的数据结构

  • 底层多种数据结构支持不同的数据类型,支持 Redis 存储不同的数据;
  • 不同数据结构的设计,使得数据存储时间复杂度降到最低。

合理的数据编码

  • 根据字符串的长度及元素的个数适配不同的编码格式。

合适的线程模型

  • I/O 多路复用模型同时监听客户端连接;
  • 单线程在执行过程中不需要进行上下文切换,减少了耗时。

硬核!15张图解Redis为什么这么快

3.五大数据类型

3.1.Redis-Key

127.0.0.1:6379[1]> move name 3									#删除key-value
(integer) 1
127.0.0.1:6379[1]> expire name 10								#设置key的过期时间,单位默认为秒
(integer) 1
127.0.0.1:6379[1]> ttl name											#查看距离过期的剩余时间
(integer) 7127.0.0.1:6379[1]> type name					#查看key的数据类型
string

3.2.String

127.0.0.1:6379[1]> append name hello				#追加字符串
(integer) 7
127.0.0.1:6379[1]> get name
"chhello"
127.0.0.1:6379[1]> strlen name							#获取字符串的长度
(integer) 7
#################################################自增,自减#################################################
127.0.0.1:6379[1]> set view 0		
OK
127.0.0.1:6379[1]> keys *
1) "name"
2) "view"
127.0.0.1:6379[1]> incr view 								#自增++
(integer) 1
127.0.0.1:6379[1]> decr view 								#自减--
(integer) 0
127.0.0.1:6379[1]> incrby view 10						#自增,步长为10
(integer) 10
#################################################范围操作#################################################
127.0.0.1:6379[1]> get name
"chhello"
127.0.0.1:6379[1]> setrange name 1 XXXXX		#替换指定位置开始的字符串
(integer) 7
127.0.0.1:6379[1]> get name
"cXXXXXo"		
127.0.0.1:6379[1]> getrange name 0 -1				#范围获取end=-1表示整个字符串
"cXXXXXo"
127.0.0.1:6379[1]> getrange name 0 2
"cXX"
#################################################set用法#################################################
127.0.0.1:6379[1]> setex password 30 "123"	#setex(set with expire)设置过期时间
OK
127.0.0.1:6379[1]> ttl password
(integer) 25
127.0.0.1:6379[1]> setnx password 321				#setnx(set if not exist)不存在设置(分布式锁中常常使用)
(integer) 0																	#0表示设置失败
127.0.0.1:6379[1]> ttl password
(integer) 7
#################################################批量设置#################################################
127.0.0.1:6379[1]> flushdb
OK
127.0.0.1:6379[1]> mset k1 v1 k2 v2 k3 v3		#批量设置
OK
127.0.0.1:6379[1]> mget k1 k2 k3						#批量获取
1) "v1"
2) "v2"
3) "v3"
127.0.0.1:6379[1]> msetnx k1 v1 k2 v2 k3 v3	#批量不存在设置,原子性操作,要么一起成功,要么一起失败
(integer) 0
#################################################对象#################################################
127.0.0.1:6379[1]> set user:1 {name:chenheng,age:22}			#法一:存储对象的json格式
OK
127.0.0.1:6379[1]> get user:1
"{name:chenheng,age:22}"
127.0.0.1:6379[1]> mset user:1:name chenheng user:1:age 22		#法二:key的巧妙设计:user:{id}:{field}
OK
127.0.0.1:6379[1]> mget user:1:name user:1:age
1) "chenheng"
2) "22"
#################################################getset#################################################
127.0.0.1:6379[1]> getset db "redis"					#先get再set。不存在key,则存储并返回nil;存在key,则获取再存储新的value
(nil)
127.0.0.1:6379[1]> get db
"redis"
127.0.0.1:6379[1]> getset db "mongodb"			
"redis"
127.0.0.1:6379[1]> get db
"mongodb"

3.3.List

#################################################push#################################################
127.0.0.1:6379[1]> lpush list one								#lpush:列表左压入一个元素
(integer) 1
127.0.0.1:6379[1]> lpush list two
(integer) 2
127.0.0.1:6379[1]> lpush list three
(integer) 3
127.0.0.1:6379[1]> lrange list 0 -1							#lrange:范围查看列表
1) "three"
2) "two"
3) "one"
127.0.0.1:6379[1]> lrange list 0 1
1) "three"
2) "two"
127.0.0.1:6379[1]> rpush list "right"						#rpush:列表右压入一个元素
(integer) 4
127.0.0.1:6379[1]> lrange list 0 -1
1) "three"
2) "two"
3) "one"
4) "right"
#################################################pop#################################################
127.0.0.1:6379[1]> lpop list										#lpop:左弹出
"three"
127.0.0.1:6379[1]> lrange list 0 -1
1) "two"
2) "one"
3) "right"
127.0.0.1:6379[1]> rpop list										#rpop:右弹出
"right"
127.0.0.1:6379[1]> lrange list 0 -1
1) "two"
2) "one"
#################################################index#################################################
127.0.0.1:6379[1]> lindex list 0								#lindex:定位列表的第几个元素,下标从0开始
"two"
127.0.0.1:6379[1]> lindex list 1
"one"
#################################################rem#################################################
127.0.0.1:6379[1]> llen list
(integer) 2
127.0.0.1:6379[1]> lrem list 1 "one"						#lrem key count element
(integer) 1
127.0.0.1:6379[1]> lrange list 0 -1
1) "two"
#################################################trim#################################################
127.0.0.1:6379[1]> rpush list 1 2 3 4 5
(integer) 5
127.0.0.1:6379[1]> ltrim list 1 3								#trim:修剪列表,只保留start~end的元素
OK
127.0.0.1:6379[1]> lrange list 0 -1
1) "2"
2) "3"
3) "4"
#################################################rpoplpush#################################################
127.0.0.1:6379[1]> flushdb
OK
127.0.0.1:6379[1]> lpush list 1 2 3 4 5
(integer) 5
127.0.0.1:6379[1]> rpoplpush list list1					#rpoplpush source destination:将source的最右边元素左压入到destination中
"1"
127.0.0.1:6379[1]> lrange list 0 -1
1) "5"
2) "4"
3) "3"
4) "2"
127.0.0.1:6379[1]> lrange list1 0 -1
1) "1"
#################################################lset#################################################
127.0.0.1:6379[1]> lset list 0 -1							#lset:下标为0的元素设置为-1,不存在则会报错
OK
#################################################linsert#################################################
127.0.0.1:6379[1]> lrange list 0 -1
1) "-1"
2) "4"
3) "3"
4) "2"
5) "1"
127.0.0.1:6379[1]> linsert list before "4" "-2"				#linsert key BEFORE|AFTER pivot element
(integer) 6
127.0.0.1:6379[1]> lrange list 0 -1
1) "-1"
2) "-2"
3) "4"
4) "3"
5) "2"
6) "1"

3.4.Set

########################################sadd&smembers&sismember#######################################
127.0.0.1:6379[1]> flushdb
OK
127.0.0.1:6379[1]> sadd set "hello"						#sadd:集合添加
(integer) 1
127.0.0.1:6379[1]> smembers set								#smembers:查看集合的所有元素
1) "hello"
127.0.0.1:6379[1]> sismember set "hello"			#sismember:集合中是否包含某个元素,这里表示set集合中是否包含"hello"元素
(integer) 1
#################################################scard#################################################
127.0.0.1:6379[1]> scard set									#scard:查看集合内元素的个数
(integer) 1
#################################################srem#################################################
127.0.0.1:6379[1]> scard set
(integer) 1
127.0.0.1:6379[1]> srem set "hello"						#srem key member:移除set中指定元素
(integer) 1
127.0.0.1:6379[1]> smembers set
(empty array)
#################################################srandmember#################################################
127.0.0.1:6379[1]> smembers set
1) "world"
2) "heng"
3) "chen"
127.0.0.1:6379[1]> srandmember set						#srandmember key count:随机获取集合中的count个元素
"world"
127.0.0.1:6379[1]> srandmember set
"chen"
#################################################集合操作#################################################
127.0.0.1:6379[1]> sadd s1 1 2 3 4 5
(integer) 5
127.0.0.1:6379[1]> sadd s2 2 3 4
(integer) 3
127.0.0.1:6379[1]> sdiff s1 s2								#差集
1) "1"
2) "5"
127.0.0.1:6379[1]> sunion s1 s2								#并集
1) "1"
2) "2"
3) "3"
4) "4"
5) "5"
127.0.0.1:6379[1]> sinter s1 s2								#交集
1) "2"
2) "3"
3) "4"
#################################################smove#################################################
127.0.0.1:6379[1]> smove s1 s2 1						#smove resource destination member
(integer) 1
127.0.0.1:6379[1]> smembers s2
1) "1"
2) "2"
3) "3"
4) "4"

3.5.Hash

数据类型:key-map

127.0.0.1:6379> hmset myhash field1 hello field2 world			#批量设置
OK
127.0.0.1:6379> hmget myhash field1 field2									#批量获取
1) "hello"
2) "world"
127.0.0.1:6379> hgetall myhash															#获取hash的所有数据,包括key
1) "field1"
2) "hello"
3) "field2"
4) "world"
#################################################hdel#################################################
127.0.0.1:6379> hdel myhash field1													#删除hash的字段
(integer) 1
127.0.0.1:6379> hgetall myhash
1) "field2"
2) "world"
#################################################hlen#################################################
127.0.0.1:6379> hlen myhash																	#获取hash的大小
(integer) 1
#################################################hexists#################################################
127.0.0.1:6379> hexists myhash field2												#判断字段是否存在
(integer) 1
127.0.0.1:6379> hexists myhash field1
(integer) 0
#################################################hkeys&hvals#################################################
127.0.0.1:6379> hkeys myhash																#只获取所有的key
1) "field2"
127.0.0.1:6379> hvals myhash																#只获取所有的value
1) "world"
#################################################hincrby#################################################
127.0.0.1:6379> hset myhash view 0
(integer) 1
127.0.0.1:6379> hincrby myhash view 5												#自增步长
(integer) 5
#################################################hsetnx#################################################
127.0.0.1:6379> hsetnx myhash view 0												#如果不存在则设置,因为viwe已存在,所以不设置
(integer) 0
127.0.0.1:6379> hget myhash view
"5"

3.6.Zset

Zset是有序集合

127.0.0.1:6379> zadd myset 1 one														#有序集合添加
(integer) 1
127.0.0.1:6379> zadd myset 2 two 3 three 
(integer) 2
127.0.0.1:6379> zrange myset 0 -1														#有序集合遍历查询
1) "one"
2) "two"
3) "three"
#################################################zrangebyscore#################################################
127.0.0.1:6379> zadd salary 2500 one 500 two 5000 three
(integer) 3
127.0.0.1:6379> zrange salary 0 -1
1) "two"
2) "one"
3) "three"
127.0.0.1:6379> zrangebyscore salary -inf +inf							#查询(-inf,+inf)的salary
1) "two"
2) "one"
3) "three"
127.0.0.1:6379> zrevrangebyscore salary +inf -inf						#倒序查询(+inf,-inf)的salary
1) "three"
2) "one"
3) "two"
127.0.0.1:6379> zrangebyscore salary -inf +inf withscores		#查询(-inf,+inf)的salary,并带有数值
1) "two"
2) "500"
3) "one"
4) "2500"
5) "three"
6) "5000"
#################################################zrem#################################################
127.0.0.1:6379> zrem salary one															#移除指定的元素
(integer) 1
127.0.0.1:6379> zrange salary 0 -1
1) "two"
2) "three"
#################################################zcard#################################################
127.0.0.1:6379> zcard salary																#有序集合内元素个数
(integer) 3
#################################################zcount#################################################
127.0.0.1:6379> zcount salary 500 2500											#获取指定区间的成员数量
(integer) 2

4.三种特殊数据类型

4.1.Geospatial

|ˌjēōˈspāSH(ə)l|

#################################################geoadd#################################################
127.0.0.1:6379> geoadd Sicily 13.361389 38.115556 Palermo 15.087269 37.502669 Catania			#添加地理信息(经,纬,地点名)
(integer) 2
#################################################geodist#################################################
127.0.0.1:6379> geodist Sicily Palermo Catania							#计算两个地点之间的距离
"166274.1516"
127.0.0.1:6379> georadius Sicily 15 37 100 km								#查询坐标为(15,37),半径100km的以内的地点
1) "Catania"
127.0.0.1:6379> georadius Sicily 15 37 200 km				
1) "Palermo"
2) "Catania"
127.0.0.1:6379> georadius Sicily 15 37 200 km withdist withcoord count 1			#外带参数距离和具体经纬度
1) 1) "Catania"
   2) "56.4413"
   3) 1) "15.08726745843887329"
      2) "37.50266842333162032"
127.0.0.1:6379> georadius Sicily 15 37 200 km withdist withcoord count 2
1) 1) "Catania"
   2) "56.4413"
   3) 1) "15.08726745843887329"
      2) "37.50266842333162032"
2) 1) "Palermo"
   2) "190.4424"
   3) 1) "13.36138933897018433"
      2) "38.11555639549629859"
#################################################geohash#################################################
127.0.0.1:6379> geohash Sicily Palermo											#获取地点的hash值
1) "sqc8b49rny0"
#################################################geopos#################################################
127.0.0.1:6379> geopos Sicily Palermo												#获取地点的经纬度
1) 1) "13.36138933897018433"
   2) "38.11555639549629859"
#################################################georadiusbymember#############################################
127.0.0.1:6379> georadiusbymember Sicily Catania 200 km			#以地点为基准
1) "Palermo"
2) "Catania"

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

127.0.0.1:6379> zrem Sicily Catania													#删除GEO中的元素
(integer) 1
127.0.0.1:6379> keys *										
1) "Sicily"
127.0.0.1:6379> zrange Sicily 0 -1													#查询GEO中的元素
1) "Palermo"

4.2.Hyperloglogs

基数统计,基数集合中不同元素的个数,{1,2,3,4,5,1,2}的基数大小为5

基数统计可以接受误差,Redis Hyperloglogs用于基数统计,相较于传统的记录id,存储在内存中,消耗大量内存,Hyperloglogs只需要12KB的固定内存,对于基数统计更具目的性,方便计数而不是存储用户id

用途:网页的UV,统计网页访问量,允许误差,一个人多次访问只算做一次

127.0.0.1:6379> pfadd mykey a b c d e f g h i j						#添加元素
(integer) 1
127.0.0.1:6379> pfadd mykey2 i j k l 
(integer) 1
127.0.0.1:6379> pfcount mykey															#统计hyperloglogs中的元素个数
(integer) 10
127.0.0.1:6379> pfcount mykey2
(integer) 4
127.0.0.1:6379> pfmerge mykey mykey2											#合并两个hyperloglogs到mykey中
OK
127.0.0.1:6379> pfcount mykey
(integer) 12

4.3.Bitmaps

用途:通知用户打卡信息,活跃不活跃的情况

127.0.0.1:6379> setbit sign 0 1														#设置bit
(integer) 0
127.0.0.1:6379> setbit sign 1 2
(error) ERR bit is not an integer or out of range
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 0
(integer) 0
127.0.0.1:6379> getbit sign 4
(integer) 0
127.0.0.1:6379> getbit sign 3
(integer) 1
127.0.0.1:6379> bitcount sign														#统计1的个数
(integer) 3

5.事务

5.1.Redis事务概述

Redis单条命令保证原子性,但是事务不保证原子性!

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

  • 一次性
  • 顺序性
  • 排他性

5.2.Redis事务流程

  1. 开启事务(multi)
  2. 命令入队(........)
  3. 关闭事务(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> exec						#关闭事务
1) OK
2) OK
3) "v1"

5.3.放弃事务

127.0.0.1:6379> flushdb
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379> set k1 v1
QUEUED
127.0.0.1:6379> discard					#放弃事务
OK
127.0.0.1:6379> get k1
(nil)

5.4.事务异常

编译型异常(代码有问题,命令有错),事务中所有命令都不会被执行

127.0.0.1:6379> flushdb
OK
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> getset k2				#错误命令
(error) ERR wrong number of arguments for 'getset' command
127.0.0.1:6379> set k3 v3
QUEUED
127.0.0.1:6379> exec						#执行事务报错
(error) EXECABORT Transaction discarded because of previous errors.
127.0.0.1:6379> get k1					#之前事务的所有命令都不会被执行
(nil)

运行时异常(例如1/0)

127.0.0.1:6379> flushdb
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379> set k1 "v1"
QUEUED
127.0.0.1:6379> incr k1			#执行失败,Redis中String不能自增
QUEUED
127.0.0.1:6379> exec				#第二条命令出错,但其他的仍然成功执行,从而验证了刚才的说法,Redis的事务不具原子性
1) OK
2) (error) ERR value is not an integer or out of range

5.5.Redis实现乐观锁

悲观锁

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

乐观锁

  • 很乐观,认为什么时候都不会出问题,所以不会上锁。更新数据的时候判断一下,在此期间是否有人修改过这个数据
  • version(库表设计时加入该字段)
  • 更新时比较version

Redis使用watch实现乐观锁

正常执行

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
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379> decrby money 20
QUEUED
127.0.0.1:6379> incrby out 20
QUEUED
127.0.0.1:6379> exec
1) (integer) 80
2) (integer) 20

多线程测试事务情况下(非事务情况下仍然可以)

127.0.0.1:6379> watch money
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379> decrby money 10
QUEUED
127.0.0.1:6379> incrby out 10
QUEUED
127.0.0.1:6379> exec					#执行之前,另一个线程修改了我们之前的值,这个时候会导致事务执行失败!
(nil)
127.0.0.1:6379> set money 200			#非事务情况下即使监视也可以修改
OK

6.Jedis

Jedis的api和上面的一模一样,所以这里不再赘述

7.SpringBoot集成Redis

7.1.说明

在SpringBoot2.x之后,原本使用jedis被替换为lettuce

  • Jedis采用直连,多个线程操作不安全,如果想要避免需要使用Jedis pool连接池,类似BIO
  • Lettuce采用netty,实例可以多个线程共享,不存在线程不安全情况,类似NIO
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(RedisOperations.class)
@EnableConfigurationProperties(RedisProperties.class)//对应的配置文件类
@Import({ LettuceConnectionConfiguration.class, JedisConnectionConfiguration.class })
public class RedisAutoConfiguration {

    @Bean
    @ConditionalOnMissingBean(name = "redisTemplate")//如果不存在自定义的redisTemplate再配置如下的Bean
    @ConditionalOnSingleCandidate(RedisConnectionFactory.class)
    public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
      //默认的RedisTemplate没有过多的配置,Redis对象都是需要序列化
      //两个泛式都是Object,Object类型,我们后续使用需要强制转化为<String,Object>
      RedisTemplate<Object, Object> template = new RedisTemplate<>();
      template.setConnectionFactory(redisConnectionFactory);
      return template;
    }

    @Bean
    @ConditionalOnMissingBean
  	//@ConditionalOnSingleCandidate表示当指定Bean在容器中只有一个,或者虽然有多个但是指定首选Bean情况下才会创建如下的Bean
    @ConditionalOnSingleCandidate(RedisConnectionFactory.class)
  	//由于<String,Object>是最常使用的类,所以单独提出一个Bean
    public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory redisConnectionFactory) {
      StringRedisTemplate template = new StringRedisTemplate();
      template.setConnectionFactory(redisConnectionFactory);
      return template;
    }

}

7.2.整合Redis

7.2.1.导入依赖

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

7.2.2.配置文件

#SpringBoot所有配置类,都有一个自动配置类,对应Redis为RedisAutoConfiguration
#自动配置类都会绑定一个properties配置文件类,对应Redis为RedisProperties
spring.redis.host=127.0.0.1
spring.redis.port=6379

7.2.3.测试

@Resource
private RedisTemplate redisTemplate;

@Test
public void test9() {
    //opsForValue   String  set——普通字符串  setBit——bitmaps
    //opsForList    List
    //opsForSet     Set
    //opsForHash    Hash
    //opsForZSet    Zset
    //opsForHyperLogLog     Hyperloglog
    //opsForGeo     Geospatial
    redisTemplate.opsForValue().set("hello", "world");
    System.out.println(redisTemplate.opsForValue().get("hello"));//world
}

7.3.自定义RedisTemplate

Redis默认的序列化是使用JDK序列化

image-20210213215912251

使用JDK序列化带来的问题是终端的中的中文字符显示为乱码

image-20210214222707460

解决这个问题需要修改默认的序列化规则,所以我们通过自定义RedisTemplate解决

@Configuration
public class RedisConfig {
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(redisConnectionFactory);

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

        //String序列化配置
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();

        //key string序列化
        template.setKeySerializer(stringRedisSerializer);
        //hash的key string序列化
        template.setHashKeySerializer(stringRedisSerializer);
        //value jackson序列化
        template.setValueSerializer(jackson2JsonRedisSerializer);
        //hash的value jackson序列化
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        template.afterPropertiesSet();

        return template;
    }
}
redis-cli --raw			#启动客户端,不然即使自定义了RedisTemplate也会出现乱码和转义字符

image-20210214231424411

7.4.Redis工具类

@Component
public final class RedisUtil {

    @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(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;
        }
    }
}

8.Redis.conf详解

8.1.单位

# Redis configuration file example.
# 
# Note that in order to read the configuration file, Redis must be
# started with the file path as first argument:
#
# ./redis-server /path/to/redis.conf

# Note on units: when memory size is needed, it is possible to specify
# it in the usual form of 1k 5GB 4M and so forth:
#
# 1k => 1000 bytes
# 1kb => 1024 bytes
# 1m => 1000000 bytes
# 1mb => 1024*1024 bytes
# 1g => 1000000000 bytes
# 1gb => 1024*1024*1024 bytes
#
# units are case insensitive so 1GB 1Gb 1gB are all the same.
# 单位大小写不敏感

8.2.包含

################################## INCLUDES ###################################

# Include one or more other config files here.  This is useful if you
# have a standard template that goes to all Redis servers but also need
# to customize a few per-server settings.  Include files can include
# other files, so use this wisely.
#
# Notice option "include" won't be rewritten by command "CONFIG REWRITE"
# from admin or Redis Sentinel. Since Redis always uses the last processed
# line as value of a configuration directive, you'd better put includes
# at the beginning of this file to avoid overwriting config change at runtime.
#
# If instead you are interested in using includes to override configuration
# options, it is better to use include as the last line.
#
# include /path/to/local.conf
# include /path/to/other.conf

可以包含其他的配置文件,类似于spring,jsp中的include,可以加入其他的配置文件,从而避免单个配置文件过于繁杂

8.3.网络

bind 127.0.0.1 ::1			#绑定本机IP地址,IPV4:127.0.0.1,IPV6:::1

# When protected mode is on and if:
#
# 1) The server is not binding explicitly to a set of addresses using the
#    "bind" directive.
# 2) No password is configured.
# 安全模式触发的时机是在无bind和无密码认证的条件同时存在时
# 
# The server only accepts connections from clients connecting from the
# IPv4 and IPv6 loopback addresses 127.0.0.1 and ::1, and from Unix domain
# sockets.
# 安全模式的作用:Redis服务端仅接受来自bind的地址或密码认证的连接,这里bind了本机,所以不会触发Redis安全模式
# 
protected-mode yes			#开启安全模式

prot 6379								#绑定端口

# TCP心跳包,监测TCP连接是否存活,默认是2小时一次,Redis默认配置是300s一次
tcp-keepalive 300

8.4.通用

daemonize yes				#以守护进程开启Redis服务

pidfile /var/run/redis_6379.pid		#以守护进程形式开启Redis,需要指定一个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)		#debug
# notice (moderately verbose, what you want in production probably)	#生产环境
# warning (only very important / critical messages are logged)			#重要信息
# 确定日志等级,符合日志等级的日志才会被打印出来
loglevel notice

logfile ""			#日志文件位置和名称

databases 16		#默认数据库16个

always-show-logo yes	#是否总是显示LOGO

8.5.快照持久化

快照持久化又称RDB持久化

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

save 900 1				#900s内,至少对key操作过1次,就进行持久化操作
save 300 10				#300s内,至少对key操作过10次,就进行持久化操作
save 60 10000			#60s内,至少对key操作过10000次,就进行持久化操作

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

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

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

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

8.6.主从复制

8.7.安全

8.7.1.控制台设置(优先)

(base) chennianzuisuideMacBook-Air:etc chennianzuisui$ redis-cli
127.0.0.1:6379> ping								#之前设置过密码,所以这里需要认证
(error) NOAUTH Authentication required.			
127.0.0.1:6379> auth ""							#认证
OK
127.0.0.1:6379> config set requirepass 123456			#设置认证密码
OK
127.0.0.1:6379> config get requirepass						#获取认证密码
1) "requirepass"
2) "123456"

8.7.2.配置文件设置

# requirepass foobared
requirepass 123456

8.8.客户端限制

maxclients 10000		#设置能连接上Redis的最多客户端数

maxmemory <bytes>		#配置最大Redis内存最大容量

maxmemory-policy noeviction		#内存达到上限之后的处理策略
	noeviction: 不删除策略, 达到最大内存限制时, 如果需要更多内存, 直接返回错误信息。(默认值)
  allkeys-lru: 所有key通用; 优先删除最近最少使用(less recently used ,LRU) 的 key。
  volatile-lru: 只限于设置了 expire 的部分; 优先删除最近最少使用(less recently used ,LRU) 的 key。
  allkeys-random: 所有key通用; 随机删除一部分 key。
  volatile-random: 只限于设置了 expire 的部分; 随机删除一部分 key。
  volatile-ttl: 只限于设置了 expire 的部分; 优先删除剩余时间(time to live,TTL) 短的key。
  volatile-lfu(least frequently used):从已设置过期时间的数据集(server.db[i].expires)中挑选最不经常使⽤的数据淘汰
	allkeys-lfu(least frequently used):当内存不⾜以容纳新写⼊数据时,在键空间中,移除最不经常使⽤的 key

8.9.AOF模式

append only file模式 aof配置

appendonly no				#默认不开启aof模式,默认开启rdb模式

appendfilename "appendonly.aof"				#默认生成aof文件名为appendonly.aof

# appendfsync always			#每执行一次操作,就sync一次
appendfsync everysec			#每秒执行一次sync,可能会丢失1s的数据(服务器可能在这1s宕机了)
# appendfsync no					#Redis不执行sync,由操作系统自己同步数据,速度最快

9.Redis持久化

9.1.RDB(Redis DataBase)

9.1.1.原理

先列出参考文献

RDB持久化操作的核心操作是bgsavebgsave的原理是fork()+copyonwrite(写时复制)

bgsave的主要流程是接受命令的主进程(进程A),派生出做持久化操作的子进程(进程B),A和B共享同一份内存,二者互不影响。

看到这里可能有人会提出疑问了,同一份内存和互不影响好像有点矛盾?

对于这个疑问,我们先了解一下fork后的父子进程的内存关系

  • 全量复制(一模一样的内存拷贝一份,极大的浪费了内存,8G内存,实则可用的就4G,剩下的4G都得用来做子进程的预留空间)
  • 非全量复制(两个进程操作都在同一份内存上,学过数据库的我们知道,这样会造成数据的脏读和幻读)
  • 折中的方法(写时复制,对于全量复制我们做一点优化,我们知道fork后的代码段是一致的,那代码段我们可以只有一份,我们把这种想法再扩散,但又要避免完全共享一块内存,那么可以模仿单例模式中的懒加载的模式,只有要修改的部分才复制,不修改的部分还是同一份,这就是写时复制)

写时复制前

img

写时复制后

img

所以fork之后,父子进程拥有一份相同的页帧(物理内存),其中页表(虚拟内存)中指向RAM代码段的部分是不会改变的,而指向数据段,堆段,栈段的会在我们将要改变父子进程各自的这部分内容时,才会将要操作的部分进行部分复制

主进程fork()子进程之后,内核把主进程中所有的内存页的权限都设为read-only,然后子进程的地址空间指向主进程。这也就是共享了主进程的内存,当其中某个进程写内存时(这里肯定是主进程写,因为子进程只负责rdb文件持久化工作,不参与客户端的请求),CPU硬件检测到内存页是read-only的,于是触发页异常中断(page-fault),陷入内核的一个中断例程。中断例程中,内核就会把触发的异常的页复制一份(这里仅仅复制异常页,也就是所修改的那个数据页,而不是内存中的全部数据),于是主子进程各自持有独立的一份。——《彻底搞懂Redis持久化之RDB原理》

总结一下的话就是bgsave的fork提高了时间的利用率,copyonwrite提高了内存的空间利用率

9.1.2触发机制

  1. save规则满足的情况下,会自动触发rdb规则(Redis客户端执行save或bgsave也可以产生rdb文件)
    • save阻塞服务器进程,直到rdb文件创建完毕,阻塞期间服务器不能处理任何命令请求
    • bgsave派生一个子进程,然后由子进程负责创建rdb文件,父进程继续处理命令请求(推荐)
  2. 执行flushall命令,也会触发rdb规则
  3. 退出Redis,也会产生rdb文件

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

9.1.3.恢复rdb文件

只需要将rdb文件放置在Redis配置文件的dir参数目录下即可通过自动检查dump.rdb恢复其中的数据(这里是/usr/local/etc)

image-20210215184543954

9.2.AOF(Append Only File)

参考文献

  1. 彻底搞懂Redis持久化之AOF原理

9.2.1.基本操作

将我们所有命令都记录下来,恢复的时候就把这个文件全部再执行一遍(命令以追加的形式加入到aof文件中)

以日志形式记录每个写操作,将Redis执行过的所有指令记录下来(读操作不记录),只追加文件但不可以写文件,Redis启动之初会读取该文件重新构建数据。

AOF保存的文件是appendonly.aof ,同时AOF模式默认不开启

appendonly yes			#开启AOF模式

如果appendonly.aof损坏,那么Redis将无法正常启动

(base) chennianzuisuideMacBook-Air:etc chennianzuisui$ redis-cli
Could not connect to Redis at 127.0.0.1:6379: Connection refused

可以使用redis-check-aof修复

(base) chennianzuisuideMacBook-Air:etc chennianzuisui$ redis-check-aof --fix appendonly.aof 
0x              a4: Expected \r\n, got: 6469
AOF analyzed: size=180, ok_up_to=139, diff=41
This will shrink the AOF from 180 bytes, with 41 bytes, to 139 bytes
Continue? [y/N]: y
Successfully truncated AOF

三种Redis的AOF同步策略

  • always(每次操作都刷新AOF缓冲区)
  • everysec(每秒刷新AOF缓冲区)
  • no(AOF缓冲区满,再刷入到AOF文件中)

9.2.2.重写原理

重写机制

混合模式就是rdb和aof一起用

Redis 4.0开始的rewrite支持混合模式,直接将rdb持久化的方式来操作将二进制内容覆盖到aof文件中(rdb是二进制,所以很小),然后再有写入的话还是继续append追加到文件原始命令,等下次文件过大的时候再次rewrite(还是按照rdb持久化的方式将内容覆盖到aof中)。但是这种模式也是配置的,默认是开,也可以关闭。

aof-use-rdb-preamble yes			#混合模式默认开启

image-20210215234955922

  • aof_rewrite_buf:rewrite(重写)缓冲区、aof_buf:写命令存放的缓冲区
  • 开始bgrewriteaof的时候,判断当前有没有bgsave/bgrewriteaof在执行,若有,则不执行,这个再rdb篇幅也有提到,以及下面很多fork()知识在rdb都有提到。彻底搞懂Redis持久化之RDB原理
  • 主进程fork()出子进程,在执行fork()这个方法的时候是阻塞的,子进程创建完毕后就不阻塞了
  • 主进程fork完子进程后,主进程能继续接收客户端的请求,所有写命令依然是写入AOF文件缓冲区并根据配置文件的策略同步到磁盘的。
  • 因为fork的子进程仅仅共享主进程fork()时的内存,后期主进程在更改内存数据,子进程是不可见的。因此Redis采取重写缓冲区(aof_rewite_buf)保存fork之后的客户端请求。防止新AOF文件生成期间丢失主进程执行的新命令所生成的数据。所以此时客户端的写请求不仅仅写入原来的aof_buf缓冲区,还写入了重写缓冲区。这就是我为什么用深蓝色的框给他两框到一起的原因。
  • 子进程通过内存快照的形式,开始生成新的aof文件。
  • 新aof文件生成完后,子进程向主进程发信号。
  • 主进程收到信号后,会把重写缓冲区(aof_rewite_buf)中的数据写入到新的AOF文件(主要是避免这部分数据丢失)
  • 使用新的AOF文件覆盖旧的AOF文件,且标记AOF重写完成。

混合持久化也是通过bgrewriteaof完成的,所以基本流程和上述一样。不同的是当开启混合模式时,fork出的子进程先将共享的内存副本全量以RDB的方式写入aof。这样提高了速度也极大的缩小了aof文件(毕竟都是二进制)。写完还是通知主进程,然后再将重写缓冲区的内容以AOF方式写入到文件,然后替换旧的aof文件。也就是说这种模式下的aof文件发生rewrite后前半部分是rdb格式(REDIS开头的二进制数据),后半部分是正常的aof追加的命令(重写缓冲区里的)。

==由上面联想到的一个问题,为什么会有两个缓冲区?==

想了一下发现如果只有一个的话,那么缓冲区中的新命令写入新AOF文件和客户端写操作写入缓冲区的操作同时发生,那不就造成了阻塞了嘛,所以正确的步骤应该是两个缓冲区,新AOF生成后,重写缓冲区写入新AOF文件中,再替换掉旧AOF,这时客户端的写操作也继续加入到缓冲区中(非重写缓冲区),待替换完成再根据AOF的同步策略把缓冲区的写命令同步到新AOF文件中。

触发条件

bgrewriteaof								#手动触发
auto-aof-rewrite-min-size		#AOF文件最小重写大小,只有当AOF文件大小大于该值时候才可能重写
auto-aof-rewrite-percentage	#当前AOF文件大小和最后一次重写后的大小之间的比率等于或者等于指定的增长百分比,如100代表当前AOF文件是上次重写的两倍时候才重写

10.Redis发布订阅

Redis发布订阅(pub/sub)是一种消息通信模型:发送者发送信息,订阅者接收信息

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

如下表格来自菜鸟教程

序号命令及描述
1[PSUBSCRIBE pattern pattern ...] 订阅一个或多个符合给定模式的频道。
2[PUBSUB subcommand argument [argument ...]] 查看订阅与发布系统状态。
3PUBLISH channel message 将信息发送到指定的频道。
4[PUNSUBSCRIBE pattern [pattern ...]] 退订所有给定模式的频道。
5[SUBSCRIBE channel channel ...] 订阅给定的一个或多个频道的信息。
6[UNSUBSCRIBE channel [channel ...]] 指退订给定的频道。

订阅端

127.0.0.1:6379> subscribe chsay		#订阅chsay频道
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "chsay"
3) (integer) 1
# 等待读取推送的信息
1) "message"			#消息
2) "chsay"				#频道
3) "helloworld"		#内容

发布端

127.0.0.1:6379> publish chsay helloworld		#发布者把内容发布到chsay频道
(integer) 1

11.Redis主从复制

11.1.主从复制概述

11.1.1.概念

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

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

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

11.1.2.作用

  1. 数据冗余:主从复制实现了数据的热备份,是持久化之外的一种数据冗余方式。
  2. 故障恢复:当主节点出现问题时,可以由从节点提供服务,实现快速的故障恢复;实际上是一种服务的冗余。
  3. 负载均衡:在主从复制的基础上,配合读写分离,可以由主节点提供写服务,由从节点提供读服务(即写Redis数据时应用连接主节点,读Redis数据时应用连接从节点),分担服务器负载;尤其是在写少读多的场景下,通过多个从节点分担读负载,可以大大提高Redis服务器的并发量。
  4. 高可用基石:除了上述作用以外,主从复制还是哨兵和集群能够实施的基础,因此说主从复制是Redis高可用的基础。

11.2.环境配置

127.0.0.1:6379> info replication
# Replication
role:master			#主机
connected_slaves:0	#从机数为0
master_replid:3b11f7cb9835e50a72048b0f632409a103c68bfa
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

复制3个配置文件,cp redis.conf redis6380.conf拷贝redis.conf,并命名为redis6380.conf,以此类推,得到3个配置文件redis6379.conf,redis6380.conf,redis6381.conf,再修改对应的信息

  1. port——端口
  2. pidfile——pid文件名
  3. logfile——log文件名
  4. dbfilename——db文件名
image-20210216200556655

11.3.一主二从

(认贼作父)一个主机(6379),两个从机(6380,6381)

11.3.1.命令行配置

#从机中查看
127.0.0.1:6380> slaveof 127.0.0.1 6379		#slaveof host prot	成为127.0.0.1:6379的从机
OK
127.0.0.1:6380> info replication
# Replication
role:slave			#角色:slave(奴隶)从机
master_host:127.0.0.1
master_port:6379
master_link_status:up
master_last_io_seconds_ago:7
master_sync_in_progress:0
slave_repl_offset:0
slave_priority:100
slave_read_only:1
connected_slaves:0
master_replid:12dd35bf737e7c06a4f49fa4df5863783124004f
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:0
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:0

#主机中查看
127.0.0.1:6379> info replication
# Replication
role:master
connected_slaves:1		#从机数为1
slave0:ip=127.0.0.1,port=6380,state=online,offset=322,lag=0		#从机详细信息
master_replid:12dd35bf737e7c06a4f49fa4df5863783124004f
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:322
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:322
image-20210216201549434

11.3.2.配置文件配置

replicaof <masterip> <masterport>		#replicaof 127.0.0.1 6379
masterauth <master-password>				#主机密码认证

11.3.3.测试

从机只能读取内容

image-20210216202339341

主机断开连接,从机保持连接

从机仍然可以读取到主机的数据,隶属关系不变;

如果主机回来了,从机仍然可以读取到主机后续添加的数据

image-20210216202700132

从机断开连接,主机保持连接

答案是不能,因为我们使用的命令行配置一主二从,重启之后会失去关系,重新变为主机,但在重新配置关系后即可获取到主机数据;

配置文件配置的话就可以在从机重连后仍然保持关系

127.0.0.1:6380> info replication
# Replication
role:master
connected_slaves:0
master_replid:23bb5d5092a0ff4117b7d6698e6ea16547610c22
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

11.4.原理

11.4.1.参考文献

彻底搞懂Redis主从复制原理及实战

11.4.2.基本流程

在这里插入图片描述

11.4.3.权限验证

在这里插入图片描述

11.4.4.同步数据集

全量同步(bgsave生成快照)

Redis全量复制流程

增量同步(这部分具体可以查看参考文献)

  1. 复制偏移量
  2. 复制积压缓冲区
  3. 服务器运行ID(runid)

11.4.5.命令持续复制

这里采用的策略视情况而定,可能是全量同步,也可能是增量同步(猜测:大多是增量,毕竟大量的全量bgsave十分消耗性能)。

11.4.6.psync(Redis2.8之后)

在这里插入图片描述

11.5.哨兵模式

参考文献:Redis哨兵(Sentinel)模式

自动选举老大的模式

主从切换技术的方法是:当主服务器宕机后,需要手动把一台从服务器切换为主服务器,这就需要人工干预,费事费力,还会造成一段时间内服务不可用。这不是一种推荐的方式,更多时候,我们优先考虑哨兵模式

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

单哨兵模式

img

多哨兵模式

img

配置

目前状态是一主二从

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

哨兵配置内容其实挺多的,这里只是最简单的。

#sentinel monitor 被监控的名称 host port 1
sentinel monitor myredis 127.0.0.1 6379 1
  1. 启动哨兵
redis-sentinel sentinel.conf

2745:X 16 Feb 2021 22:15:31.167 # oO0OoO0OoO0Oo Redis is starting oO0OoO0OoO0Oo
2745:X 16 Feb 2021 22:15:31.167 # Redis version=6.0.8, bits=64, commit=00000000, modified=0, pid=2745, just started
2745:X 16 Feb 2021 22:15:31.167 # Configuration loaded
2745:X 16 Feb 2021 22:15:31.169 * Increased maximum number of open files to 10032 (it was originally set to 256).
                _._                                                  
           _.-``__ ''-._                                             
      _.-``    `.  `_.  ''-._           Redis 6.0.8 (00000000/0) 64 bit
  .-`` .-```.  ```\/    _.,_ ''-._                                   
 (    '      ,       .-`  | `,    )     Running in sentinel mode
 |`-._`-...-` __...-.``-._|'` _.-'|     Port: 26379
 |    `-._   `._    /     _.-'    |     PID: 2745
  `-._    `-._  `-./  _.-'    _.-'                                   
 |`-._`-._    `-.__.-'    _.-'_.-'|                                  
 |    `-._`-._        _.-'_.-'    |           http://redis.io        
  `-._    `-._`-.__.-'_.-'    _.-'                                   
 |`-._`-._    `-.__.-'    _.-'_.-'|                                  
 |    `-._`-._        _.-'_.-'    |                                  
  `-._    `-._`-.__.-'_.-'    _.-'                                   
      `-._    `-.__.-'    _.-'                                       
          `-._        _.-'                                           
              `-.__.-'                                               

2745:X 16 Feb 2021 22:15:31.172 # Sentinel ID is bae4948ef37bd023905b863efdd002fd86fd78dc
2745:X 16 Feb 2021 22:15:31.172 # +monitor master myredis 127.0.0.1 6379 quorum 1
2745:X 16 Feb 2021 22:15:31.174 * +slave slave 127.0.0.1:6381 127.0.0.1 6381 @ myredis 127.0.0.1 6379
2745:X 16 Feb 2021 22:15:31.176 * +slave slave 127.0.0.1:6380 127.0.0.1 6380 @ myredis 127.0.0.1 6379

如果主机断开连接,哨兵会从从机中选举出一个主机

从下面的部分日志中我们可以看出,主机角色的交换,以及6379再次启动后也会变为从机

2745:X 16 Feb 2021 22:18:34.813 # +switch-master myredis 127.0.0.1 6379 127.0.0.1 6381
2745:X 16 Feb 2021 22:18:34.813 * +slave slave 127.0.0.1:6380 127.0.0.1 6380 @ myredis 127.0.0.1 6381
2745:X 16 Feb 2021 22:18:34.813 * +slave slave 127.0.0.1:6379 127.0.0.1 6379 @ myredis 127.0.0.1 6381
2745:X 16 Feb 2021 22:19:04.851 # +sdown slave 127.0.0.1:6379 127.0.0.1 6379 @ myredis 127.0.0.1 6381
image-20210216222259839

6379重启,6379果然变成了6381的从机

2745:X 16 Feb 2021 22:24:29.679 * +convert-to-slave slave 127.0.0.1:6379 127.0.0.1 6379 @ myredis 127.0.0.1 6381
image-20210216222544705

12.Redis缓存穿透和雪崩

参考资料:缓存穿透,缓存击穿,缓存雪崩解决方案分析

12.1.缓存穿透

12.1.1.概念

缓存穿透说简单点就是⼤量请求的 key 根本不存在于缓存中,导致请求直接到了数据库上,根本没有经过缓存这⼀层。举个例⼦:某个⿊客故意制造我们缓存中不存在的 key 发起⼤量请求,导致⼤量请求落到数据库。

image-20210216233221228

12.1.2.解决策略

  1. 缓存无效key

    • 如果缓存和数据库都查不到某个 key 的数据就写⼀个到 Redis 中去并设置过期时间,具体命令如下: SET key value EX 10086 。

    • 这种⽅式可以解决请求的 key 变化不频繁的情况,如果⿊客恶意攻击,每次构建不同的请求 key,会导致 Redis 中缓存⼤量⽆效的 key 。

    • 很明显,这种⽅案并不能从根本上解决此问题。如果⾮要⽤这种⽅式来解决穿透问题的话,尽量将⽆效的 key 的过期时

      间设置短⼀点⽐如 1 分钟。

  2. 布隆过滤器

本质上布隆过滤器是一种数据结构,比较巧妙的概率型数据结构(probabilistic data structure),特点是高效地插入和查询,可以用来告诉你 “某样东西一定不存在或者可能存在”。相比于传统的 List、Set、Map 等数据结构,它更高效、占用空间更少,但是缺点是其返回的结果是概率性的,而不是确切的。

具体参考:详解布隆过滤器的原理,使用场景和注意事项

12.2.缓存雪崩

12.2.1.概念

缓存在同⼀时间⼤⾯积的失效,后⾯的请求都直接落到了数据库上,造成数据库短时间内承受⼤量请求。

举个例⼦ :秒杀开始 12 个⼩时之前,我们统⼀存放了⼀批商品到 Redis 中,设置的缓存过期时间也是 12 个⼩时,那么秒杀开始的时候,这些秒杀的商品的访问直接就失效了。导致的情况就是,相应的请求直接就落到了数据库上,就像雪崩⼀样可怕。

12.3.缓存击穿

12.3.1.概念

对于一些设置了过期时间的key,如果这些key可能会在某些时间点被超高并发地访问,是一种非常“热点”的数据。这个时候,需要考虑一个问题:缓存被“击穿”的问题,这个和缓存雪崩的区别在于这里针对某一key缓存,前者则是很多key。

缓存在某个时间点过期的时候,恰好在这个时间点对这个Key有大量的并发请求过来,这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存,这个时候大并发的请求可能会瞬间把后端DB压垮。

12.3.2.解决方案

==这里的解决方案也是缓存雪崩的解决方案==

限流降级,使用互斥锁

为了缓解大并发流量,我们也可以使用限流降级的方式防止缓存雪崩。例如,在缓存失效后,通过加锁或者使用队列来控制读数据库写缓存的线程数量。具体点就是设置某些Key只允许一个线程查询数据和写缓存,其他线程等待。则能够有效的缓解大并发流量对数据库打来的巨大冲击。

业界比较常用的做法,是使用mutex。简单地来说,就是在缓存失效的时候(判断拿出来的值为空),不是立即去load db,而是先使用缓存工具的某些带成功操作返回值的操作(比如Redis的SETNX或者Memcache的ADD)去set一个mutex key,当操作返回成功时,再进行load db的操作并回设缓存;否则,就重试整个get缓存的方法。

SETNX,是「SET if Not eXists」的缩写,也就是只有不存在的时候才设置,可以利用它来实现锁的效果。在redis2.6.1之前版本未实现setnx的过期时间,所以这里给出两种版本代码参考:

public String get(key) {
      String value = redis.get(key);
      if (value == null) { //代表缓存值过期
        //设置3min的超时,防止del操作失败的时候,下次缓存过期一直不能load db
        if (redis.setnx(key_mutex, 1, 3 * 60) == 1) {  //代表设置成功
          value = db.get(key);
          redis.set(key, value, expire_secs);
          redis.del(key_mutex);
        } else {  //这个时候代表同时候的其他线程已经load db并回设到缓存了,这时候重试获取缓存值即可
          sleep(50);
          get(key);  //重试
        }
      } else {
        return value;      
      }
 }

提前使用互斥锁

即设置一个超时时间,如果超过就自动去DB中获取。

设置不同的失效时间(缓存雪崩)

⽐如随机设置缓存的失效时间。

缓存永不失效

针对某些特殊的情况,我们可以进行数据预热,即提前加载好数据,和提前使用互斥锁有异曲同工之妙,都是提前做好准备,以防万一的策略。

评论

Your browser is out-of-date!

Update your browser to view this website correctly. Update my browser now

×