首页 / REDIS / Redis进阶（二）个人参考使用
Redis进阶（二）个人参考使用
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了Redis进阶（二）个人参考使用，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含16301字，纯文字阅读大概需要24分钟。
内容图文
分布式缓存架构-Redis(二)
Redis发布订阅
Redis 发布订阅(pub/sub)是一种消息通信模式：发送者(pub)发送消息，订阅者(sub)接收消息。
Redis 客户端可以订阅任意数量的频道。
下图展示了频道 channel1 ， 以及订阅这个频道的三个客户端 —— client2 、 client5 和 client1 之间的关系：
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-bf5D6wgG-1607954996201)(img\1.png)]
当有新消息通过 PUBLISH 命令发送给频道 channel1 时， 这个消息就会被发送给订阅它的三个客户端：
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Qy5ahu3Z-1607954996203)(img\2.png)]
订阅消息命令
SUBSCRIBE channel1
发布消息命令
PUBLISH channel1 hello
使用Stream完成消息发布与消费
Stream发布和消费的的流程
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-fHiFvS4M-1607954996205)(img\image-20200812051912502.png)]
使用到的指令
- XADD：创建Stream
- XGROUP：是用来创建删除和管理消费组的
- XREADGROUP：是用来通过消费组从Stream里读消息的
- XACK：是用来确认消息已经消费的
创建Stream
xadd streamkey * field val
# 为了防止消息太长可以使用maxlen限制消息长度
# xadd streamkey maxlen 1000 * field val
创建消费组
# 消费组1
xgroup create streamkey g1 0-0
# 消费组2
xgroup create streamkey g2 $
# 查看消费组信息
xinfo groups streamkey
# consumers 该消费组消费者数量
# pending 该消费组正在处理的消息的数量
消费消息
# 消费者1
xreadgroup GROUP g1 c1 count 1 streams streamkey >
# 消费者2
xreadgroup GROUP g1 c2 count 1 streams streamkey >
# >号表示从当前消费组的last_delivered_id后面开始读
# 这里任然可是使用 block 0 进行阻塞等待
#立即确认消息
xack streamkey g1 ID ...
# 观察消费组指定消费者
xinfo consumers streamkey g1
# name
# pending
# idle 空闲了多长时间ms没有读取消息了
# 消息确认
# 获取消费历史
XREADGROUP GROUP g1 c1 STREAMS streamkey 0
# 修正组的时间戳
XGROUP SETID streamkey g1 1591840117074-0
# 确认消息
xack streamkey g1 ID ...
Redis持久化
什么是Redis持久化
就是将内存数据保存到硬盘。
Redis 持久化存储 (AOF 与 RDB 两种模式)
RDB持久化
RDB 是以二进制文件，是在某个时间点将数据写入一个临时文件，持久化结束后，用这个临时文件替换上次持久化的文件，达到数据恢复。
***优点：***使用单独子进程来进行持久化，主进程不会进行任何 IO 操作，保证了 redis 的高性能
缺点：
- RDB 是间隔一段时间进行持久化，如果持久化之间 redis 发生故障，会发生数据丢失。
- 子进程需要开销和主进程相同的内存完成数据保存，可能会导致内存溢出。
所以这种方式更适合数据要求不严谨的时候。
这个执行数据写入到临时文件的时间点是通过配置确定的，通过配置redis 在 n 秒内如果超过 m 个 key 被修改这执行一次 RDB 操作。这个操作就类似于在这个时间点来保存一次 Redis 的所有数据，一次快照数据。所有这个持久化方法也通常叫做快照（snapshots）。
RDB 默认开启，redis.conf 中的具体配置参数如下：
#dbfilename：持久化数据存储在本地的文件
dbfilename dump.rdb
#dir：持久化数据存储在本地的路径，如果是在/redis/redis-3.0.6/src下启动的redis-cli，则数据会存储在当前src目录下
dir ./
##snapshot触发的时机，save
##如下为900秒后，至少有一个变更操作，才会snapshot
##对于此值的设置，需要谨慎，评估系统的变更操作密集程度
##可以通过“save “””来关闭snapshot功能
#save时间，以下分别表示更改了1个key时间隔900s进行持久化存储；更改了10个key300s进行存储；更改10000个key60s进行存储。
save 900 1
save 300 10
save 60 10000
##当snapshot时出现错误无法继续时，是否阻塞客户端“变更操作”，“错误”可能因为磁盘已满/磁盘故障/OS级别异常等
stop-writes-on-bgsave-error yes
##是否启用rdb文件压缩，默认为“yes”，压缩往往意味着“额外的cpu消耗”，同时也意味这较小的文件尺寸以及较短的网络传输时间
rdbcompression yes
AOF持久化
Append-only file，将“操作 + 数据”以格式化指令的方式追加到操作日志文件的尾部，在 append 操作返回后(已经写入到文件或者即将写入)，才进行实际的数据变更，“日志文件”保存了历史所有的操作过程；当 server 需要数据恢复时，可以直接 replay 此日志文件，即可还原所有的操作过程。AOF 相对可靠，它和 mysql 中 bin.log、apache.log、zookeeper 中 txn-log 简直异曲同工。AOF 文件内容是字符串，非常容易阅读和解析。
***优点：***可以保持更高的数据完整性，如果设置追加 file 的时间是 1s，如果 redis 发生故障，最多会丢失 1s 的数据；且如果日志写入不完整支持 redis-check-aof 来进行日志修复；AOF 文件没被 rewrite 之前（文件过大时会对命令进行合并重写），可以删除其中的某些命令（比如误操作的 flushall）。
***缺点：***AOF 文件比 RDB 文件大，且恢复速度慢。
我们可以简单的认为 AOF 就是日志文件，此文件只会记录“变更操作”(例如：set/del 等)，如果 server 中持续的大量变更操作，将会导致 AOF 文件非常的庞大，意味着 server 失效后，数据恢复的过程将会很长；事实上，一条数据经过多次变更，将会产生多条 AOF 记录，其实只要保存当前的状态，历史的操作记录是可以抛弃的；因为 AOF 持久化模式还伴生了“AOF rewrite”。
AOF 的特性决定了它相对比较安全，如果你期望数据更少的丢失，那么可以采用 AOF 模式。如果 AOF 文件正在被写入时突然 server 失效，有可能导致文件的最后一次记录是不完整，你可以通过手工或者程序的方式去检测并修正不完整的记录，以便通过 aof 文件恢复能够正常；同时需要提醒，如果你的 redis 持久化手段中有 aof，那么在 server 故障失效后再次启动前，需要检测 aof 文件的完整性。
AOF 默认关闭，开启方法，修改配置文件 reds.conf：appendonly yes
##此选项为aof功能的开关，默认为“no”，可以通过“yes”来开启aof功能
##只有在“yes”下，aof重写/文件同步等特性才会生效
appendonly yes
##指定aof文件名称
appendfilename appendonly.aof
##指定aof操作中文件同步策略，有三个合法值：always everysec no,默认为everysec
appendfsync everysec
##在aof-rewrite期间，appendfsync是否暂缓文件同步，"no"表示“不暂缓”，“yes”表示“暂缓”，默认为“no”
no-appendfsync-on-rewrite no
##aof文件rewrite触发的最小文件尺寸(mb,gb),只有大于此aof文件大于此尺寸是才会触发rewrite，默认“64mb”，建议“512mb”
auto-aof-rewrite-min-size 64mb
##相对于“上一次”rewrite，本次rewrite触发时aof文件应该增长的百分比。
##每一次rewrite之后，redis都会记录下此时“新aof”文件的大小(例如A)，那么当aof文件增长到A*(1 + p)之后
##触发下一次rewrite，每一次aof记录的添加，都会检测当前aof文件的尺寸。
auto-aof-rewrite-percentage 100
AOF 是文件操作，对于变更操作比较密集的 server，那么必将造成磁盘 IO 的负荷加重；此外 linux 对文件操作采取了“延迟写入”手段，即并非每次 write 操作都会触发实际磁盘操作，而是进入了 buffer 中，当 buffer 数据达到阀值时触发实际写入(也有其他时机)，这是 linux 对文件系统的优化，但是这却有可能带来隐患，如果 buffer 没有刷新到磁盘，此时物理机器失效(比如断电)，那么有可能导致最后一条或者多条 aof 记录的丢失。通过上述配置文件，可以得知 redis 提供了 3 中 aof 记录同步选项：
-
always：每一条 aof 记录都立即同步到文件，这是最安全的方式，也以为更多的磁盘操作和阻塞延迟，是 IO 开支较大。
-
everysec：每秒同步一次，性能和安全都比较中庸的方式，也是 redis 推荐的方式。如果遇到物理服务器故障，有可能导致最近一秒内 aof 记录丢失(可能为部分丢失)。
-
no：redis 并不直接调用文件同步，而是交给操作系统来处理，操作系统可以根据 buffer 填充情况 / 通道空闲时间等择机触发同步；这是一种普通的文件操作方式。性能较好，在物理服务器故障时，数据丢失量会因 OS 配置有关。
触发 rewrite 的时机可以通过配置文件来声明，同时 redis 中可以通过 bgrewriteaof 指令人工干预。
redis-cli -h ip -p port bgrewriteaof
因为 rewrite 操作 /aof 记录同步 /snapshot 都消耗磁盘 IO，redis 采取了“schedule”策略：无论是“人工干预”还是系统触发，snapshot 和 rewrite 需要逐个被执行。
AOF rewrite 过程并不阻塞客户端请求。系统会开启一个子进程来完成。
Redis主从复制
概述
1、redis的复制功能是支持多个数据库之间的数据同步。一类是主数据库（master）一类是从数据库（slave），主数据库可以进行读写操作，当发生写操作的时候自动将数据同步到从数据库，而从数据库一般是只读的，并接收主数据库同步过来的数据，一个主数据库可以有多个从数据库，而一个从数据库只能有一个主数据库。
2、通过redis的复制功能可以很好的实现数据库的读写分离，提高服务器的负载能力。主数据库主要进行写操作，而从数据库负责读操作。
主从复制过程：见下图
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-g7L7sH1N-1607954996206)(img\3.png)]
执行过程：
1：当一个从数据库启动时，会向主数据库发送sync命令，
2：主数据库接收到sync命令后会开始在后台保存快照（执行rdb操作），并将保存期间接收到的命令缓存起来
3：当快照完成后，redis会将快照文件和所有缓存的命令发送给从数据库。
4：从数据库收到后，会载入快照文件并执行收到的缓存的命令。
修改从Redis从配置文件
# 设置主服务器地址和端口
# slaveof <masterip> <masterport> 3.x版本使用指令
replicaof <masterip> <masterport>
# 主Redis配置了密码,则需要配置
# masterauth 123456
可以使用info命令查看主从信息
info replication
Redis哨兵机制
什么是哨兵机制
Redis的哨兵(sentinel) 系统用于管理多个 Redis 服务器,该系统执行以下三个任务:
- 监控(Monitoring): 哨兵(sentinel) 会不断地检查你的Master和Slave是否运作正常。
- 提醒(Notification):当被监控的某个 Redis出现问题时, 哨兵(sentinel) 可以通过 API 向管理员或者其他应用程序发送通知。
- 自动故障迁移(Automatic failover):当一个Master不能正常工作时，哨兵(sentinel) 会开始一次自动故障迁移操作,它会将失效Master的其中一个Slave升级为新的Master, 并让失效Master的其他Slave改为复制新的Master; 当客户端试图连接失效的Master时,集群也会向客户端返回新Master的地址,使得集群可以使用Master代替失效Master。
哨兵(sentinel) 是一个分布式系统,你可以在一个架构中运行多个哨兵(sentinel) 进程,这些进程使用流言协议(gossipprotocols)来接收关于Master是否下线的信息,并使用投票协议(agreement protocols)来决定是否执行自动故障迁移,以及选择哪个Slave作为新的Master.
每个哨兵(sentinel) 会向其它哨兵(sentinel)、master、slave定时发送消息,以确认对方是否”活”着,如果发现对方在指定时间(可配置)内未回应,则暂时认为对方已挂(所谓的”主观认为宕机” Subjective Down,简称sdown).
若“哨兵群”中的多数sentinel,都报告某一master没响应,系统才认为该master"彻底死亡"(即:客观上的真正down机,Objective Down,简称odown),通过一定的vote算法,从剩下的slave节点中,选一台提升为master,然后自动修改相关配置.
虽然哨兵(sentinel) 释出为一个单独的可执行文件 redis-sentinel ,但实际上它只是一个运行在特殊模式下的 Redis 服务器，你可以在启动一个普通 Redis 服务器时通过给定 --sentinel 选项来启动哨兵(sentinel).
哨兵(sentinel) 的一些设计思路和zookeeper非常类似，如下图所示。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-6JEyVmFa-1607954996207)(img\4.png)]
哨兵模式修改配置
实现步骤:
1 拷贝到一份sentinel.conf
cp sentinel.conf /usr/local/redis/redis.6.0.5/sentinel-1/sentinel.conf
2 修改sentinel.conf配置文件
#根据情况调整配置
#端口
port 26379
#日志文件配置
logfile "/var/run/redis-sentinel-1.log"
#进程号文件配置
pidfile /var/run/redis-sentinel-1.pid
#工作目录
dir "/usr/local/redis/redis.6.0.5/sentinel-1"
#固定配置
#后台启动
daemonize yes
#主节点 名称 IP 端口号 选举次数(需要几个哨兵同意则认为主服务器失效)
sentinel monitor mymaster 192.168.220.128 6379 1
#主节点密码
#sentinel auth-pass mymaster 123456
#修改心跳检测 5000毫秒
sentinel down-after-milliseconds mymaster 5000
#在执行故障转移时， 最多可以有多少个从服务器同时对新的主服务器进行同步
sentinel parallel-syncs mymaster 2
#进行故障转移时如果超过了配置的<times>时间就表示故障转移超时失败
sentinel failover-timeout mymaster 15000
3 启动哨兵模式
redis-sentinel /usr/local/redis/redis.6.0.5/sentinel-1/sentinel.conf
4 停止哨兵模式
redis-cli -p [Sentinel PORT] -h [Sentinel IP] shutdown
SpringBoot集成主从复制
只需要修改Redis配置即可
#配置redis连接
#选择数据库,默认值为0
spring.redis.database=0
#密码
spring.redis.password=
#请求超时时间
spring.redis.timeout=0
#配置jedis连接池
spring.redis.jedis.pool.max-active=8
spring.redis.jedis.pool.max-wait=-1
spring.redis.jedis.pool.max-idle=8
spring.redis.jedis.pool.min-idle=0
#设置哨兵监听主服务器
spring.redis.sentinel.master=mymaster
#设置哨兵群，多个哨兵使用,号分割
spring.redis.sentinel.nodes=192.168.220.128:26379
Redis事务
Redis 事务可以一次执行多个命令， 并且带有以下两个重要的保证：
事务是一个单独的隔离操作：事务中的所有命令都会序列化、按顺序地执行。事务在执行的过程中，不会被其他客户端发送来的命令请求所打断。
事务是一个原子操作：事务中的命令要么全部被执行，要么全部都不执行。
一个事务从开始到执行会经历以下三个阶段：
-
开始事务。
-
命令入队。
-
执行事务。
以下是一个事务的例子， 它先以 MULTI 开始一个事务， 然后将多个命令入队到事务中， 最后由 EXEC 命令触发事务， 一并执行事务中的所有命令：
#开启事务
MULTI
#指令1
SET book-name "Mastering C++ in 21 days"
#指令2
GET book-name
#指令3
SADD tag "C++" "Programming" "Mastering Series"
#执行事务
EXEC
Redis 缓存相关问题
缓存穿透
缓存穿透是指查询一个数据库一定不存在的数据。
我们以前正常的使用Redis缓存的流程大致是：
1、数据查询首先进行缓存查询
2、如果数据存在则直接返回缓存数据
3、如果数据不存在，就对数据库进行查询，并把查询到的数据放进缓存
4、如果数据库查询数据为空，则不放进缓存
例如， 我们的数据表中主键是自增产生的，所有的主键值都大于0。此时如果用户传入的参数为-1，会是怎么样的？
这个-1，就是一定不存在的对象。程序就会每次都去查询数据库，而每次查询都是空，每次又都不会进行缓存。假如有人恶意攻击，就可以利用这个漏洞，对数据库造成压力，甚至压垮我们的数据库。
解决方案：
（1）利用互斥锁，缓存失效的时候，先去获得锁，得到锁了，再去请求数据库。没得到锁，则休眠一段时间重试。
使用到的相关指令：
- SETNX：将key设置值为value，如果key不存在，这种情况下等同SET命令。 当key存在时，什么也不做
- GETSET：先查询出原来的值，值不存在就返回nil。然后再设置值
定义锁组件
@Component
@Slf4j
public class RedisLock {
@Autowired
private RedisTemplate redisTemplate;
/**
* 加锁
* @param key 如商品的唯一标志 1
* @param value 当前时间+超时时间 1号线程 10:50:01 2 10:50:01
* @return
*/
public boolean lock(String key,String value){
//表示第一个拿到锁
if(redisTemplate.opsForValue().setIfAbsent(key,value)){//对应setnx命令
//可以成功设置,也就是key不存在
return true;
}
//判断锁超时 - 防止原来的操作异常，没有运行解锁操作 防止死锁
String currentValue = (String) redisTemplate.opsForValue().get(key);
//如果锁过期
if(!StringUtils.isEmpty(currentValue) && Long.parseLong(currentValue) < System.currentTimeMillis()){//currentValue不为空且小于当前时间
//获取上一个锁的时间value
String oldValue = (String)redisTemplate.opsForValue().getAndSet(key,value);//对应getset，如果key存在
//假设两个线程同时进来，key被占用了。获取的值currentValue=A(get取的旧的值肯定是一样的),两个线程的value都是B,key都是K.锁时间已经过期了。
//而这里面的getAndSet一次只会一个执行，也就是一个执行之后，上一个的value已经变成了B。只有一个线程获取的上一个值会是A，另一个线程拿到的值是B。
if(!StringUtils.isEmpty(oldValue) && oldValue.equals(currentValue) ){
//oldValue不为空且oldValue等于currentValue，也就是校验是不是上个对应的商品时间戳，也是防止并发
return true;
}
}
//无锁
return false;
}
/**
* 解锁
* @param key
* @param value
*/
public void unlock(String key,String value){
try {
String currentValue = (String)redisTemplate.opsForValue().get(key);
if(!StringUtils.isEmpty(currentValue) && currentValue.equals(value) ){
redisTemplate.opsForValue().getOperations().delete(key);//删除key
}
} catch (Exception e) {
log.error("[Redis分布式锁] 解锁出现异常了，{}",e);
}
}
}
使用流程流程
public void orderProductMocckDiffUser(String productId) {
//1 加锁
int tryTimes = 3;
long time = -1;
do{
time = System.currentTimeMillis() + TIMEOUT;
if(!redisLock.lock(productId,String.valueOf(time))){
//休眠1s
Thread.sleep(1000);
}else{
//加锁成功
break;
}
time = -1;
tryTimes --;
}while(tryTimes > 0);
if(time != -1){
//2 省略数据处理流程
//3 解锁
redisLock.unlock(productId,String.valueOf(time));
}
}
（2）采用异步更新策略，无论 key 是否取到值，都直接返回。value 值中维护一个缓存失效时间，缓存如果过期，异步起一个线程去读数据库，更新缓存。需要做缓存预热(项目启动前，先加载缓存)操作。
（3）提供一个能迅速判断请求是否有效的拦截机制（将key缓存起来）。迅速判断出，请求所携带的 Key 是否合法有效。如果不合法，则直接返回。
缓存雪崩
缓存雪崩，是指在某一个时间段，缓存集中过期失效。在缓存集中失效的这个时间段对数据的访问查询，都落到了数据库上，对于数据库而言，就会产生周期性的压力波峰。为了避免缓存雪崩的发生，我们可以将缓存的数据设置不同的失效时间，这样就可以避免缓存数据在某个时间段集中失效。
解决方案：
（1）给缓存的失效时间，加上一个随机值，避免集体失效。
（2）使用互斥锁，但是该方案吞吐量明显下降了。
（3）双缓存。我们有两个缓存，缓存 A 和缓存 B。缓存 A 的失效时间为 20 分钟，缓存 B 不设失效时间。自己做缓存预热操作。然后细分以下几个小点：
- 从缓存 A 读数据库，有则直接返回
- A 没有数据，直接从 B 读数据，直接返回，并且异步启动一个更新线程。
- 更新线程同时更新缓存 A 和缓存 B。
缓存击穿
缓存击穿，是指一个key非常热点（例如双十一期间进行抢购的商品数据），在不停的扛着高并发，高并发集中对这一个点进行访问，当这个key在失效的瞬间，持续的高并发就穿破缓存，直接请求到数据库上，就像在一个屏障上凿开了一个洞。
我们同样可以将这些热点数据设置永不过期就可以解决缓存击穿的问题了。
自动过期设定
设置过期时间
expire(key,有效时间(秒))
expireAt(key,系统时间毫秒数)
获取过期时间
TTL(key)
映射到Redistemplate中的方法
设置过期时间
//在指定时间过期
redisTemplate.expireAt("user2", Instant.ofEpochMilli(System.currentTimeMillis() + 10000));
//设置缓存有效时间
redisTemplate.expire("user2",Duration.ofSeconds(20L));
获取过期时间
Long user21 = redisTemplate.getExpire("user2");
内容总结
以上是互联网集市为您收集整理的Redis进阶（二）个人参考使用全部内容，希望文章能够帮你解决Redis进阶（二）个人参考使用所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。