Redis进阶

缓存雪崩缓存穿透缓存击穿

Redis在项目中常用作缓存来使用，主要用两大作用：

1.提升系统的性能

Redis基于内存，IO效率远高于MySql数据库

2.减少数据库压力

Redis处理很多请求，使用Redis作为缓存可以减少数据库的请求量，避免数据库因为请求过多、压力过大而导致宕机。

在这里插入图片描述
击穿问题解决案例

解决方法：使用双检锁DCL机制优化方法，通过给代码块加上synchronized锁，同步代码块的方式。再加上if else判断的方式来优化方法，解决击穿的问题，优化性能。

@Override
    public Student getStudentById(Long id) {
        //获得字符串操作对象
        ValueOperations<String, Object> ops = redisTemplate.opsForValue();
        //先查询Redis
        Student stu = (Student) ops.get(PREFIX + id);
        if(stu == null) {
            synchronized (this) {
                System.out.println("进入了同步锁");
                //先查询Redis
                stu = (Student) ops.get(PREFIX + id);
                //如果Redis缓存存在数据，就直接返回
                if (stu != null) {
                    System.out.println("Redis查到，返回" + stu);
                    return stu;
                }
                //如果Redis没有查到数据，就查询MySQL
                stu = studentMapper.selectById(id);
                //MySQL查到数据，就保存到Redis
                if (stu != null) {
                    System.out.println("MySQL查到，返回" + stu);
                    ops.set(PREFIX + id, stu);
                    return stu;
                }
                //MySQL没有数据，就返回null
                System.out.println("MySQL没有数据，就返回null");
            }
        }else {
            System.out.println("没有执行同步锁");
        }
        return stu;
    }

穿透问题

缓存空对象，空对象设置过期时间

@Override
    public Student getStudentById(Long id) {
        //获得字符串操作对象
        ValueOperations<String, Object> ops = redisTemplate.opsForValue();
        //先查询Redis，如果存在数据就不执行同步代码块，直接返回
        Student stu = (Student) ops.get(PREFIX + id);
        if(stu == null) {
            synchronized (this) {
                System.out.println("进入了同步锁");
                //先查询Redis
                stu = (Student) ops.get(PREFIX + id);
                //如果Redis缓存存在数据，就直接返回
                if (stu != null) {
                    System.out.println("Redis查到，返回" + stu);
                    return stu;
                }
                //如果Redis没有查到数据，就查询MySQL
                stu = studentMapper.selectById(id);
                //MySQL查到数据，就保存到Redis
                if (stu != null) {
                    System.out.println("MySQL查到，返回" + stu);
                    ops.set(PREFIX + id, stu);
                    return stu;
                }else {
                    //MySQL没有数据,在Redis保存空对象,设置过期时间
                    System.out.println("MySQL没有数据");
                    Student student = new Student();
                    ops.set(PREFIX + id, student,5, TimeUnit.SECONDS);
                }
            }
        }else {
            System.out.println("没有执行同步锁");
        }
        return stu;
    }

2.redis消息通知广播
消息接收端

@Test
@DisplayName("测试Redis,消息通知")
public void test5(){
 
    // 创建 Jedis 对象连接 Redis 服务器
    try (Jedis jedis = new Jedis("localhost",6379)) {
        // 创建订阅者对象
        JedisPubSub subscriber = new JedisPubSub() {
            @Override
            public void onMessage(String channel, String message) {
                // 接收到消息时的处理逻辑
                System.out.println("Received message: " + message + " from channel: " + channel);
            }
        };
 
        // 订阅名为 "channel_name" 的频道
        jedis.subscribe(subscriber, "channel_name");
 
        // 阻塞直到接收到消息
        Thread.sleep(0);
    } catch (Exception e) {
        e.printStackTrace();
    }
}

消息发送端

@Test
@DisplayName("测试Redis，发送消息通知")
public void test6(){
    // 创建 Jedis 对象连接 Redis 服务器
    try (Jedis jedis = new Jedis("localhost",6379)) {
        // 发布消息到名为 "channel_name" 的频道
        jedis.publish("channel_name", "Hello !!!!!");
    } catch (Exception e) {
        e.printStackTrace();
    }
}

