【业务场景实战】如何优雅地进行缓存预热？

从Java基础到中间件再到微服务，我们学了这么多，但遇到真实项目的时候，还是不会根据所学知识，对项目进行改造；或者太久不用早已忘记。学会用才是走得更远！

缓存穿透、雪崩，大家都不陌生，但其中针对的解决方案，有自己手动去实现过吗？

下面带大家去实现一下！

一、问题和解决方案

场景：**缓存在同一时间大面积的失效，导致大量的请求都直接落到了数据库上，对数据库造成了巨大的压力。** 缓存服务宕机也会导致缓存雪崩现象，导致所有的请求都落到了数据库上。

为了保证非热点数据不占用太多内存空间，我们设置了逻辑过期时间。

但是如果热点数据出现过期就会造成缓存穿透、雪崩这些问题。为了解决这些问题，我们需要对已经过期或者将要过期的数据进行缓存重建。

重新导入数据到Redis中，并且重新设置逻辑过期时间。缓存重建需要对一些热点数据进行预热。之前我是这么预热的

@Test
    void testSaveShop() {
        //测试id=1，时间10s
        redisUtils.saveShop2Redis(1L, 120L);
    }

这样一个一个的写入id号，效率着实有点太慢了，而且万一不记得了，程序就会出现报错了。

针对缓存重建问题，我这里介绍使用缓存预热的两种方法来实现

二、缓存预热两种方案

1、定时任务

使用`@EnableScheduling`开启定时任务

1）获取ID列表

我们在mapper上创建方法，获取数据id号

@Select("SELECT id FROM tb_shop")
    List<Integer> selectAllIds();

2）缓存重建逻辑

这段缓存重建的逻辑：
先根据传入的id号从数据库中获取值，
封装逻辑过期时间和数据，最后将数据进行写入

//缓存重建
public void saveShop2Redis(Long id, Long expireSecond) {
    String key = CACHE_SHOP_KEY + id;
    //1、查询店铺数据
    Shop shop = shopMapper.selectById(id);
    //2、封装逻辑过期时间
    RedisData redisData = new RedisData();
    redisData.setExpireTime(LocalDateTime.now().plusSeconds(expireSecond));
    redisData.setData(shop);
    //3、写入redis
    stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(redisData));
}

3）定时任务缓存类

因为不需要特别复杂的逻辑，所以我这里就使用最为简单的spring自带的定时任务。
创建缓存重建定时任务类

- 首先开启定时任务注解

- 调用shopMapper方法，查询所有id号，遍历

- 将遍历的id号传入saveShop2Redis方法中，并设置逻辑过期时间60秒

- 日志打印输出

@Component

@Slf4j

public class CachePreheatTask {

    @Resource
    private RedisUtils redisUtils;
    @Resource
    private ShopMapper shopMapper;

    // 执行缓存预热任务的方法
    @Scheduled(cron = "0 19 9 * * ?")
    public void preheatCache() {
        // 执行缓存预热逻辑
        List<Integer> selectAllIds = shopMapper.selectAllIds();
        for (Integer allId : selectAllIds) {
            //测试id=1，时间10s
            redisUtils.saveShop2Redis(Long.valueOf(allId), 60L);
            log.debug("缓存数据预热成功，id为：{}" ,allId);
        }
    }
}

这里的表达式：cron = "0 19 9 * * ?"
表示的是每天上午9点19执行定时任务
调用 `selectAllIds`方法，把数据库表中的id查询出来，然后进行遍历
再利用for循环，把每次查询出的id传给`saveShop2Redis`方法进行缓存重建
为了方便测试，我设置逻辑过期时间为60秒。

`@Scheduled`注解是Spring框架中用于创建定时任务的注解，它有三个不同类型的参数：`cron`、`fixedDelay`、`fixedRate`，分别用于不同的定时任务需求。

1、 `cron`参数：用于指定一个cron表达式，可以精确控制任务的执行时间。cron表达式是一个字符串，包含六个或七个空格分隔的时间字段，用于指定秒、分、时、日、月、周几等时间点。例如，`"0 * * * * ?"`表示每分钟执行一次。

2、 `fixedDelay`参数：用于指定任务执行结束后到下一次任务开始的间隔时间，单位为毫秒。即任务的执行周期是任务结束后延迟指定的时间后再执行。

例如，`@Scheduled(fixedDelay = 1000)`表示任务执行结束后延迟1秒后再执行。

