Redis入门到入坑（二）

Redis 数据持久化

简介

背景

Redis是一种内存数据库，在断电时数据可能会丢失。比如你redis整个挂了，然后redis不可用了，如果没有持久化的话，redis就会丢失所有的数据，如果通过持久化将数据搞一份儿到磁盘上去，然后再定期同步到一些云存储服务上去，那么就可以保证一些数据不丢失，保证数据的可靠性。

持久化方式

Redis中为了保证在系统宕机(类似进程被杀死)情况下，能更快的进行故障恢复，设计了两种数据持久化方案，分别为rdb和aof方式。

Rdb方式持久化

概述

Rdb方式是通过手动(save-阻塞式，bgsave-异步)或周期性方式保存redis中key/value的一种机制,Rdb方式一般为redis的默认数据持久化方式.系统启动时会自动开启这种方式的持久化机制。

RDB方式配置

RDB方式的持久化是默认开启的，也可按规则自己配置，例如，打开redis.conf文件，例如

# 这里表示每隔60s，如果有超过1000个key发生了变更，那么就生成一个新的dump.rdb文件，就是当前redis内存中完整的数据快照，这个操作也被称之为snapshotting(快照)。
save 60 1000
# 持久化 rdb文件遇到问题时，主进程是否接受写入，yes 表示停止写入，如果是no 表示redis继续提供服务。
stop-writes-on-bgsave-error yes    
# 在进行快照镜像时,是否进行压缩。yes:压缩，但是需要一些cpu的消耗。no:不压缩，需要更多的磁盘空间。
rdbcompression yes
# 一个CRC64的校验就被放在了文件末尾，当存储或者加载rbd文件的时候会有一个10%左右的性能下降，为了达到性能的最大化，你可以关掉这个配置项。
rdbchecksum yes
# 快照的文件名
dbfilename dump.rdb
# 存放快照的目录
dir /var/lib/redis

Rdb方式持久化实践

试验一

在redis中保存几条数据，然后执行shutdown关闭redis,然后再重启redis，看看刚才插入的数据是否还在？假如数据还在，为什么？
因为,通过redis-cli shutdown这种方式去停掉redis，其实是一种安全退出的模式，redis在退出的时候会将内存中的数据立即生成一份完整的rdb快照,例如

127.0.0.1:6379> set phone 11111111
OK
127.0.0.1:6379> shutdown   #默认也会进行持久化
[root@centos7964 ~]#  docker start redis01
[root@centos7964 ~]# docker exec -it redis01 redis-cli
127.0.0.1:6379> keys *
1) "pone"

试验二

在redis中再保存几条新的数据，用kill -9粗暴杀死redis进程，模拟redis故障异常退出，导致内存数据丢失的场景？
这次就发现，redis进程异常被杀掉，几条最新的数据就丢失了。例如:

首先,打开第一个客户端,先清除redis内存和磁盘对应的数据

[root@centos7964 data]# docker exec -it redis01 redis-cli
127.0.0.1:6379> flushall
OK
127.0.0.1:6379> exit
[root@centos7964 data]# ls
dump.rdb
[root@centos7964 data]# rm –f dump.rdb
[root@centos7964 data]# ls

然后,打开并登录第二个客户端,并向redis存储一些数据,例如

[root@centos7964 ~]# docker exec -it redis01 redis-cli
127.0.0.1:6379> set one mybatis
OK
127.0.0.1:6379> set two spring
OK
127.0.0.1:6379> keys *
1) "one"
2) "two"

接下来,再次回到第一个客户端,杀掉redis进程,例如

[root@centos7964 data]# ps -ef | grep redis
polkitd    6995   6974  0 14:44 ?        00:00:00 redis-server *:6379
root       7064   6974  0 14:44 pts/0    00:00:00 redis-cli
root       7111   6467  0 14:47 pts/1    00:00:00 docker exec -it redis01 redis-cli
root       7130   6974  0 14:47 pts/1    00:00:00 redis-cli
root       7278   7180  0 14:51 pts/0    00:00:00 grep --color=auto redis
[root@centos7964 data]# kill -9 6995
[root@centos7964 data]# docker start redis01

最后,打开第一个客户端,登录redis,检查key是否还存在.

[root@centos7964 ~]# docker exec -it redis01 redis-cli
127.0.0.1:6379> keys *
(empty array)
127.0.0.1:6379>
[root@centos7964 ~]#

试验三

手动调用save(同步保存)或者bgsave(异步保存)执行rdb快照生成.然后杀掉redis进程,再重启检测是否还有刚刚保存的数据.

127.0.0.1:6379> set id 100
OK
127.0.0.1:6379> set name jack
OK
127.0.0.1:6379> save  #阻塞式持久化
OK
127.0.0.1:6379> set address beijing
OK
127.0.0.1:6379> bgsave  #异步方式持久化
Background saving started

小节面试分析

Redis中的save和bgsave有什么不同？

  • Redis Save 命令执行一个同步保存操作，将当前 Redis 实例的所有数据快照(snapshot)以 RDB 文件的形式保存到硬盘。
  • BGSAVE 命令执行之后立即返回 OK ，然后 Redis fork 出一个新子进程，原来的 Redis 进程(父进程)继续处理客户端请求，而子进程则负责将数据保存到磁盘，然后退出。

RDB持久化机制有哪些优点？

第一：RDB会生成多个数据文件，每个数据文件都代表了某一个时刻中redis的数据，这种多个数据文件的方式，非常适合做冷备，可以将这种完整的数据文件发送到一些远程云服务上去，在国内可以是阿里云的ODPS分布式存储上，以预定好的备份策略来定期备份redis中的数据.
第二：RDB对redis对外提供的读写服务，影响非常小，可以让redis保持高性能，因为redis主进程只需要fork一个子进程，让子进程执行磁盘IO操作来进行RDB持久化即可。
第三：相对于AOF持久化机制来说，直接基于RDB数据文件来重启和恢复redis进程，更加快速。

RDB持久化机制有哪些缺点？

假如redis故障时，要尽可能少的丢失数据，那么RDB方式不太好，它都是每隔5分钟或更长时间做一次快照，这个时候一旦redis进程宕机，那么会丢失最近几分钟的数据。

Aof方式数据持久化

概述

Aof方式是通过记录写操作日志的方式,记录redis数据的一种持久化机制,这个机制默认是关闭的。

AOF方式配置