优点：

高性能： Redis 的消息通知是基于发布/订阅模式实现的，使用轻量级的事件通知机制，因此具有很高的性能和低延迟。

松耦合：发布者和订阅者之间是解耦的，发布者只需发布消息，而订阅者可以根据需要选择性地订阅感兴趣的频道或主题，这种松耦合性使得系统更容易维护和扩展。

实时性： Redis 提供了实时性的消息传递机制，能够快速地将消息传递给所有订阅者，使得系统可以快速响应和更新。

缺点：

消息丢失： Redis 是内存数据库，当服务宕机或断电时，如果消息尚未持久化到磁盘，可能会导致消息丢失。为了解决这个问题，可以采用 Redis 的持久化机制（如 AOF 或 RDB）来避免数据丢失。

无法保证可靠性： Redis 的发布/订阅机制本身不保证消息的可靠传输。订阅者可能会因为某些原因错过一些消息，因此不能保证每个订阅者都能收到所有消息。

无法跨越网络边界： Redis 是一个内存数据库，因此在不同的网络边界之间传递消息可能会受到限制。如果要在多个网络边界之间进行消息传递，可能需要使用其他的消息队列或者消息中间件来解决跨网络通信问题。

总体来说，Redis 的消息通知机制适用于许多实时性要求高、轻量级、非关键性的消息通知场景。但在一些对消息的可靠性和持久性有较高要求的场景下，可能需要考虑其他更为健壮的消息队列系统或消息中间件。

3.redis高可用哨兵模式集群模式

Redis 高可用有三种主要模式：主从复制模式（Master-Slave），哨兵模式（Sentinel）和集群模式（Cluster）。它们都旨在提供高可用性和容错性，但在实现方式和应用场景上有所不同。

哨兵模式（Sentinel）
特点：哨兵模式是一种分布式系统，用于监控 Redis 实例的健康状态，并在主服务器宕机时自动进行故障转移。

工作原理：哨兵模式由若干哨兵进程组成，这些哨兵会监视 Redis 主服务器和从服务器的状态。当主服务器宕机或无法访问时，哨兵会选举出新的主服务器，并将从服务器提升为新的主服务器，以确保系统的高可用性。

适用场景：适用于对数据的持久性要求较高，但对于读写速度不是特别高的应用场景。

集群模式（Cluster）
特点：集群模式是 Redis 3.0 版本后引入的功能，用于实现分布式存储和高性能的数据访问。

工作原理：集群模式通过将数据分片存储在多个节点上，以提供水平扩展性和负载均衡。它支持自动数据分片和节点的动态增减，每个节点负责管理一部分数据，并通过内部通信协议进行数据交换和传输。

适用场景：适用于对于大规模数据、高性能读写都有较高要求的应用场景。

主从复制模式（Master-Slave）
2.1 主从复制原理
主从复制是Redis的一种基本集群模式，它通过将一个Redis节点（主节点）的数据复制到一个或多个其他Redis节点（从节点）来实现数据的冗余和备份。

主节点负责处理客户端的写操作，同时从节点会实时同步主节点的数据。客户端可以从从节点读取数据，实现读写分离，提高系统性能。
在这里插入图片描述
2.2 主从复制配置和实现
配置主节点：在主节点的redis.conf配置文件中，无需进行特殊配置，主节点默认监听所有客户端请求。

主节点默认端口号6379

port 6379
配置从节点：在从节点的redis.conf配置文件中，添加如下配置，指定主节点的地址和端口：

从节点设置端口号6380

port 6380

replicaof 主节点IP 主节点端口

replicaof 127.0.0.1 6379

或者，通过Redis命令行在从节点上执行如下命令：
redis> replicaof 127.0.0.1 6379
验证主从复制：在主节点上执行写操作，然后在从节点上进行读操作，检查数据是否一致。
2.3 主从复制的优缺点
优点：

配置简单，易于实现。
实现数据冗余，提高数据可靠性。
读写分离，提高系统性能。
缺点：

