redis常见问题及解决方案

缓存预热

定义

缓存预热是一种优化方案，它可以提高用户的使用体验。
缓存预热是指在系统启动的时候，先把查询结果预存到缓存中，以便用户后面查询时可以直接从缓存中读取，节省用户等待时间

实现思路

把需要缓存的方法写在初始化方法中，让程序启动时自动加载数据并缓存数据。
把需要缓存的方法挂在某个页面或是后端接口上，手动触发缓存预热。
设置定时任务，定时进行缓存预热。

解决方案

使用 @PostConstruct 初始化白名单数据

缓存雪崩(大量数据同时失效/Redis 崩了，没有数据了)

定义

缓存雪崩是指在短时间内大量缓存同时过期，导致大量请求直接查询数据库，从而对数据库造成巨大压力，严重情况下可能会导致数据库宕机

解决方案

加锁排队：起到缓冲作用，防止大量请求同时操作数据库，但缺点是增加了系统的响应时间，降低了系统的吞吐量，牺牲一部分用户体验。
设置二级缓存：二级缓存是除了 Redis 本身的缓存，再设置一层缓存，当 Redis 失效后，就先去查询二级缓存
随机化过期时间：为了避免缓存同时过期，可以设置缓存时添加随机时间，这样就可以极大的避免大量缓存同时失效
redis 缓存集群实现高可用
主从 + 哨兵
Redis 集群
开启Redis 持久化机制 aof / rdb，尽快恢复缓存集群

服务降级
Hystrix 或者 sentinel 限流 & 降级

// 缓存原本的失效时间
int exTime = 10 * 60;
// 随机数⽣成类
Random random = new Random();
// 缓存设置
jedis.setex(cacheKey, exTime+random.nextInt(1000) , value);

缓存穿透(黑客攻击/空数据/穿过 Redis 和数据库)

定义

缓存穿透是指查询数据库和缓存都无数据，因此每次请求都会去查询数据库

解决方案

**缓存空结果：**对查询的空结果也进行缓存，如果是集合，可以缓存一个空的的集合，如果是缓存单个对象，可以字段标识来区分，避免请求穿透到数据库。
**布隆过滤器处理：**将所有可能对应的数据为空的 key 进行统一的存放，并在请求前做拦截，避免请求穿透到数据库（这样的方式实现起来相对麻烦，比较适合命中不高，但是更新不频繁的数据）。
双锁锁策略机制

package com.redis.redis01.service;

import com.redis.redis01.bean.RedisBs;
import com.redis.redis01.mapper.RedisBsMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.beans.Transient;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
@Service
public class RedisBsService {

    //定义key前缀/命名空间
    public static final String CACHE_KEY_USER = "user:";
    @Autowired
    private RedisBsMapper mapper;
    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    private static ReentrantLock lock = new ReentrantLock();

    /**
     * 业务逻辑没有写错，对于中小长(qps<=1000)可以使用，但是大厂不行：大长需要采用双检加锁策略
     *
     * @param id
     * @return
     */
    @Transactional
    public RedisBs findUserById(Integer id,int type,int qps) {
        //qps<=1000
        if(qps<=1000){
            return qpsSmall1000(id);
        }
        //qps>1000
        return qpsBig1000(id, type);
    }

