Redis知识总结[二]

跳转至：Redis知识总结[一]

Redis企业级应用

Redis脑裂

概念解决方案 Redis缓存预热

概念 Redis缓存穿透

概念解决方法 Redis缓存击穿

概念解决方案 Redis缓存雪崩

概念解决方案 Redis分布式锁(SpringBoot版)

概念代码实现(一)代码实现(二)代码实现(三)代码实现(四) Redis实现消息队列

List消息队列发布/订阅消息队列

假设现在有三台机器，分别安装了redis服务，结构如图
如果此时master服务器由于网络波动导致和两台slave机器无法正常通信，但是和客户端的连接是正常的。那么sentinel就会从两台slave机器中选举其中一个作为新的master来处理客户端请求。
这个时候，已经存在两台master服务器，client发送的数据会持续保存在旧的master服务器中，而新的master和slave中没有新的数据。如果一分钟以后，网络恢复正常，服务之间能够正常通信。此时，sentinel会把旧的master会变成新的master的slave节点。
问题出现了，slave会从master中同步数据，保持主从数据一致。这个时候，变成了slave节点的旧master会丢失掉通信异常期间从客户端接收到的数据。 解决方案

redis.conf中添加下面的配置：

//最少的slave节点为1个
min-replicas-to-write 1
//数据复制和同步的延迟不能超过10秒
min-replicas-max-lag 10

配置了这两个参数之后，如果发生脑裂，原master会在客户端写入操作的时候拒绝请求。这样可以避免大量数据丢失。

新启动的系统没有任何缓存数据，在缓存重建数据的过程中，系统性能和数据库负载都不太好，所以最好是在系统上线之前就把要缓存的热点数据加载到缓存中，这种缓存预加载手段就是缓存预热。 Redis缓存穿透 概念

如果数据库中没有对应的数据，从而导致缓存中也没有对应数据，所以每次请求都会穿过缓存直接到数据库进行查询，并发量高的情况下进而导致数据库直接宕机，这就是缓存穿透。

缓存穿透是数据库没有该数据，无法向缓存中存入数据，所以每次访问都必须去请求数据库。

对空值缓存：如果一个查询返回的数据为空（不管数据是否存在），我们仍然把这个空结果缓存，设置空结果的过期时间会很短，最长不超过5分钟。设置白名单：使用bitmaps类型定义一个可以访问的名单，用户id作为偏移量，每次访问查询是否在白名单中，如果不存在，则拒绝访问。 Redis缓存击穿 概念

某一个热点数据，在缓存过期的一瞬间，同时有大量的请求打进来，由于此时缓存过期了，所以请求最终都会走到数据库，造成瞬时数据库请求量大、压力骤增，甚至可能打垮数据库。

缓存击穿是数据库存在该数据，只是在某一瞬间缓存中的数据过期，导致请求都去找数据库了。

加互斥锁：在并发的多个请求中，只有第一个请求线程能拿到锁并执行数据库查询操作，其他的线程拿不到锁就阻塞等着，等到第一个线程将数据写入缓存后，其他线程直接查询缓存。热点数据不过期：直接将缓存设置为不过期，然后由定时任务去异步加载数据，更新缓存。 Redis缓存雪崩 概念

缓存雪崩是指在我们设置缓存时采用了相同的过期时间，导致缓存在某一时刻同时失效，请求全部转发到数据库，数据库瞬时压力过重雪崩。

和缓存击穿不同的是，缓存击穿指并发查同一条数据，缓存雪崩是不同数据都过期了，很多数据都查不到从而查数据库。

缓存数据的过期时间设置随机，防止同一时间大量数据过期现象发生。如果缓存数据库是分布式部署，将热点数据均匀分布在不同搞得缓存数据库中。设置热点数据永远不过期。 Redis分布式锁(SpringBoot版) 概念

两台或者两台以上的服务要去操作同一个数据的值的时候，可能会造成错误。不加锁会出现库存变为负数的情况
加锁可以避免这种情况：

下面是用springBoot整合redis实现分布式锁

public void test1(){
    ValueOperations valueOperations = redisTemplate.opsForValue();
    //占位，如果key不存在则设置成功
    Boolean isLock = valueOperations.setIfAbsent("k1","v1");
    //如果占位成功，进行正常操作
    if(isLock){
        valueOperations.set("name","xxxx");
        String name = (String) valueOperations.get("name");
        System.out.println("name = "+name);
        //操作结束，删除锁
        redisTemplate.delete("k1");
    }else{
        System.out.println("有线程在使用，请稍后重试!!");
    }
}

(一)版本存在的问题：如果某个线程执行时抛出异常，那么锁就一直不会释放，导致错误

为了解决上面的问题，为锁加上失效时间，失效时间要略大于业务代码执行时间

//占位，如果key不存在则设置成功
Boolean isLock = valueOperations.setIfAbsent("k1","v1",5,TimeUnit.SECONDS);