# 是否开启AOF，默认关闭
appendonly yes
# 指定 AOF 文件名
appendfilename appendonly.aof
# Redis支持三种刷写模式：
# appendfsync always #每次收到写命令就立即强制写入磁盘，类似MySQL的sync_binlog=1,是最安全的。但该模式下速度也是最慢的，一般不推荐使用。
appendfsync everysec #每秒钟强制写入磁盘一次，在性能和持久化方面做平衡，推荐该方式。
# appendfsync no     #完全依赖OS的写入，一般为30秒左右一次，性能最好但是持久化最没有保证，不推荐。    
#在日志重写时，不进行命令追加操作，而只是将其放在缓冲区里，避免与命令的追加造成DISK IO上的冲突。
#设置为yes表示rewrite期间对新写操作不fsync,暂时存在内存中,等rewrite完成后再写入，默认为no，建议yes
no-appendfsync-on-rewrite yes
#当前AOF文件大小是上次日志重写得到AOF文件大小的二倍时，自动启动新的日志重写过程。
auto-aof-rewrite-percentage 100
#当前AOF文件启动新的日志重写过程的最小值，避免刚刚启动Reids时由于文件尺寸较小导致频繁的重写。
auto-aof-rewrite-min-size 64mb

AOF方式持久化实践

第一：打开AOF的开关，启用AOF持久化
第二：写入一些数据，观察AOF文件（appendonly.aof）中的日志内容
第三：kill -9杀掉redis进程，重新启动redis进程，发现数据被恢复回来了，就是从AOF文件中恢复回来的，redis进程启动的时候，直接就会从appendonly.aof中加载所有的日志，把内存中的数据恢复回来。

小节面试分析

如何理解AOF方式中的rewrite操作？

redis中的数据其实有限的，很多数据可能会自动过期，可能会被用户删除，可能会被redis用缓存清除的算法清理掉。也就是说redis中的数据会不断淘汰掉旧的，只有一部分常用的数据会被自动保留在redis内存中，所以可能很多之前的已经被清理掉的数据，对应的写日志还停留在AOF中，AOF日志文件就一个，会不断的膨胀，到很大很大。
所以,AOF会自动在后台每隔一定时间做rewrite操作，比如日志里已经存放了针对100w数据的写日志了; redis内存只剩下10万; 基于内存中当前的10万数据构建一套最新的日志，到AOF中; 覆盖之前的老日志; 确保AOF日志文件不会过大，保持跟redis内存数据量一致.

AOF持久化机制有哪些优点？

第一：AOF可以更好的保护数据不丢失，一般AOF会每隔1秒，通过一个后台线程执行一次fsync操作，最多丢失1秒钟的数据.
第二：AOF日志文件以append-only模式写入，所以没有任何磁盘寻址的开销，写入性能非常高，而且文件不容易破损，即使文件尾部破损，也很容易修复。
第三：AOF日志文件即使过大的时候，出现后台重写操作，也不会影响客户端的读写。因为在rewrite log的时候，会对其中的指导进行压缩，创建出一份需要恢复数据的最小日志出来。再创建新日志文件的时候，老的日志文件还是照常写入。当新的merge后的日志文件ready的时候，再交换新老日志文件即可。
第四：AOF日志文件的命令通过易读的方式进行记录，这个特性非常适合做灾难性的误删除的紧急恢复。比如某人不小心用flushall命令清空了所有数据，只要这个时候后台rewrite还没有发生，那么就可以立即拷贝AOF文件，将最后一条flushall命令给删了，然后再将该AOF文件放回去，就可以通过恢复机制，自动恢复所有数据.

AOF持久化机制有哪些缺点？

第一：对于同一份数据来说，AOF日志文件通常比RDB数据快照文件更大。
第二：AOF开启后，支持的写QPS会比RDB支持的写QPS低，因为AOF一般会配置成每秒fsync一次日志文件，当然，每秒一次fsync，性能也还是很高的。
第三：AOF这种基于命令日志方式，比基于RDB每次持久化一份完整的数据快照文件的方式，更加脆弱一些，容易有bug。不过AOF为了避免rewrite过程导致的bug，因此每次rewrite并不是基于旧的指令日志进行merge的，而是基于当时内存中的数据进行指令的重新构建，这样健壮性会好很多。

如何选择redis的持久化方式？

第一：不要仅仅使用RDB，因为那样会导致你丢失很多数据。
第二：也不要仅仅使用AOF，因为AOF做冷备没有RDB做冷备进行数据恢复的速度快,并且RDB简单粗暴的数据快照方式更加健壮。
第三：综合使用AOF和RDB两种持久化机制，用AOF来保证数据不丢失，作为数据恢复的第一选择; 用RDB来做不同程度的冷备。

Redis事务处理

简介

概述

Redis采用了乐观锁方式进行事务控制，它使用watch命令监视给定的key，当exec（提交事务）的时候,如果监视的key从调用watch后发生过变化，则整个事务会失败。也可以调用watch多次监视多个key。注意watch的key是对整个连接有效的，如果连接断开，监视和事务都会被自动清除。当然exec，discard，unwatch命令都会清除连接中的所有监视。

基本指令

redis进行事务控制时，通常是基于如下指令进行实现，例如：

  • multi 开启事务
  • exec 提交事务
  • discard 取消事务
  • watch 监控，如果监控的值发生变化，则提交事务时会失败
  • unwatch 去掉监控

Redis保证一个事务中的所有命令要么都执行，要么都不执行(原子性)。如果在发送EXEC命令前客户端断线了，则Redis会清空事务队列，事务中的所有命令都不会执行。而一旦客户端发送了EXEC命令，所有的命令就都会被执行，即使此后客户端断线也没关系，因为Redis中已经记录了所有要执行的命令。

Redis事务控制实践

exec提交事务

例如：模拟转账，tony 500，jack 200，tony转给jack100。过程如下：

127.0.0.1:6379> set tony 500
OK
127.0.0.1:6379> set jack 200
OK
127.0.0.1:6379> mget tony jack
1) "500"
2) "200"
127.0.0.1:6379> multi #开启事务
OK
127.0.0.1:6379(TX)> decrby tony 100 #所有指令操作会进入到队列
QUEUED
127.0.0.1:6379(TX)> incrby jack 100
QUEUED
127.0.0.1:6379(TX)> mget tony jack
QUEUED 
127.0.0.1:6379(TX)> exec  #提交事务
1) (integer) 400
2) (integer) 300
3) 1) "400"
   2) "300"
127.0.0.1:6379> mget tony jack
1) "400"
2) "300"
discard取消事务

注意redis事务太简单，没有回滚，而只有取消。