主节点故障时，需要手动切换到从节点，故障恢复时间较长。
主节点承担所有写操作，可能成为性能瓶颈。
无法实现数据分片，受单节点内存限制。
2.4 主从复制场景应用
主从复制模式适用于以下场景：

数据备份和容灾恢复：通过从节点备份主节点的数据，实现数据冗余。
读写分离：将读操作分发到从节点，减轻主节点压力，提高系统性能。
在线升级和扩展：在不影响主节点的情况下，通过增加从节点来扩展系统的读取能力。
总结：主从复制模式适合数据备份、读写分离和在线升级等场景，但在主节点故障时需要手动切换，不能自动实现故障转移。如果对高可用性要求较高，可以考虑使用哨兵模式或Cluster模式。

哨兵模式（Sentinel）
3.1 哨兵模式原理
哨兵模式是在主从复制基础上加入了哨兵节点，实现了自动故障转移。哨兵节点是一种特殊的Redis节点，它会监控主节点和从节点的运行状态。当主节点发生故障时，哨兵节点会自动从从节点中选举出一个新的主节点，并通知其他从节点和客户端，实现故障转移。

3.2 哨兵模式配置和实现
配置主从复制：首先按照主从复制模式的配置方法，搭建一个主从复制集群（上面已经讲过）。
配置哨兵节点：在哨兵节点上创建一个新的哨兵配置文件（如：sentinel.conf），并添加如下配置：

sentinel节点端口号

port 26379

sentinel monitor 被监控主节点名称主节点IP 主节点端口 quorum

sentinel monitor mymaster 127.0.0.1 6379 2

sentinel down-after-milliseconds 被监控主节点名称毫秒数

sentinel down-after-milliseconds mymaster 60000

sentinel failover-timeout 被监控主节点名称毫秒数

sentinel failover-timeout mymaster 180000
其中，quorum是指触发故障转移所需的最小哨兵节点数。down-after-milliseconds表示主节点被判断为失效的时间。failover-timeout是故障转移超时时间。
为什么只配置了sentinel监控主节点，没有配置监控从节点？因为通过主节点，就可以找到从节点。

启动哨兵节点：使用如下命令启动哨兵节点：
redis> redis-sentinel /path/to/sentinel.conf
验证哨兵模式：手动停止主节点，观察哨兵节点是否自动选举出新的主节点，并通知其他从节点和客户端。
3.3 哨兵模式的优缺点
优点：

自动故障转移，提高系统的高可用性。
具有主从复制模式的所有优点，如数据冗余和读写分离。
缺点：

配置和管理相对复杂。
依然无法实现数据分片，受单节点内存限制。
3.4 哨兵模式场景应用
哨兵模式适用于以下场景：

高可用性要求较高的场景：通过自动故障转移，确保服务的持续可用。
数据备份和容灾恢复：在主从复制的基础上，提供自动故障转移功能。
总结：哨兵模式在主从复制模式的基础上实现了自动故障转移，提高了系统的高可用性。然而，它仍然无法实现数据分片。如果需要实现数据分片和负载均衡，可以考虑使用Cluster模式。

Cluster模式
4.1 Cluster模式原理
Cluster模式是Redis的一种高级集群模式，它通过数据分片和分布式存储实现了负载均衡和高可用性。在Cluster模式下，Redis将所有的键值对数据分散在多个节点上。每个节点负责一部分数据，称为槽位。通过对数据的分片，Cluster模式可以突破单节点的内存限制，实现更大规模的数据存储。

在这里插入图片描述
4.2 数据分片与槽位
Redis Cluster将数据分为16384个槽位，每个节点负责管理一部分槽位。当客户端向Redis Cluster发送请求时，Cluster会根据键的哈希值将请求路由到相应的节点。具体来说，Redis Cluster使用CRC16算法计算键的哈希值，然后对16384取模，得到槽位编号。

4.3 Cluster模式配置和实现
配置Redis节点：为每个节点创建一个redis.conf配置文件，并添加如下配置：

cluster节点端口号

port 7001

开启集群模式