    /**
     * 加强补充，避免突然key失效了，或者不存在的key穿透redis打爆mysql，做一下预防，尽量不出现缓存击穿的情况，进行排队等候
     * @param id
     * @param type 0使用synchronized重锁，1ReentrantLock轻量锁
     * @return
     */
    private RedisBs qpsBig1000(Integer id, int type) {
        RedisBs redisBs = null;
        String key = CACHE_KEY_USER + id;
        //1先从redis里面查询，如果有直接返回，没有再去查mysql
        redisBs = (RedisBs) redisTemplate.opsForValue().get(key);
        if (null == redisBs) {
            switch (type) {
                case 0:
                    //加锁，假设请求量很大，缓存过期，大厂用，对于高qps的优化，进行加锁保证一个请求操作，让外面的redis等待一下，避免击穿mysql
                    synchronized (RedisBsService.class) {
                        //第二次查询缓存目的防止加锁之前刚好被其他线程缓存了
                        redisBs = (RedisBs) redisTemplate.opsForValue().get(key);
                        if (null != redisBs) {
                            //查询到数据直接返回
                            return redisBs;
                        } else {
                            //数据缓存
                            //查询mysql，回写到redis中
                            redisBs = mapper.findUserById(id);
                            if (null == redisBs) {
                                // 3 redis+mysql都没有数据，防止多次穿透(redis为防弹衣，mysql为人，穿透直接伤人，就是直接访问mysql)，优化：记录这个null值的key，列入黑名单或者记录或者异常
                                return new RedisBs(-1, "当前值已经列入黑名单");
                            }
                            //4 mysql有，回写保证数据一致性
                            //setifabsent
                            redisTemplate.opsForValue().setIfAbsent(key, redisBs,7l, TimeUnit.DAYS);
                        }
                    }
                    break;
                case 1:
                    //加锁，大厂用，对于高qps的优化，进行加锁保证一个请求操作，让外面的redis等待一下，避免击穿mysql
                    lock.lock();
                    try {
                        //第二次查询缓存目的防止加锁之前刚好被其他线程缓存了
                        redisBs = (RedisBs) redisTemplate.opsForValue().get(key);
                        if (null != redisBs) {
                            //查询到数据直接返回
                            return redisBs;
                        } else {
                            //数据缓存
                            //查询mysql，回写到redis中
                            redisBs = mapper.findUserById(id);
                            if (null == redisBs) {
                                // 3 redis+mysql都没有数据，防止多次穿透(redis为防弹衣，mysql为人，穿透直接伤人，就是直接访问mysql)，优化：记录这个null值的key，列入黑名单或者记录或者异常
                                return new RedisBs(-1, "当前值已经列入黑名单");
                            }
                            //4 mysql有，回写保证数据一致性
                            redisTemplate.opsForValue().set(key, redisBs);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally {
                        //解锁
                        lock.unlock();
                    }
            }
        }
        return redisBs;
    }
    private RedisBs qpsSmall1000(Integer id) {
        RedisBs redisBs = null;
        String key = CACHE_KEY_USER + id;
        //1先从redis里面查询，如果有直接返回，没有再去查mysql
        redisBs = (RedisBs) redisTemplate.opsForValue().get(key);
        if (null == redisBs) {
            //2查询mysql，回写到redis中
            redisBs = mapper.findUserById(id);
            if (null == redisBs) {
                // 3 redis+mysql都没有数据，防止多次穿透(redis为防弹衣，mysql为人，穿透直接伤人，就是直接访问mysql)，优化：记录这个null值的key，列入黑名单或者记录或者异常
                return new RedisBs(-1, "当前值已经列入黑名单");
            }
            //4 mysql有，回写保证数据一致性
            redisTemplate.opsForValue().set(key, redisBs);
        }
        return redisBs;
    }

}

package com.redis.redis01.service;

import com.google.common.collect.Lists;
import com.redis.redis01.bean.RedisBs;
import com.redis.redis01.mapper.RedisBsMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
@Service
public class BitmapService {
    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    private static ReentrantLock lock = new ReentrantLock();
    @Autowired
    private RedisBsMapper redisBsMapper;

    /**
     * 场景一：布隆过滤器解决缓存穿透问题(null/黑客攻击)；利用redis+bitmap实现
     * 有可能有，没有一定没有
     *                                                    无-------------》mysql查询
     *                     有--------》redis查询----------》有-----------》返回
     * 请求-----》布隆过滤器-----------》
     *                      无-------终止
     *
     * @param type:0初始化，1常规查询
     */
    public void booleanFilterBitmap(int type, Integer id) {
        
        switch (type) {
            case 0://初始化数据
                for (int i = 0; i < 10; i++) {
                    RedisBs initBs = RedisBs.builder().id(i).name("赵三" + i).phone("1580080569" + i).build();
                    //1 插入数据库
                    redisBsMapper.insert(initBs);
                    //2 插入redis
                    redisTemplate.opsForValue().set("customer:info" + i, initBs);
                }
                //3 将用户id插入布隆过滤器中,作为白名单
                for (int i = 0; i < 10; i++) {
                    String booleanKey = "customer:booleanFilter:" + i;
                    //3.1 计算hashvalue
                    int abs = Math.abs(booleanKey.hashCode());
                    //3.2 通过abs和2的32次方取余，获得布隆过滤器/bitmap对应的下标坑位/index
                    long index = (long) (abs % Math.pow(2, 32));
                    log.info("坑位:{}", index);
                    //3.3 设置redis里面的bitmap对应类型的白名单
                    redisTemplate.opsForValue().setBit("whiteListCustomer", index, true);
                }
                break;
            case 1://常规查询
                //1 获取当前传过来的id对应的哈希值
                String inputBooleanKey = "customer:booleanFilter:" + id;
                int abs = Math.abs(inputBooleanKey.hashCode());
                long index = (long) (abs % Math.pow(2, 32));
                Boolean whiteListCustomer = redisTemplate.opsForValue().getBit("whiteListCustomer", index);
                //加入双检锁
                //加锁，大厂用，对于高qps的优化，进行加锁保证一个请求操作，让外面的redis等待一下，避免击穿mysql
                lock.lock();
                try {
                    if (null == whiteListCustomer) {
                        whiteListCustomer = redisTemplate.opsForValue().getBit("whiteListCustomer", index);
                        if (null != whiteListCustomer && whiteListCustomer) {//布隆过滤器中存在，则可能存在
                            //2 查找redis
                            Object queryCustomer = redisTemplate.opsForValue().get("customer:info" + id);
                            if (null != queryCustomer) {
                                log.info("返回客户信息:{}", queryCustomer);
                                break;
                            } else {
                                //3 redis没有查找mysql
                                RedisBs userById = redisBsMapper.findUserById(id);
                                if (null != userById) {
                                    log.info("返回客户信息:{}", queryCustomer);
                                    redisTemplate.opsForValue().set("customer:info" + id, userById);
                                    break;
                                } else {
                                    log.info("当前客户信息不存在:{}", id);
                                    break;
                                }
                            }
                        } else {//redis没有，去mysql中查询
                            //3 redis没有查找mysql
                            RedisBs userById = redisBsMapper.findUserById(id);
                            if (null != userById) {
                                log.info("返回客户信息:{}", userById);
                                redisTemplate.opsForValue().set("customer:info" + id, userById);
                                break;
                            } else {
                                log.info("当前客户信息不存在:{}", id);
                                break;
                            }
                        }

                    }
                } finally {
                    lock.unlock();
                }
                log.info("当前客户信息不存在:{}", id);

                break;
            default:
                break;
        }
    }
}

缓存击穿(热点数据/刚失效/定点打击)

定义

缓存击穿是指某个经常使用的缓存，在某一个时刻恰好失效了（例如缓存过期），并且此时刚好有大量的并发请求，这些请求就会给数据库造成巨大的压力

解决方案

**加锁排队：**和处理缓存雪崩的加锁类似，都是在查询数据库的时候加锁排队，缓存操作请求以此来减少服务器的运行压力。
**设置永不过时：**对于某些经常使用的缓存，我们可以设置为永不过期，这样就能保证缓存的稳定性，但要注意在数据更改后，要及时更新此热点缓存，否则就会造成查询结果误差。

总结

脑裂

分布式session

分布式锁

分布式锁需要的条件和刚需
- 独占性
  - 任何时刻有且只有一个线程持有这个锁
- 高可用
  - 若redis集群环境下，不能因为某一个节点挂了而出现获取锁和释放锁失败的情况
  - 高并发请求下，依旧性能很好
- 防死锁
  - 不能出现死锁问题，必须有超时重试机制或者撤销操作，有个终止跳出的途径
- 不乱抢
  - 防止张冠李戴，只能解锁自己的锁，不能把别人的锁给释放了
- 重入性
  - 同一节点的同一线程如果获得锁之后，他可以再次获取这个锁

v 8.0 其实面对不是特别高的并发场景足够用了，单机redis也够用了

要兼顾锁的重入性
- setnx不满足了，需要hash结构的hset
上锁和解锁都用 Lua 脚本来实现原子性
引入工厂模式 DistributedLockFactory，实现 Lock 接口，实现redis的可重入锁
- lock() 加锁的关键逻辑
  - 加锁实际上就是在redis中，给Key键设置一个值，为避免死锁，并给定一个过期时间
  - 自旋
  - 续期
- unlock() 解锁关键逻辑
- 将 Key 键删除，但是也不能乱删，只能自己删自己的锁
实现自动续期功能的完善，后台自定义扫描程序，如果规定时间内没有完成业务逻辑，会调用加钟自动续期的脚本

    @Autowired
    private DistributedLockFactory distributedLockFactory;


    /**
     * v8.0  实现自动续期功能的完善，后台自定义扫描程序，如果规定时间内没有完成业务逻辑，会调用加钟自动续期的脚本
     *
     * @return
     */
    public String sale() {
        String retMessage = "";

        Lock redisLock = distributedLockFactory.getDistributedLock("redis");
        redisLock.lock();
        try {
            //1 查询库存信息
            String result = stringRedisTemplate.opsForValue().get("inventory001");
            //2 判断库存是否足够
            Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result);
            //3 扣减库存
            if (inventoryNumber > 0) {
                stringRedisTemplate.opsForValue().set("inventory001", String.valueOf(--inventoryNumber));
                retMessage = "成功卖出一个商品，库存剩余: " + inventoryNumber;

                // 演示自动续期的的功能
//                try {
//                    TimeUnit.SECONDS.sleep(120);
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
            } else {
                retMessage = "商品卖完了，o(╥﹏╥)o";
            }
        } finally {
            redisLock.unlock();
        }
        return retMessage + "\t" + "服务端口号：" + port;
    }