127.0.0.1:6379> mget tony jack
1) "400"
2) "300"
127.0.0.1:6379> multi
OK
127.0.0.1:6379> incrby jack 100
QUEUED
127.0.0.1:6379> discard
OK
127.0.0.1:6379> get jack
"300"
127.0.0.1:6379> exec
(error) ERR EXEC without MULTI

当出现错误指令时，事务也会自动取消。

127.0.0.1:6379> mget tony jack
1) "400"
2) "300"
127.0.0.1:6379> multi
OK
127.0.0.1:6379(TX)> incrby jack 100
QUEUED
127.0.0.1:6379(TX)> abcd
(error) ERR unknown command `abcd`, with args beginning with:
127.0.0.1:6379(TX)> get jack
QUEUED
127.0.0.1:6379(TX)> exec
(error) EXECABORT Transaction discarded because of previous errors.
127.0.0.1:6379> get jack
"300"
秒杀抢票事务处理

基于一个秒杀,抢购案例,演示redis乐观锁方式,例如

第一步:打开客户端1,执行如下操作

127.0.0.1:6379> set ticket 1
OK
127.0.0.1:6379> set money 0
OK
127.0.0.1:6379> watch ticket		#乐观锁，对值进行观察，改变则事务失败
OK
127.0.0.1:6379> multi				#开启事务
OK
127.0.0.1:6379> decr ticket
QUEUED
127.0.0.1:6379> incrby money 100
QUEUED

第二步:打开客户端2,执行如下操作,演示还没等客户端1提交事务，此时客户端2把票买到了。

127.0.0.1:6379> get ticket
"1"
127.0.0.1:6379> decr ticket
(integer) 0

第三步,回到客户端1：提交事务,检查ticket的值

127.0.0.1:6379> exec
(nil) #执行事务，失败
127.0.0.1:6379> get ticket
“0”
127.0.0.1:6379> unwatch #取消监控
Jedis 客户端事务操作

基于Jedis进行事务测试,代码如下:

package com.jt;
import org.junit.Test;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Transaction;

/**
 * Redis中的事务测试
 * 什么是事务？事务是一个工作单元，是一个业务，一般要求这个是五所有的操作要么全部执行，要么全部执行
 * 为什么要使用事务(使用事务的目的是什么)？保证数据正确
 * 事务基于什么特性保证数据的正确？ACID
 * redis 中的事务指令：
 *      1.multi(手动开启事务)
 *      2.discard(手动取消事务)
 *      3.exec(手动提交事务)
 */
public class JedisTransactionTests {
    @Test
    public void textTransaction(){
        //1.创建redis链接对象
        Jedis jedis = new Jedis("192.168.126.129",6379);
        //2.定义业务初始数据
        jedis.set("tony", "1000");
        jedis.set("jack", "300");
        //3.开启事务
        Transaction multi = jedis.multi();
        //4.执行业务(tony给jack转100)
        try {
            multi.decrBy("tony", 100);
            multi.incrBy("jack", 100);
            //int num = 100/0;//模拟异常
            //5.提交事务
            multi.exec();
        }catch (Exception e){
            //出现异常取消事务
            multi.discard();
        }
        String tonyMoney=jedis.get("tony");
        String jackMoney=jedis.get("jack");
        System.out.println("tonyMoney="+tonyMoney);
        System.out.println("jackMoney="+jackMoney);
        //6.释放资源
        jedis.close();
    }
}
Jedis 客户端秒杀操作实践
package com.jt.redis;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Transaction;
import java.util.List;

/**
 * 基于redis实现一个简单的多线程抢票操作
 * 这个操作中重点演示一下乐观锁的应用
 * 乐观锁：允许多个线程同时对一条记录修改，但只能有一个线程修改成功
 */
public class SecondsKillDemo02 {
    static final String IP = "192.168.126.129";
    static final int PORT = 6379;
    
    public static void secKill(){
        //1.建立redis链接
        Jedis jedis = new Jedis(IP,PORT);
        //2.监控redis中的key
        jedis.watch("ticket","money");
        String ticket = jedis.get("ticket");
        if(ticket == null || Integer.valueOf(ticket)==0)
            throw new RuntimeException("余票不足！");
        //3.开启事务并执行业务
        Transaction multi = jedis.multi();
        try {
            multi.decr("ticket");
            multi.incrBy("money", 100);
            //4.提交事务
            List<Object> exec = multi.exec();
            System.out.println(exec);//假如输出的是null，表示此事务没有对数据进行修改
            System.out.println("transaction ok！");
        }catch (Exception e){
            e.printStackTrace();
            multi.discard();
        }finally {
            //5.取消监控
            jedis.unwatch();
            //6.释放资源
            jedis.close();
        }
    }
    
    public static void main(String[] args) throws InterruptedException {
        //1.定义redis并初始化数据
        Jedis jedis = new Jedis(IP,PORT);
        jedis.set("ticket", "1");
        jedis.expire("ticket", 2);//ticket有效时间是2秒
        jedis.set("money", "0");
        jedis.close();
        Thread.sleep(1000);
        //2.创建多个线程，在线程中执行抢票操作
        Thread t1 = new Thread(){
            @Override
            public void run() {
                //执行抢票操作
                secKill();
            }
        };
        Thread t2 = new Thread(){
            @Override
            public void run() {
                //执行抢票操作
                secKill();
            }
        };
        t1.start();
        t2.start();
    }
}

Redis架构设计

Redis主从复制

简介

单个Redis支持的读写能力还是有限的,此时我们可以使用多个redis来提高redis的并发处理能力,这些redis如何协同,就需要有一定的架构设计,这里我们首先从主从(Master/Slave)架构进行分析和实现.

基本架构

redis主从架构如图所示:
在这里插入图片描述
其中,master负责读写,并将数据同步到slave,从节点负责读操作.

主从快速入门

基于redis,设计一主从架构,一个Master,两个Slave,其中Master负责Redis读写操作,并将数据同步到Slave,Slave只负责读.,其步骤如下:

第一：将redis01拷贝两份，例如:

cp -r redis01/ redis02

cp -r redis01/ redis03

第二步：假如已有redis服务，先将原先所有redis服务停止(docker rm -f redis容器名)，并启动新的redis容器，例如：

docker run -p 6379:6379 --name redis6379 \
-v /usr/local/docker/redis01/data:/data \
-v /usr/local/docker/redis01/conf/redis.conf:/etc/redis/redis.conf \
-d redis redis-server /etc/redis/redis.conf \
--appendonly yes

docker run -p 6380:6379 --name redis6380 \
-v /usr/local/docker/redis02/data:/data \
-v /usr/local/docker/redis02/conf/redis.conf:/etc/redis/redis.conf \
-d redis redis-server /etc/redis/redis.conf \
--appendonly yes