cluster-enabled yes

节点超时时间

cluster-node-timeout 15000
像这样的配置，一共需要创建6个，我们做一个三主三从的集群。

启动Redis节点：使用如下命令启动6个节点：
redis> redis-server redis_7001.conf
创建Redis Cluster：使用Redis命令行工具执行如下命令创建Cluster：
redis> redis-cli --cluster create 127.0.0.1:7001 127.0.0.1:7002 127.0.0.1:7003 127.0.0.1:7004 127.0.0.1:7005 127.0.0.1:7006 --cluster-replicas 1
cluster-replicas 表示从节点的数量，1代表每个主节点都有一个从节点。
验证Cluster模式：向Cluster发送请求，观察请求是否正确路由到相应的节点。
4.4 Cluster模式的优缺点
优点：

数据分片，实现大规模数据存储。
负载均衡，提高系统性能。
自动故障转移，提高高可用性。
缺点：

配置和管理较复杂。
一些复杂的多键操作可能受到限制。
4.5 Cluster模式场景应用
Cluster模式适用于以下场景：

大规模数据存储：通过数据分片，突破单节点内存限制。
高性能要求场景：通过负载均衡，提高系统性能。
高可用性要求场景：通过自动故障转移，确保服务的持续可用。
总结：Cluster模式在提供高可用性的同时，实现了数据分片和负载均衡，适用于大规模数据存储和高性能要求的场景。然而，它的配置和管理相对复杂，且某些复杂的多键操作可能受到限制。

总结
本文详细介绍了Redis的三大集群模式：主从复制、哨兵模式和Cluster模式。每种模式都有其特点和应用场景，具体如下：

主从复制模式：适用于数据备份和读写分离场景，配置简单，但在主节点故障时需要手动切换。
哨兵模式：在主从复制的基础上实现自动故障转移，提高高可用性，适用于高可用性要求较高的场景。
Cluster模式：通过数据分片和负载均衡实现大规模数据存储和高性能，适用于大规模数据存储和高性能要求场景。

4.分布式锁
1 redis 分布式锁实现原理
所谓分布式锁，应当基本如下几项核心性质：

独占性：对于同一把锁，在同一时刻只能被一个取锁方占有，这是锁最基础的一项特征
健壮性：即不能产生死锁（dead lock）. 假如某个占有锁的使用方因为宕机而无法主动执行解锁动作，锁也应该能够被正常传承下去，被其他使用方所延续使用
对称性：加锁和解锁的使用方必须为同一身份. 不允许非法释放他人持有的分布式锁
高可用：当提供分布式锁服务的基础组件中存在少量节点发生故障时，应该不能影响到分布式锁服务的稳定性

在使用 redis 分布式锁时，为避免持有锁的使用方因为异常状况导致无法正常解锁，进而引发死锁问题，我们可以使用到 redis 的数据过期时间 expire 机制.

我们通常会在插入分布式锁对应的 kv 数据时设置一个过期时间 expire time，这样即便使用方在持有锁期间发生宕机无法正常解锁，锁对应的数据项也会在达到过期时间阈值后被自动删除，实现分布式锁释放的效果. 此处我们可以通过 redis 的 SETEX 指令，一气呵成地完成上锁+设置过期时间的两个步骤.

在这里插入图片描述
java实现分布式锁例子

import redis.clients.jedis.Jedis;
 
public class RedisDistributedLock {
    private Jedis jedis;
    private String lockKey;
    private String uniqueIdentifier;
    private int lockTimeout;
 
    public RedisDistributedLock(Jedis jedis, String lockKey, String uniqueIdentifier, int lockTimeout) {
        this.jedis = jedis;
        this.lockKey = lockKey;
        this.uniqueIdentifier = uniqueIdentifier;
        this.lockTimeout = lockTimeout;
    }
 
    public boolean acquireLock() {
        String result = jedis.set(lockKey, uniqueIdentifier, "NX", "EX", lockTimeout);
        return "OK".equals(result);
    }
 
    public void releaseLock() {
        // 校验并释放锁
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        jedis.eval(script, 1, lockKey, uniqueIdentifier);
    }
 