    /**
     * v7.0     兼顾锁的可重入性   setnx不满足了，需要hash结构的hset
     * 上锁和解锁都用 Lua 脚本实现原子性
     * 引入工厂模式 DistributedLockFactory    实现Lock接口 ，实现 redis的可重入锁
     *
     * @return
     */
//    //private Lock redisDistributedLock = new RedisDistributedLock(stringRedisTemplate, "xfcyRedisLock");
//
//    public String sale() {
//        String retMessage = "";
//
//        Lock redisLock = distributedLockFactory.getDistributedLock("redis");
//        redisLock.lock();
//
//        //redisDistributedLock.lock();
//        try {
//            //1 查询库存信息
//            String result = stringRedisTemplate.opsForValue().get("inventory001");
//            //2 判断库存是否足够
//            Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result);
//            //3 扣减库存
//            if (inventoryNumber > 0) {
//                stringRedisTemplate.opsForValue().set("inventory001", String.valueOf(--inventoryNumber));
//                retMessage = "成功卖出一个商品，库存剩余: " + inventoryNumber;
//                System.out.println(retMessage);
//
//                // 测试可重入性
//                //testReEntry();
//
//            } else {
//                retMessage = "商品卖完了，o(╥﹏╥)o";
//            }
//        } finally {
//            redisLock.unlock();
//            //redisDistributedLock.unlock();
//        }
//        return retMessage + "\t" + "服务端口号：" + port;
//    }
//
//    private void testReEntry() {
//        Lock redisLock = distributedLockFactory.getDistributedLock("redis");
//        redisLock.lock();
//
//        //redisDistributedLock.lock();
//        try {
//            System.out.println("测试可重入锁");
//        } finally {
//            redisLock.unlock();
//            //redisDistributedLock.unlock();
//        }
//    }

package com.xfcy.mylock;

import cn.hutool.core.util.IdUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.locks.Lock;

/**
 * @author 晓风残月Lx
 * @date 2023/4/1 22:14
 */
@Component
public class DistributedLockFactory {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    private String lockName;
    private String uuidValue;