docker run -p 6381:6379 --name redis6381 \
-v /usr/local/docker/redis03/data:/data \
-v /usr/local/docker/redis03/conf/redis.conf:/etc/redis/redis.conf \
-d redis redis-server /etc/redis/redis.conf \
--appendonly yes

第三步 检测redis服务角色

启动三个客户端,分别登陆三台redis容器服务，通过info指令进行角色查看，默认新启动的三个redis服务角色都为master.

127.0.0.1:6379> info replication

\# Replication
role:master
connected_slaves:0
master_repl_offset:3860
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:2
repl_backlog_histlen:3859

第四步：检测redis6379的ip设置

docker inspect redis6379

……
"Networks": {
    "bridge": {
        "IPAMConfig": null,
        "Links": null,
        "Aliases": null,
        "NetworkID": "c33071765cb48acb1efed6611615c767b04b98e6e298caa0dc845420e6112b73",
        "EndpointID": "4c77e3f458ea64b7fc45062c5b2b3481fa32005153b7afc211117d0f7603e154",
        "Gateway": "172.17.0.1",
        "IPAddress": "172.17.0.2",
        "IPPrefixLen": 16,
        "IPv6Gateway": "",
        "GlobalIPv6Address": "",
        "GlobalIPv6PrefixLen": 0,
        "MacAddress": "02:42:ac:11:00:02",
        "DriverOpts": null
    }
}

第五步：设置Master/Slave架构

分别登陆redis6380/redis6381，然后执行如下语句

slaveof ip port # ip:主节点的IP地址，port:主节点的端口号

slaveof 172.17.0.2 6379

说明,假如master有密码,需要在slave的redis.conf配置文件中添加"masterauth 你的密码"这条语句,然后重启redis再执行slaveof 指令操作.

第六步：再次登陆redis6379，然后检测info

[root@centos7964 ~]# docker exec -it redis6379 redis-cli
127.0.0.1:6379> info replication

\# Replication
role:master
connected_slaves:2
slave0:ip=172.17.0.3,port=6379,state=online,offset=2004,lag=1
slave1:ip=172.17.0.4,port=6379,state=online,offset=2004,lag=1
master_failover_state:no-failover
master_replid:5baf174fd40e97663998abf5d8e89a51f7458488
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:2004
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:2004

第七步: 登陆redis6379测试，master读写都可以

[root@centos7964 ~]# docker exec -it redis6379 redis-cli
127.0.0.1:6379> set role master6379
OK
127.0.0.1:6379> get role
"master6379"

第八步: 登陆redis6380/6381测试，slave只能读，不能写。

[root@centos7964 ~]# docker exec -it redis6380 redis-cli
127.0.0.1:6379> get role
"master6379"
127.0.0.1:6379> set role slave6380
(error) READONLY You can't write against a read only replica.

Java中的读写测试分析，代码如下：

package com.jt;
import org.junit.Test;
import redis.clients.jedis.Jedis;

/**
 * 主从架构结构分析
 */
public class MasterSlaveTests {
    @Test //Master节点(支持读写操作)
    public void testWriteRead(){
        Jedis jedis = new Jedis("192.168.126.129",6379);
        jedis.set("a1", "100");
        String a1 = jedis.get("a1");
        System.out.println(a1);
        jedis.close();
    }
    
    @Test //Slave节点(只支持读操作)
    public void testRead(){
        Jedis jedis = new Jedis("192.168.126.129",6380);
        //jedis.set("a1", "200");//这里不允许写操作
        String a1 = jedis.get("a1");
        System.out.println(a1);
        jedis.close();
    }
}
//可靠性

主从同步原理分析

Redis的主从结构可以采用一主多从结构，Redis主从复制可以根据是否是全量分为全量同步和增量同步。

  • Redis全量同步

Redis全量复制一般发生在Slave初始化阶段，这时Slave需要将Master上的所有数据都复制一份。具体步骤如下：
1）从服务器连接主服务器，发送sync命令；
2）主服务器接收到sync命名后，开始执行bgsave命令生成rdb文件并使用缓冲区记录此后执行的所有写命令；
3）主服务器bgsave执行完后，向所有从服务器发送快照文件，并在发送期间继续记录被执行的写命令；
4）从服务器收到快照文件后丢弃所有旧数据，载入收到的快照；
5）主服务器快照发送完毕后开始向从服务器发送缓冲区中的写命令；
6）从服务器完成对快照的载入，开始接收命令请求，并执行来自主服务器缓冲区的写命令；

  • Redis增量同步

Redis增量复制是指Slave初始化后，开始正常工作时主服务器发生的写操作同步到从服务器的过程。 增量复制的过程主要是主服务器每执行一个写命令就会向从服务器发送相同的写命令，从服务器接收并执行收到的写命令。

小节面试分析
  • 如果redis要支持10万+的的并发你会怎么做？

单机的redis几乎不太可能说QPS超过10万+，除非一些特殊情况，比如你的机器性能特别好，配置特别高，物理机，维护做的特别好，而且你的整体的操作不是太复杂，一般的单机也就在几万。真正实现redis的高并发，需要读写分离。对缓存而言，一般都是用来支撑读高并发的，写的请求是比较少的，可能写请求也就一秒钟几千。读的请求相对就会比较多,例如,一秒钟二十万次读。所以redis的高并发可以基于主从架构与读写分离机制进行实现。

  • Redis的replication机制是怎样的？

（1）redis采用异步方式复制数据到slave节点。
（2）一个master node是可以配置多个slave node的。
（3）slave node也可以连接其他的slave node。
（4）slave node做复制的时候，是不会block master node的正常工作的。
（5）slave node在做复制的时候，也不会block对自己的查询操作，它会用旧的数据集来提供服务; 但是复制完成的时候，需要删除旧数据集，加载新数据集，这个时候就会暂停对外服务了。
（6）slave node主要用来进行横向扩容，做读写分离，扩容的slave node可以提高读的吞吐量。

Redis哨兵模式

简介

哨兵（Sentinel）是Redis的主从架构模式下，实现高可用性（high availability）的一种机制。
由一个或多个Sentinel实例（instance）组成的Sentinel系统（system）可以监视任意多个主服务器，以及这些主服务器属下的所有从服务器，并在被监视的主服务器进入下线状态时，自动将下线主服务器属下的某个从服务器升级为新的主服务器，然后由新的主服务器代替已下线的主服务器继续处理命令请求。

基本架构

在这里插入图片描述

哨兵快速入门