    public static void main(String[] args) {
        String redisHost = "localhost";
        int redisPort = 6379;
        String lockKey = "my_lock";
        String uniqueIdentifier = "unique_id";
        int lockTimeout = 10; // 锁的过期时间
 
        try (Jedis jedis = new Jedis(redisHost, redisPort)) {
            RedisDistributedLock distributedLock = new RedisDistributedLock(jedis, lockKey, uniqueIdentifier, lockTimeout);
 
            if (distributedLock.acquireLock()) {
                try {
                    // 成功获取到锁，执行业务逻辑
                    System.out.println("Do something here...");
                } finally {
                    // 释放锁
                    distributedLock.releaseLock();
                }
            } else {
                // 获取锁失败
                System.out.println("Failed to acquire the lock");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

这种 expire 机制的使用会引入一个新的问题——过期时间不精准. 因为此处设置的过期时间只能是一个经验值（通常情况下偏于保守），既然是经验值，那就做不到百分之百的严谨性. 试想假如占有锁的使用方在业务处理流程中因为一些异常的耗时（如 IO、GC等），导致业务逻辑处理时间超过了预设的过期时间，就会导致锁被提前释放. 此时在原使用方的视角中，锁仍然持有在自己手中，但在实际情况下，锁数据已经被删除，其他取锁方可能取锁成功，于是就可能引起一把锁同时被多个使用方占用的问题，锁的基本性质——独占性遭到破坏.

为解决这个问题，redisson提出新的方案-看门狗策略

看门狗机制

etcd 续约机制与 watchDog 看门狗机制非常类似.

租约，顾名思义，是一份具有时效性的协议，一旦达到租约上规定的截止时间，租约就会失去效力

在使用 etcd 的租约能力时，用户会先预设一个租约过期时间，但并非一个绝对意义的截止时间，因为租约是支持动态续约操作的. 接下来用户可以异步启动一个续约协程，按照指定的时间节奏进行续约操作，延长租约的过期时间. 这样实现的好处在于：

使用方规避了因为业务逻辑处理过长，导致租约数据（包含了分布式锁）提前过期释放的问题（因为有续约协程持续进行续约）
规避了因锁持有方宕机导致租约数据无法释放，内部包含的分布式锁产生死锁问题（倘若持有方宕机了，那续约协程也就停止工作了，续约工作未正常执行，租约会在下一个过期时间节点被回收，包含的锁数据也就释放了）

在这里插入图片描述

import redis.clients.jedis.Jedis;
 
public class RedisDistributedLockWithWatchdog {
    // 其他代码...
 
    private volatile boolean running = true;
 
    public void stopWatchdog() {
        running = false;
    }
 
    public boolean acquireLockWithWatchdog() {
        // 获取锁的代码...
 
        if ("OK".equals(result)) {
            // 获取锁成功，开启看门狗定时刷新任务
            Thread watchdogThread = new Thread(() -> {
                try {
                    while (running) {
                        Thread.sleep(WATCHDOG_INTERVAL * 1000L); // 等待一段时间后刷新锁
                        jedis.expire(lockKey, lockTimeout); // 刷新锁的过期时间
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
            watchdogThread.setDaemon(true);
            watchdogThread.start();
            return true;
        }
        return false;
    }
 
    public static void main(String[] args) {
        // 其他代码...
 
        try (Jedis jedis = new Jedis(redisHost, redisPort)) {
            RedisDistributedLockWithWatchdog distributedLock = new RedisDistributedLockWithWatchdog(jedis, lockKey, uniqueIdentifier, LOCK_TIMEOUT);
 
            if (distributedLock.acquireLockWithWatchdog()) {
                try {
                    // 执行业务逻辑...
 
                    // 成功获取到锁，业务处理结束后释放锁并停止看门狗
                    distributedLock.releaseLock();
                    distributedLock.stopWatchdog();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    // 释放锁
                    distributedLock.releaseLock();
                }
            } else {
                // 获取锁失败
                System.out.println("Failed to acquire the lock");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}