    public DistributedLockFactory() {
        this.uuidValue = IdUtil.simpleUUID();
    }

    public Lock getDistributedLock(String lockType) {
        if (lockType == null) {
            return null;
        }
        if (lockType.equalsIgnoreCase("REDIS")) {
            this.lockName = "xfcyRedisLock";
            return new RedisDistributedLock(stringRedisTemplate, lockName, uuidValue);
        }else if (lockType.equalsIgnoreCase("ZOOKEEPER")) {
            this.lockName = "xfcyZookeeperLock";
            // TODO zoookeeper 版本的分布式锁
            return null;
        }else if (lockType.equalsIgnoreCase("MYSQL")){
            this.lockName = "xfcyMysqlLock";
            // TODO MYSQL 版本的分布式锁
            return null;
        }
        return null;
    }

}

package com.xfcy.mylock;

import cn.hutool.core.util.IdUtil;
import com.sun.org.apache.xpath.internal.operations.Bool;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;

import java.util.Arrays;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * @author 晓风残月Lx
 * @date 2023/4/1 21:38
 * 自研的redis分布式锁，实现 Lock 接口
 */
// @Component 引入DistributedLockFactory工厂模式，从工厂获得即可
public class RedisDistributedLock implements Lock {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    private String lockName;    // KEYS[1]
    private String uuidValue;   // ARGV[1]
    private long expireTime;    // ARGV[2]