第一步：分别进入3台redis容器内部进行配置，在容器指定目录/etc/redis创建sentinel.conf文件,文件内容为：

sentinel monitor redis6379 172.17.0.2 6379 1

其中, 如上指令表示要的监控的master, redis6379为服务名, 172.17.0.2和6379为master的ip和端口,1表示多少个sentinel认为一个master失效时，master才算真正失效.

假如，这里如果出现 bash: vi: command not found，可依次执行如下两个指令来安装vim

apt-get update 
apt-get install vim

说明:假如网络不好,也可以在宿主机对应的挂载目录去创建文件,或者在容器中执行:

cat <<EOF > /etc/redis/sentinel.conf 
sentinel monitor redis6379 172.17.0.2 6379 1
EOF

在这里插入图片描述
第二步：在每个redis容器内部的/etc/redis目录下(上图的目录)执行如下指令（最好是在多个客户端窗口执行），启动哨兵服务

redis-sentinel sentinel.conf

第三步：打开一个新的客户端连接窗口，关闭redis6379服务（这个服务是master服务）

docker stop redis6379

其它客户端窗口，检测日志输出(默认等待30秒)，例如

410:X 11 Jul 2021 09:54:27.383 # +switch-master redis6379 172.17.0.2 6379 172.17.0.4 6379
410:X 11 Jul 2021 09:54:27.383 * +slave slave 172.17.0.3:6379 172.17.0.3 6379 @ redis6379 172.17.0.4 6379
410:X 11 Jul 2021 09:54:27.383 * +slave slave 172.17.0.2:6379 172.17.0.2 6379 @ redis6379 172.17.0.4 6379

第四步：登陆ip为172.17.0.4对应的服务进行info检测，例如：

127.0.0.1:6379> info replication

\# Replication
role:master
connected_slaves:1
slave0:ip=172.17.0.3,port=6379,state=online,offset=222807,lag=0
master_failover_state:no-failover
master_replid:3d63e8474dd7bcb282ff38027d4a78c413cede53
master_replid2:5baf174fd40e97663998abf5d8e89a51f7458488
master_repl_offset:222807
second_repl_offset:110197
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:29
repl_backlog_histlen:222779
127.0.0.1:6379>

从上面的信息输出发现,redis6381服务现在已经由slave变为master。

Sentinel 配置进阶

对于sentinel.conf文件中的内容,我们还可以基于实际需求,进行增强配置,例如:

sentinel monitor redis6379 172.17.0.2 6379 1 
daemonize yes #后台运行
logfile "/var/log/sentinel_log.log" #运行日志
sentinel down-after-milliseconds redis6379 30000 #默认30秒

其中:
1)daemonize yes表示后台运行(默认为no)
2)logfile 用于指定日志文件位置以及名字
3)sentinel down-after-milliseconds 表示master失效了多长时间才认为失效

例如: 基于cat指令创建sentinel.conf文件,并添加相关内容.

cat <<EOF > /etc/redis/sentinel.conf
sentinel monitor redis6379 172.17.0.2 6379 1
daemonize yes 
logfile "/var/log/sentinel_log.log"
sentinel down-after-milliseconds redis6379 30000 
EOF

哨兵工作原理分析

1)：每个Sentinel以每秒钟一次的频率向它所知的Master，Slave以及其他 Sentinel 实例发送一个 PING 命令。

2)：如果一个实例（instance）距离最后一次有效回复 PING 命令的时间超过 down-after-milliseconds 选项所指定的值(这个配置项指定了需要多少失效时间，一个master才会被这个sentinel主观地认为是不可用的。 单位是毫秒，默认为30秒)， 则这个实例会被 Sentinel 标记为主观下线。

3)：如果一个Master被标记为主观下线，则正在监视这个Master的所有 Sentinel 要以每秒一次的频率确认Master的确进入了主观下线状态。

4)：当有足够数量的 Sentinel（大于等于配置文件指定的值）在指定的时间范围内确认Master的确进入了主观下线状态， 则Master会被标记为客观下线 。

5)：在一般情况下， 每个 Sentinel 会以每 10 秒一次的频率向它已知的所有Master，Slave发送 INFO 命令 。

6)：当Master被 Sentinel 标记为客观下线时，Sentinel 向下线的 Master 的所有 Slave 发送 INFO 命令的频率会从 10 秒一次改为每秒一次 。

7)：若没有足够数量的 Sentinel 同意 Master 已经下线， Master 的客观下线状态就会被移除。
8): 若 Master 重新向 Sentinel 的 PING 命令返回有效回复， Master 的主观下线状态就会被移除。

Redis集群高可用

简介

Redis单机模式可靠性保证不是很好，容易出现单点故障，同时其性能也受限于CPU的处理能力，实际开发中Redis必然是高可用的，所以单机模式并不是我们的终点，我们需要对目前redis的架构模式进行升级。
Sentinel模式做到了高可用，但是实质还是只有一个master在提供服务(读写分离的情况本质也是master在提供服务)，当master节点所在的机器内存不足以支撑系统的数据时，就需要考虑集群了。

Redis集群架构实现了对redis的水平扩容，即启动N个redis节点，将整个数据分布存储在这N个redis节点中，每个节点存储总数据的1/N。redis集群通过分区提供一定程度的可用性，即使集群中有一部分节点失效或无法进行通讯，集群也可以继续处理命令请求。

基本架构

对于redis集群(Cluster),一般最少设置为6个节点,3个master,3个slave,其简易架构如下:
在这里插入图片描述

docker创建集群

第一步：准备网络环境
创建虚拟网卡，主要是用于redis-cluster能于外界进行网络通信，一般常用桥接模式。

docker network create redis-net

查看docker的网卡信息，可使用如下指令

docker network ls

查看docker网络详细信息，可使用命令

docker network inspect redis-net

第二步：准备redis配置模板

mkdir -p /usr/local/docker/redis-cluster

cd /usr/local/docker/redis-cluster

vim redis-cluster.tmpl

在redis-cluster.tmpl中输入以下内容

port ${PORT}
cluster-enabled yes
cluster-config-file nodes.conf
cluster-node-timeout 5000
cluster-announce-ip 192.168.126.129  #改成自己的ip地址
cluster-announce-port ${PORT}
cluster-announce-bus-port 1${PORT}
appendonly yes

其中：

各节点解释如下所示：

  1. port：节点端口，即对外提供通信的端口
  2. cluster-enabled：是否启用集群
  3. cluster-config-file：集群配置文件
  4. cluster-node-timeout：连接超时时间
  5. cluster-announce-ip：宿主机ip
  6. cluster-announce-port：集群节点映射端口
  7. cluster-announce-bus-port：集群总线端口
  8. appendonly：持久化模式