但是此时还存在问题，就是删除的锁不是自己的，而是其他线程的。比如：A线程在执行业务代码的时候出现网络波动，但是此时A线程设置的锁到了失效时间，此时B线程拿到了锁开始执行业务代码，A线程执行完业务代码，去删锁，此时删掉的就是B线程设置的锁。造成误删的结果。

解决方法：在设置锁的时候设置的值为一个随机生成的值，删除锁的时候进行比对，如果是自己的锁就删除。

public void test3(){
    ValueOperations valueOperations = redisTemplate.opsForValue();
    String value = UUID.randomUUID().toString();
    //占位，如果key不存在则设置成功
    Boolean isLock = valueOperations.setIfAbsent("k1",value,5,TimeUnit.SECONDS);
    //如果占位成功，进行正常操作
    if(isLock){
        valueOperations.set("name","xxxx");
        String name = (String) valueOperations.get("name");
        System.out.println("name = "+name);
        if(valueOperations.get("k1").equal(value)){
           redisTemplate.delete("k1");
        }
    }else{
        System.out.println("有线程在使用，请稍后重试!!");
    }
}

但是此时还有一个问题，就是比对锁删除锁的操作不是原子性的，这就可能导致A服务器在比对成功之后，锁刚好过期，而此时B服务器又设置好了锁，此时A删除的就是B的锁。
处理方法：使用lua脚本保证操作原子性。

在resource下添加lua脚本，使得比较锁、删除锁是一个原子性的操作：(文件名为lock.lua)

if redis.call("get",KEYS[1])==ARGV[1] then
   return redis.call("del",KEYS[1])
else
   return 0
end

添加配置类：

@Bean
public DefaultRedisscript script(){
    DefaultRedisscript redisscript = new DefaultRedisscript<>();
    //与application.yml同级目录
    redisscript.setLocation(new ClassPathResource("lock.lua"));
    redisscript.setResultType(Boolean.class);
    return redisscript;
}

测试代码：

public void test3(){
    ValueOperations valueOperations = redisTemplate.opsForValue();
    String value = UUID.randomUUID().toString();
    //占位，如果key不存在则设置成功
    Boolean isLock = valueOperations.setIfAbsent("k1",value,5,TimeUnit.SECONDS);
    //如果占位成功，进行正常操作
    if(isLock){
        valueOperations.set("name","xxxx");
        String name = (String) valueOperations.get("name");
        System.out.println("name = "+name);
        //执行lua脚本
        redisTemplate.execute(redisscript, Collections.singletonList("k1"),value);
    }else{
        System.out.println("有线程在使用，请稍后重试!!");
    }
}

就是用List数据类型模拟消息队列，生产者使用lpush从队列左边添加消息，消费者使用rpop从右边消费消息。

例如：

生产者：
lpush queue msg1
lpush queue msg2
消费者：
rpop queue
rpop queue

一般编写消费者逻辑时，通过一个“死循环”实现，如果此时队列为空，那消费者依旧会频繁拉取消息，造成资源浪费。

while(true)
{
    String msg = jedis.rpop("queue");
}

Redis 提供阻塞式拉取消息的命令：brpop / blpop。

brpop key timeout

brpop key timeout：移除并返回最后一个值，同时需要传入一个超时时间（timeout），如果设置为0，则表示不设置超时，直到有新消息才返回，否则会在指定的超时时间后返回 NULL。

消费者如下：

Jedis jedis = new Jedis("192.168.56.31",6379);
System.out.println("开始监听");
while (true)
{
    List msg = jedis.brpop(0,"queue");
    System.out.println("接受消息：");
    //一般来说 一条消息分为两部分，第一部分是list的key，第二部分为value
    for (String m : msg){
      System.out.print(m + "");
    }
}

生产者如下：

Jedis jedis = new Jedis("192.168.56.31",6379);
Scanner sc = new Scanner(System.in);
while (true)
{
    System.out.println("输入发送的消息：");
    String msg = sc.next();
    jedis.lpush("queue",msg);
}

Redis 提供了 PUBLISH / SUBSCRIBE 命令，来完成发布、订阅的操作。

//订阅queue频道
SUBSCRIBE queue

//向queue频道发送一条消息
PUBLISH queue msg1

消费者：

public class Customer extends JedisPubSub {
     public void onMessage(String channel, String message) {
       System.out.println("接收到消息:" + channel + ":" + message);
     }
     public static void main(String[] args) {
       Jedis jedis = new Jedis("192.168.56.31",6379);
       //通过jedis订阅频道，需要一个JedisPubSub子类对象，并重写onMessage方法用于接受消息
       jedis.subscribe(new Customer(),"queue");
     }
}

生产者：

//第一个参数是ip地址，第二个参数是端口
Jedis jedis = new Jedis("192.168.56.31",6379); 
Scanner sc = new Scanner(System.in);
while (true){
   System.out.println("输入发送的消息：");
   String msg = sc.next();
   jedis.publish("queue",msg);
}