    public RedisDistributedLock(StringRedisTemplate stringRedisTemplate, String lockName, String uuidValue) {
        this.stringRedisTemplate = stringRedisTemplate;
        this.lockName = lockName;
        this.uuidValue = uuidValue + ":" + Thread.currentThread().getId();
        this.expireTime = 30L;
    }

    @Override
    public void lock() {
        tryLock();
    }

    @Override
    public boolean tryLock() {
        try {
            tryLock(-1L, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
           e.printStackTrace();
        }
        return false;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        if (time == -1L) {
            String script = "if redis.call('exists',KEYS[1]) == 0 or redis.call('hexists',KEYS[1],ARGV[1]) == 1 then " +
                    "redis.call('hincrby',KEYS[1],ARGV[1],1)  " +
                    "redis.call('expire',KEYS[1],ARGV[2]) " +
                    "return 1 " +
                    "else " +
                    "return 0 " +
                    "end";
            System.out.println("lockName = " + lockName +"\t" + "uuidValue = " + uuidValue);
            while (!stringRedisTemplate.execute(new DefaultRedisScript<>(script, Boolean.class), Arrays.asList(lockName), uuidValue, String.valueOf(expireTime))) {
                // 暂停 60ms
                Thread.sleep(60);
            }
            // 新建一个后台扫描程序，来监视key目前的ttl，是否到我们规定的 1/2 1/3 来实现续期
            resetExpire();
            return true;
        }
        return false;
    }

    @Override
    public void unlock() {
        String script = "if redis.call('HEXISTS',KEYS[1],ARGV[1]) == 0 then " +
                "return nil " +
                "elseif redis.call('HINCRBY',KEYS[1],ARGV[1],-1) == 0 then  " +
                "return redis.call('del',KEYS[1]) " +
                "else  " +
                "return 0 " +
                "end";
        // nil = false  1 = true  0 = false
        Long flag = stringRedisTemplate.execute(new DefaultRedisScript<>(script, Long.class), Arrays.asList(lockName), uuidValue, String.valueOf(expireTime));
        if (null == flag) {
            throw new RuntimeException("this lock doesn't exists0");
        }
    }

    private void resetExpire() {
        String script = "if redis.call('HEXISTS',KEYS[1],ARGV[1]) == 1 then " +
                "return redis.call('expire',KEYS[1],ARGV[2]) " +
                "else " +
                "return 0 " +
                "end";
        new Timer().schedule(new TimerTask() {
            @Override
            public void run() {
                if (stringRedisTemplate.execute(new DefaultRedisScript<>(script, Boolean.class), Arrays.asList(lockName), uuidValue, String.valueOf(expireTime))) {
                    resetExpire();
                }
            }
        }, (this.expireTime * 1000) / 3);
    }


    // 下面两个用不上
    // 下面两个用不上
    // 下面两个用不上

    @Override
    public void lockInterruptibly() throws InterruptedException {

    }

    @Override
    public Condition newCondition() {
        return null;
    }
}