在这里插入图片描述
第三步：创建节点配置文件

在redis-cluser目录下执行以下命令

for port in $(seq 8010 8015); \
do \
  mkdir -p ./${port}/conf  \
  && PORT=${port} envsubst < ./redis-cluster.tmpl > ./${port}/conf/redis.conf \
  && mkdir -p ./${port}/data; \
done

在这里插入图片描述
其中：

for 变量 in $(seq var1 var2);do …; done为linux中的一种shell 循环脚本, 例如：

[root@centos7964 ~]# for i in $(seq 1 5);
> do echo $i;
> done;
1
2
3
4
5
[root@centos7964 ~]#

指令envsubst <源文件>目标文件,用于将源文件内容更新到目标文件中.
通过cat指令查看配置文件内容

cat /usr/local/docker/redis-cluster/801{0..5}/conf/redis.conf

第四步：创建集群中的redis节点容器

for port in $(seq 8010 8015); \
do \
   docker run -it -d -p ${port}:${port} -p 1${port}:1${port} \
  --privileged=true -v /usr/local/docker/redis-cluster/${port}/conf/redis.conf:/usr/local/etc/redis/redis.conf \
  --privileged=true -v /usr/local/docker/redis-cluster/${port}/data:/data \
  --restart always --name redis-${port} --net redis-net \
  --sysctl net.core.somaxconn=1024 redis redis-server /usr/local/etc/redis/redis.conf; \
done

其中, --privileged=true表示让启动的容器用户具备真正root权限, --sysctl net.core.somaxconn=1024 这是一个linux的内核参数,用于设置请求队列大小,默认为128,后续启动redis的启动指令需要先放到这个请求队列中,然后依次启动.
创建成功以后，通过docker ps指令查看节点内容。

第五步：创建redis-cluster集群配置

docker exec -it redis-8010 bash

redis-cli --cluster create 192.168.126.129:8010 192.168.126.129:8011 192.168.126.129:8012 192.168.126.129:8013 192.168.126.129:8014 192.168.126.129:8015 --cluster-replicas 1

在这里插入图片描述
如上指令要尽量放在一行执行，其中最后的1表示主从比例，当出现选择提示信息时，输入yes即可。

第六步：连接redis-cluster，并添加数据到redis

redis-cli -c -h 192.168.126.129 -p 8010

在这里插入图片描述
其它：
在搭建过程，可能在出现问题后，需要停止或直接删除docker容器，可以使用以下参考命令

批量停止docker 容器：

docker ps -a | grep -i "redis-801*" | awk '{print $1}' | xargs docker stop

批量删除docker 容器：

docker ps -a | grep -i "redis-801*" | awk '{print $1}' | xargs docker rm -f

批量删除文件

rm -rf 801{0..5}/conf/redis.conf

以上就是基于docker搭建redis-cluster的简单步骤，实际应用中可能并没有这么简单，该文也仅用于参考。

Jedis读写数据测试
package com.jt;
import org.junit.Test;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPoolConfig;
import java.util.HashSet;
import java.util.Set;

public class JedisClusterTests {
    @Test
    public void testJedisCluster(){
        //1.创建Jedis集群操作对象
        Set<HostAndPort> nodes=new HashSet<>();
        nodes.add(new HostAndPort("192.168.126.129", 8010));
        nodes.add(new HostAndPort("192.168.126.129", 8011));
        nodes.add(new HostAndPort("192.168.126.129", 8012));
        nodes.add(new HostAndPort("192.168.126.129", 8013));
        nodes.add(new HostAndPort("192.168.126.129", 8014));
        nodes.add(new HostAndPort("192.168.126.129", 8015));        
        JedisCluster jedisCluster = new JedisCluster(nodes);        
        //2.执行数据读写操作
        jedisCluster.set("username", "tony");
        String username = jedisCluster.get("username");
        System.out.println(username);
        //3.释放资源
        jedisCluster.close();
    }
    
    @Test
    public void testJedisClusterPool(){
        //1.创建Jedis集群操作对象
        Set<HostAndPort> nodes=new HashSet<>();
        nodes.add(new HostAndPort("192.168.126.129", 8010));
        nodes.add(new HostAndPort("192.168.126.129", 8011));
        nodes.add(new HostAndPort("192.168.126.129", 8012));
        nodes.add(new HostAndPort("192.168.126.129", 8013));
        nodes.add(new HostAndPort("192.168.126.129", 8014));
        nodes.add(new HostAndPort("192.168.126.129", 8015));       
        JedisPoolConfig config = new JedisPoolConfig();
        config.setMaxTotal(100);
        config.setMaxIdle(60);
        JedisCluster jedisCluster = new JedisCluster(nodes,config);        
        //2.执行数据读写操作
        jedisCluster.set("username", "tony");
        String username = jedisCluster.get("username");
        System.out.println(username);        
        //3.释放资源
        jedisCluster.close();
    }
}//集群优势：扩容，高可用(可靠性)
//Redis如何解决数据的可靠性问题的？
//	1)持久化(rdb,aof)
//	2)主从+哨兵，集群
RedisTemplate读写数据测试

第一步：配置application.yml,例如：

spring:
  redis:
    cluster: #redis 集群配置
      nodes: 192.168.126.128:8010,192.168.126.128:8011,192.168.126.128:8012,192.168.126.128:8013,192.168.126.128:8014,192.168.126.128:8015
      max-redirects: 3 #最大跳转次数
    timeout: 5000 #超时时间
    database: 0
    jedis: #连接池
      pool:
        max-idle: 8
        max-wait: 0

第二步：编写单元测试类，代码如下：

package com.cy.redis;

@SpringBootTest
public class RedisClusterTests {
    @Autowired
    private RedisTemplate redisTemplate;
    
    @Test
    void testMasterReadWrite(){
        //1.获取数据操作对象
        ValueOperations valueOperations = redisTemplate.opsForValue();
        //2.读写数据
        valueOperations.set("city","beijing");
        Object city=valueOperations.get("city");
        System.out.println(city);
    }
}
小节面试分析
  • Redis的高并发跟整个系统的高并发是什么关系？