3.、`fixedRate`参数：用于指定任务开始执行后到下一次任务开始的间隔时间，单位为毫秒。即任务的执行周期是任务开始后固定的时间间隔再执行。例如，`@Scheduled(fixedRate = 1000)`表示任务开始后每隔1秒执行一次。

这些参数可以根据实际需求来选择，`cron`表达式适用于需要精确控制执行时间的场景，`fixedDelay`适用于任务执行时间不固定的场景，`fixedRate`适用于固定频率执行任务的场景。

表达式
表达式	意义
每隔5秒钟执行一次	/5 * * * ?
每隔1分钟执行一次	0 * /1 * * * ?
每天1点执行一次	0 0 1 * * ?
每天23点55分执行一次	0 55 23 * * ？

这样就能达到我想要的效果，可以随便设置定时任务的执行时间，这样就可以提前进行预热了。

2、消息队列

下面再介绍一种可以进行数据预热的方式——消息队列。

思考一下：我们的诉求是什么？

我们需要将数据进行预热，那我们是不是要拿到数据的id号。

拿到了id号呢，我们怎么让程序自动地去执行这段重建逻辑呢？

对的，使用消息队列，把id号传给消息队列，然后在项目启动的时候，让生产者去发送这个消息。消费者拿到消息之后，就会去执行重建的逻辑了。

这里一些关于MQ配置什么的我就不写了，都是固定的，

1）生产者代码

@Component
public class MyMessageProducer {
    @Resource
    private RabbitTemplate rabbitTemplate;
    // 向指定交换机发送消息
    public void sendMessage(String exchange, String routingKey, String message) {
        //将消息发送到指定的交换机和路由键
        rabbitTemplate.convertAndSend(exchange,routingKey,message);

    }
}

2）消费者代码

@Component
@Slf4j
public class MyMessageConsumer {

    @Resource
    private RedisUtils redisUtils;

    /**
     * 接收消息的方法
     *
     * @param message
     * @param channel
     * @param deliveryTag
     */
    //使用@SneakyThrows注解简化异常处理
    @SneakyThrows
    //使用该注解指定程序要监听的队列，，并设置消息的确认机制为手动
    @RabbitListener(queues = {"hmdp_queue"}, ackMode = "MANUAL")
    //@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag 用于从消息头中获取投递标签deliveryTag
    //在mq中，每条消息都会被分配一个唯一投递标签，用于标识该消息在通道中的投递状态和顺序，使用该注解可以从消息头中获取该投递标签，并将其赋值给deliveryTag参数，
    public void receiveMessage(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        long shopId = Long.parseLong(message);
        redisUtils.saveShop2Redis(shopId, 60L);
        log.info("收到消息，传入的shopId为：{}", message+"，缓存数据预热成功!");
        //手动确认消息，消息确认标志设置为false，消息才能被确认
        channel.basicAck(deliveryTag, false);
    }
}

3）创建队列交换机

在程序执行前创建好交换机和对列

public class biInitMain {
    public static void main(String[] args) {
        try {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.88.130");
            factory.setPort(5672);
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();

            String EXCHANGE_NAME = "hmdp_exchange";
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");

            // 声明一个队列，并且设置持久化消息
            String queueName = "hmdp_queue";
            String ROUTING_KEY="hmdp_routingKey";
            channel.queueDeclare(queueName, true, false, false, null);
            //队列绑定交换机，routing_key用于指定消息应该发送到哪个队列。
            channel.queueBind(queueName, EXCHANGE_NAME, ROUTING_KEY);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

4）发送消息

用于获取热点id并将其发送到消息队列。这个程序应该只执行一次，以确保不会重复发送相同的id。

`@PostConstruct`注解会让项目启动时初始化这段代码，被执行一次。这样消息也就被发送给消费者了，缓存重建的逻辑也就执行成功了

@PostConstruct
    public void init() {
        myMessageProducer.sendMessage("hmdp_exchange"
,"hmdp_routingKey",String.valueOf(1L));
       
    }

看看控制台

其实代码到这里还是有点小问题的，细心的兄弟应该看出这里的问题了。

对的，之前使用定时任务，获取的是所有数据的id,获取的是所有的数据。
而这次消息队列改造，传入的是一个固定的id值。其实这里应该需要去获取一些热点数据id，再将这些id号传给方法。其中涉及到日志记录、监控数据判断是否是热点数据。

到这里我的缓存预热就结束了，其实就类似于项目的一个小优化的一样