第一:Redis你要搞高并发的话，不可避免，要把底层的缓存搞得很好。例如，mysql的高并发，是通过一系列复杂的分库分表，订单系统，事务要求的，QPS到几万，比较高了。
第二：要做一些电商的商品详情页，真正的超高并发，QPS上十万，甚至是百万，一秒钟百万的请求量，只有redis是不够的，但是redis是整个大型的缓存架构中，支撑高并发的架构里面，非常重要的一个环节。
第三：你的底层的缓存中间件，缓存系统，必须能够支撑的起我们说的那种高并发，其次，再经过良好的整体的缓存架构的设计（多级缓存架构、热点缓存），支撑真正的上十万，甚至上百万的高并发。

Redis在SpringBoot工程中的综合应用

业务描述

从一个博客数据库中查询所有的文章标签，然后存储到缓存(Cache)，后续查询时可从缓存获取。提高其查询性能。

准备工作

初始化数据

初始化数据库中数据，SQL脚本如下：

DROP DATABASE IF EXISTS `blog`;
CREATE DATABASE `blog` DEFAULT character set utf8mb4;
SET names utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;
USE `blog`;
CREATE TABLE `tb_tag` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'id',
  `name` varchar(255) NOT NULL COMMENT 'data_id',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='tb_tag';
insert into `tb_tag` values (null,"mysql"),(null,"redis");
添加项目依赖

在jt-template工程的原有依赖基础上添加mysql数据库访问依赖，例如：

<!--mysql-->
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
</dependency>

<!--mybatis-->
<dependency>
    <groupId>com.baomidou</groupId>
    <artifactId>mybatis-plus-boot-starter</artifactId>
    <version>3.4.2</version>
</dependency>

<!--lombok-->
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
</dependency>
添加数据库访问配置

在项目的配置文件（例如application.yml）中添加数据库访问配置，例如：

# Redis的单机配置
spring:
  redis: # redis-cli -c -h 192.168.126.129 -p 8010
    host: 192.168.126.129
    port: 6379
  datasource:
    url: jdbc:mysql:///blog?serverTimezone=Asia/Shanghai&characterEncoding=utf8
    username: root
    password: root
# 日志配置，可以可以知道查询到的数据是从redis中查到的还是数据库中查到的
logging:
  level: 
    com.jt: debug

业务逻辑代码设计及实现

pojo对象设计

创建一个Tag类，基于此类型的对象存储Tag(标签信息)，代码如下：

package com.jt.blog.pojo;

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;
import java.io.Serializable;

@Data
@TableName("tb_tag")
public class Tag implements Serializable {
    private static final long serialVersionUID = -5820108337706914611L;
    
    @TableId(type = IdType.AUTO)
    private Integer id;
    
    @TableField("name")
    private String name;
}
Dao 逻辑对象设计

创建Tag信息的数据访问接口，代码如下：

package com.jt.blog.Dao;

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.jt.blog.pojo.Tag;
import org.apache.ibatis.annotations.Mapper;

@Mapper
public interface TagMapper extends BaseMapper<Tag> {
}

创建单元测试类，TagMapper中的相关方法进行单元测试，例如：

package com.jt.blog.dao;

import com.jt.blog.domain.Tag;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import java.util.List;

@SpringBootTest
public class TagMapperTests {
    @Autowired
    private TagMapper tagMapper; 
       
    @Test
    void testSelectList(){
        List<Tag> tags = tagMapper.selectList(null);
        for(Tag t:tags){
            System.out.println(t);
            //System.out.println(t.getId()+"/"+t.getName());
        }
    }
}
Service 逻辑对象设计

设计TagService接口及实现类，定义Tag（标签）业务逻辑。
第一步：定义TagService接口，代码如下：

package com.jt.blog.service;

import com.jt.blog.pojo.Tag;
import java.util.List;

public interface TagService {
    List<Tag> selectTags();
}

第二步：定义TagServiceImpl类，代码如下：

package com.jt.blog.service;

import com.jt.blog.Dao.TagMapper;
import com.jt.blog.pojo.Tag;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import java.util.List;

@Service
public class TagServiceImpl implements TagService {
    @Autowired
    private TagMapper tagMapper;
    
    @Autowired
    private RedisTemplate redisTemplate;
    
    @Override
    public List<Tag> selectTags() {
        //1.从redis中查询Tag信息，redis有则直接返回
        ValueOperations valueOperations = redisTemplate.opsForValue();
        List<Tag> tags = (List<Tag>) valueOperations.get("tags");
        if(tags != null && tags.isEmpty()) return tags;
        //2.从redis没有查询到tag信息，查询mysql
        tags = tagMapper.selectList(new QueryWrapper<>(null));        
        //3.从mysql查询到的tag信息存储到redis中
        valueOperations.set("tags", tags);
        return tags;
    }
}

说明，假如将List存储到redis，此时Tag必须实现Serializable接口。

第三步：定义TagServiceTests单元测试类并进行单元测试，代码如下：

package com.jt.blog;

import com.jt.blog.pojo.Tag;
import com.jt.blog.service.TagService;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import java.util.List;

@SpringBootTest
public class TagServiceTests {
    @Autowired
    private TagService tagService;
    
    @Test
    public void testSelectService(){
        List<Tag> redis = tagService.selectTags();
        System.out.println(redis);
    }
}
Controller逻辑对象设计

创建Tag控制逻辑对象，用于处理请求和响应逻辑，代码如下：

package com.jt.blog.controller;

import com.jt.blog.domain.Tag;
import com.jt.blog.service.TagService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.ArrayList;
import java.util.List;

@RestController
@RequestMapping("/tag")
public class TagController {
    @Autowired
    private TagService tagService;  
      
    @GetMapping
    public  List<Tag> doSelectTags(){
      return  tagService.selectTags());//1.redis,2.mysql
    }
}

在这里插入图片描述
启动服务，打开浏览器进行访问测试。同时思考，我们是否可以在这个层加一个本地cache。

业务逻辑代码优化

定制RedisTemplate对象

RedisTemplate默认采用的是JDK的序列化方式，假如对系统对序列化做一些调整，可以自己定义RedisTemplate对象，例如：

package com.jt.blog.config;

@Configuration
public class RedisConfig {
    //代码制定参考RedisAutoConfiguration类
    @Bean
    public RedisTemplate<String, List<Tag>> redisTemplate(RedisConnectionFactory redisConnectionFactory) throws UnknownHostException {
        RedisTemplate<String, List<Tag>> template = new RedisTemplate();
        template.setConnectionFactory(redisConnectionFactory);
        //定义redisTemplate对象的序列化方式
        //1.定义key的序列化方式
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        template.setKeySerializer(stringRedisSerializer);
        template.setHashKeySerializer(stringRedisSerializer);
        //2.定义value的序列化方式
        Jackson2JsonRedisSerializer jsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        objectMapper.activateDefaultTyping(objectMapper.getPolymorphicTypeValidator(),
                ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY);
        jsonRedisSerializer.setObjectMapper(objectMapper);
        template.setValueSerializer(jsonRedisSerializer);
        template.setHashValueSerializer(jsonRedisSerializer);
        //3.修改redisTemplate默认特性(除了序列化，其他原有特性不丢)
        template.afterPropertiesSet();
        return template;
    }
}
//序列化：
//狭义层面：将对象转换为字节(I/O)
//广义层面：将对象转换为指定格式子字符串(例如json)

//封装：
//狭义层面：属性私有化，方法能公开则公开
//广义层面：一个系统有哪些服务(子系统)，一个服务有哪些模块，一个模块有哪些对象，一个对象有哪些属性
Service中缓存应用优化

目标:简化缓存代码的编写
解决方案:基于AOP(面向切面编程)方式实现缓存应用
实践步骤:
第一步:在启动上类添加@EnableCaching注解(开启AOP方式的缓存配置),例如:
在这里插入图片描述
第二步:重构TagServiceImpl中的selectTags()方法,方法上使用@Cacheable注解,例如:

package com.jt.blog.service;

import com.jt.blog.Dao.TagMapper;
import com.jt.blog.pojo.Tag;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.stereotype.Service;
import java.util.List;

@Service
public class TagServiceImpl implements TagService {
    @Autowired
    private TagMapper tagMapper; 
       
    //调用此方法时系统底层会先从redis缓存中查找数据，假如缓存中没有数据，就会从mysql数据库查找数据
    //@Cacheable这个注解假如想要生效，需要在启动类或者配置类中添加@EnableCaching注解，表示启动AOP方式的缓存配置
    @Cacheable(value="tagCache")//基于此注解(@Cacheable)描述的方法为缓存切入点方法
    @Override
    public List<Tag> selectTags() {
        return tagMapper.selectList(null);
    }
}

其中,@Cacheable描述的方法为AOP中的一个切入点方法,访问这个方法时,系统底层会通过一个拦截器,检查缓存中是否有你要的数据,假如有则直接返回,没有则执行方法从数据库查询数据.

我们还可以定义Redis中key和value的序列化方式,例如:

package com.jt.blog.config;

@Configuration
public class RedisConfig {
    /**
     * 定义缓存key生成器，不定义也可以使用默认的
     * @return
     */
    @Bean
    public KeyGenerator keyGenerator() {
        return (Object target, Method method, Object... params) -> {
            StringBuilder sb = new StringBuilder();
            sb.append(target.getClass().getName()); // 类目
            sb.append("::");
            sb.append(method.getName()); // 方法名
            for (Object param : params) {
                sb.append(param.toString());
            }
            return sb.toString();
        };
    }
    
    /**
     * 自定义Cache管理器对象，不定义也可以，有默认的，但假如希望基于AOP方式实现Redis的操作时，
     * 按照指定的序列化方式进行序列化，可以对CacheManager进行自定义。
     * @param connectionFactory
     * @return
     */
    @Bean
    public RedisCacheManager cacheManager(
            RedisConnectionFactory connectionFactory){
        RedisCacheConfiguration config =
                RedisCacheConfiguration.defaultCacheConfig()
                        .entryTtl(Duration.ofSeconds(60))//设置key的有效时间，60s缓存失效
        //设置key的序列化方式
        .serializeKeysWith(RedisSerializationContext
                .SerializationPair.fromSerializer(
                        new StringRedisSerializer()))
        //设置value的序列化方式
        .serializeValuesWith(RedisSerializationContext
                .SerializationPair.fromSerializer(
                        new Jackson2JsonRedisSerializer<Object>(
                                Object.class)))
                 //不缓存null值
                .disableCachingNullValues();
        return RedisCacheManager
                .builder(connectionFactory)
                .cacheDefaults(config)//修改默认配置
                .transactionAware()//启动默认事务
                .build();
    }
}
//序列化：
//狭义层面：将对象转换为字节(I/O)
//广义层面：将对象转换为指定格式子字符串(例如json)

//封装：
//狭义层面：属性私有化，方法能公开则公开
//广义层面：一个系统有哪些服务(子系统)，一个服务有哪些模块，一个模块有哪些对象，一个对象有哪些属性

其中,写好这个配置类后,可以进行单元测试,检测redis中数据的存储.
在这里插入图片描述

Controller中添加本地缓存

在Controller中添加一个本地缓存,减少对远程redis缓存的访问,例如:

package com.jt.blog.controller;

import com.jt.blog.pojo.Tag;
import com.jt.blog.service.TagService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

@RestController
@RequestMapping("/tag")
public class TagController {
    @Autowired
    private TagService tagService;
    
    //CopyOnWriteArrayList对象是一个线程安全的List集合
    //这个集合在修修改内容时，会先将集合内容拷贝到线程自己的内存区域，然后修改，
    //修改后，再将私有区域的数据写到共享区域(假如原有的集合内容)
    private List<Tag> tags = new CopyOnWriteArrayList<>();//本地缓存
    
    @GetMapping
    public List<Tag> doSelectTags(){
        if(tags.isEmpty()){
            synchronized (tags) {//同步代码块
                if (tags.isEmpty()) {
                    tags.addAll(tagService.selectTags());
                }
            }//保证业务的原子性
        }
        //tags = tagService.selectTags();//先查询redis，再查询mysql
        return tags;
    }
}
//redis 分布式缓存
//Tomcat 本地缓存(JVM内部提供)
//Browser(查询) -> Nginx -> Gateway,Gateway... -> (Tomcat,Tomcat...)
//查询 -> 本地cache(JVM本地缓存) -> 分布式cache(redis缓存) -> mysql
缓存数据一致性分析

当我们从数据库查询数据以后,假如将数据存入到了缓存,后续更新了数据库的数据,但假如没有更新缓存就会出现缓存数据与数据库数据不一致的这样的现象,对于这样问题,有时允许在一定时间范围之内存在。假如我们希望在更新了数据库数据以后要更新缓存,如何实现呢?

# redis # bootstrap # 前端
Logo
华为开发者空间

华为开发者空间，是为全球开发者打造的专属开发空间，汇聚了华为优质开发资源及工具，致力于让每一位开发者拥有一台云主机，基于华为根生态开发、创新。

更多推荐

cover

GaussDB性能调优

avatar 华为开发者空间
cover

领取云主机，带你基于PyTorch构建高效手写体识别系统

avatar 华为开发者空间
cover

华为“云上先锋”实践赛·AI赛道现场会活动成功举办

avatar 华为开发者空间