双十一大放送：深入了解性能提升大杀器—缓存

2021-11-11 本文已影响0人小螺旋丸

引言

在我们平时的开发工作中，对于一些复杂的业务接口可以通过优化业务逻辑来增加接口的吞吐量，但在更多时候，简单的优化业务逻辑，调节系统参数仍然无法满足性能提升的要求，为了系统性能的提升，我们选择将部分数据放入缓存当中，数据库则主要承担数据落盘的工作，合理，正确的使用缓存可以极大的提升系统的性能，大幅提升接口的响应速度！下面就让我们一起深入了解性能提升大杀器—缓存。

缓存的使用

缓存固然香，但绝不是万金油，盲目为了提升性能而滥用缓存往往会弄巧成拙，需要我们根据业务去具体分析。

哪些数据适合放入缓存

即时性、数据一致性要求不高的数据，例如商品分类
访问量大且不经常更新的数据，例如物流信息

读模式缓存的大体使用流程如下：

读模式缓存使用流程

特别注意：在开发中，凡是放入缓存中的数据我们都应该设置过期时间，这样就算没有主动更新缓存的机制也可以触发数据加载进入缓存的流程，避免因业务崩溃导致的数据永久不一致问题。

springboot整合redis

我们最常用的缓存中间件非redis莫属，redis是一款非常优秀的键值对存储数据库，springboot集成redis的方式十分简单：

首先，引入maven依赖：

<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-data-redis</artifactId>
   <version>1.5.8.RELEASE</version>
</dependency>
 <!--序列化使用-->
<dependency>
     <groupId>com.alibaba</groupId>
     <artifactId>fastjson</artifactId>
     <version>1.2.68</version>
 </dependency>

配置application.yml：

spring:
  redis:
    # ip地址
    host: localhost
    port: 6379
    password:
    timeout: 3000
    lettuce.pool.max-active: 10
    lettuce.pool.max-wait: 5
    lettuce.pool.max-idle: 8
    lettuce.pool.min-idle: 0

配置redisConfig（指定序列化方式）：

@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(factory);

        FastJsonRedisSerializer fastJsonRedisSerializer = new FastJsonRedisSerializer<>(Object.class);
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();

        /**
         * fastJson旧版本有漏洞，容易遭黑客攻击，新版本需要添加类型转换白名单
         * @see https://github.com/alibaba/fastjson/wiki/enable_autotype
         */
        ParserConfig.getGlobalInstance().addAccept("需要使用序列化的包名");

        // key采用String的序列化方式
        template.setKeySerializer(stringRedisSerializer);
        // hash的key也采用String的序列化方式
        template.setHashKeySerializer(stringRedisSerializer);
        // value序列化方式采用fastjson
        template.setValueSerializer(fastJsonRedisSerializer);
        // hash的value序列化方式采用fastjson
        template.setHashValueSerializer(fastJsonRedisSerializer);
        template.afterPropertiesSet();

        return template;
    }
}

使用redisTemplate操作redis：

@Autowired
private RedisTemplate<String, Object> redisTemplate;

@Test 
public void test(){
    String key = "key";
    String value= "value";
    long time = 60L;
    //设置或更新 key value
    redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);
    //key不存在时存入
    boolean isSave = redisTemplate.opsForValue().setIfAbsent(key, value, time, TimeUnit.SECONDS);
    //获取key的value
    Object o = redisTemplate.opsForValue().get(key);
    //删除key
    redisTemplate.delete(key);
    //设置key的过期时间
    redisTemplate.expire(key, time, TimeUnit.SECONDS);
    //获取过期时间
    long expireTime = redisTemplate.getExpire(key, TimeUnit.SECONDS);
    //判断key是否存在
    boolean isExist = redisTemplate.hasKey(key);
}

缓存雪崩、缓存穿透、缓存击穿问题

在大并发读的情景下有很大概率遇到缓存失效问题，缓存失效问题主要有以下三种：缓存雪崩，缓存击穿和缓存穿透。

缓存雪崩

缓存雪崩是指我们设置缓存的时候采用了相同的过期时间，导致缓存在某一时刻同时失效，恰巧那一时刻并发量很大，导致所有请求转发到数据库，数据库瞬时压力过重崩溃，从而导致系统崩溃，如大批鬼子进村。

解决：通常我们会给缓存在原来失效时间的基础上增加一个随机值，例如1-5分钟随机，这样会大大降低缓存过期时间的重复概率，从而避免缓存集体失效。

缓存击穿

某些被设置了过期时间的 key 可能被同一时间被超高并发访问，比如双十一定点抢购某一商品的信息，这种数据被称为热点数据。如果这个 key 在大量请求同时进来的时候正好过期失效，那么对于这个 key 的相关查询全部落到数据库，导致数据库崩溃，这种现象称之为缓存击穿，如八路军炸鬼子碉堡。

解决：使用分布式锁，在缓存同时失效的情况下，保证只放行一个请求读取数据库并更新缓存。

缓存穿透

缓存穿透是指查询一个一定不存在的数据，由于缓存不命中，程序会去查询数据库，但是数据库也无此记录，这将导致这个不存在的数据每次请求都要到数据库去查询，缓存形同虚设，当流量大时，数据库可能因此挂掉。如特务进城。

解决：缓存空结果、设置短的过期时间（简单粗暴）；布隆过滤器。

缓存数据一致性问题

为了解决缓存和数据库的数据一致性问题，主要有以下几种方法：

双写模式

顾名思义，双写模式是指写数据库的同时更新缓存，但这种模式只能保证数据的最终一致性，如图：

双写模式
用户1 和用户2 先后执行了数据库的写操作，然后更新缓存，理论上缓存的最新数据应该是用户2 写入的数据，但由于用户1 在写数据库的时候产生了卡顿，导致用户2 先写完数据并更新缓存，最后缓存更新为了用户1 写入的数据，导致数据库中的数据是用户2 写入的数据，但缓存中的数据却是用户1 写入的数据，产生了数据不一致问题。

但随着后续的数据更新，或者缓存过期后，可以保证数据的最终一致性。

失效模式

失效模式是指写数据的时候删除缓存，从而下次查询触发缓存更新。如图：

失效模式

但是很遗憾，这种模式也只能保证数据的最终一致性。

例如：用户1 和用户2 先后执行了数据库的写操作，并删除缓存，正常情况下缓存最终更新的数据应该是用户2 写入的数据，但由于用户2 的写操作卡顿，数据库没有写入完毕，这个时候用户3 读取数据的时候读取到的则是用户1 写入的数据，随之更新的缓存也是用户1 写入的数据，从而导致数据不一致的问题产生。

分布式读写锁

使用分布式读写锁可以完美解决缓存数据不一致的问题，想要读数据必须等待写数据整个操作完成，这个之后的小节里会详细介绍。

使用阿里中间件canal

canal是阿里推出的一个数据同步中间件，可以用于实时监听数据库的数据变动操作，原理是伪造成数据库的从节点，订阅binlog日志来实现实时监听。

这样当数据变动的时候，我们可以根据监听的具体操作去更新缓存，达到缓存数据一致性，但可能存在一定延迟。

canal在大数据系统中可以用来解决数据异构问题，可以根据订阅表的相关操作进行数据整合和分析，生成想要的结构数据，比如监听每个用户的浏览访问记录，进行整合分析，生成用户推荐表，这样用户浏览的时候便可以看到自己喜欢的内容。

canal

分布式锁

利用分布式锁我们可以很好的解决缓存击穿和数据一致性问题。

通过加锁操作我们可以让写数据变为原子性操作，不被其他线程干扰。

在单体应用环境下如果我们想要操作具备原子性可以使用 synchronized 关键字或者ReentrantLock解决，但在分布式环境下显然已不再适用，因为不同的应用没有共享JVM，本地锁也就失去了意义。

自己手写一个分布式锁

在分布式环境下，我们可以使用 redis 来实现分布式锁：

定义一个 key 的值。
当请求进来，判断 redis 中是否存在此 key，若不存在，新增 key，生成一个随机值作为value，并设置过期时间，业务执行之后，获取 key 对应的 value 与之前生成的 value 比较，若相等，删除 key；若 key 存在，等待一段时间，再次尝试获取，直到获取到为止。

redis 的每个命令都是原子性的，所以我们不用考虑在命令执行的过程中被其他进程干扰的问题。同时加锁操作和解锁操作必须保证其原子性，否则在大并发情况下就会出现问题。

加锁 = 判断 key + 新增 key + 设置过期时间
解锁 = 判断 key 的 value 是否一致 + 删除 key

加锁操作 redis 有原生的命令可以支持，解锁操作则需要使用 lua 脚本解决，否则无法保证其原子性，设计如下：

redis锁

大致代码如下：

public void testLock() {
    String uuid = UUID.randomUUID().toString();
    Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock",uuid,300,TimeUnit.SECONDS);
    if (lock) { 
       System.out.println("获取分布式锁成功...");
       try {
          //执行业务代码。。。
       } finally {
            String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
            //删除锁
            redisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class) , Arrays.asList("lock"), uuid);
       }
    } else {
       System.out.println("获取分布式锁失败...等待重试");
       try {
         Thread.sleep(200); 
       } catch (Exception ignored){}
       //自旋获取锁
       testLock();
    }
}

使用 redisson 实现分布式锁

前面我们自己利用 redis 实现了分布式锁的基本功能，虽然比较简陋，但有助于我们理解其中的原理，在实际的开发中，我们可以直接利用 Redisson 来实现相关功能！

Redisson 是架设在 Redis 基础上的一个 Java 驻内存数据网格（In-Memory Data Grid）。充分利用了 Redis 键值数据库提供的一系列优势，基于Java 实用工具包中常用接口，为使用者提供了一系列具有分布式特性的常用工具类。使得原本作为协调单机多线程并发程序的工具包获得了协调分布式多机多线程并发系统的能力，大大降低了设计和研发大规模分布式系统的难度。同时结合各富特色的分布式服务，进一步简化了分布式环境中程序相互之间的协作。

springboot 集成 Redisson

引入 maven 依赖：

<dependency>
     <groupId>org.redission</groupId>
     <artifactId>redission</artifactId>
     <version>3.12.0</version>
 </dependency>

使用方法如下：

配置 MyRedissonConfig：

@Configuration
public class MyRedissonConfig {

    /**
     * 所有对Redisson的使用都是通过RedissonClient
     * @return
     * @throws IOException
     */
    @Bean
    public RedissonClient redisson() throws IOException {
        //1、创建配置
        Config config = new Config();
        //集群模式
        /* config.useClusterServers()
              .addNodeAddress("127.0.0.1:6379","127.0.0.1:6380")*/
       //单点模式
       //Redis url should start with redis:// or rediss://（启用SSL）
        config.useSingleServer()
              .setAddress("redis://127.0.0.1:6379");
        //2、根据Config创建出RedissonClient实例
        RedissonClient redissonClient = Redisson.create(config);
        return redissonClient;
    }

}

引入 MyRedissonConfig 使用：

@Autowired
private RedissonClient redissonClient;

@Test
private void testLock() {
    //获取一把锁，只要锁名字相同，就是一把锁
    RLock lock = redisson.getLock("my-lock");
    //加锁
    lock.lock();
    // 尝试加锁，最多等待 100 秒，上锁以后 10 秒自动解锁 
    //boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
    // 加锁以后 10 秒钟自动解锁
    //lock.lock(10, TimeUnit.SECONDS);
    try {
      //执行业务代码
    } finally {
        //解锁    
        lock.unlock()
    }
    
}

也许大家会有疑问，在使用 lock 方法没有指定过期时间的情况下，如果负责储存分布式锁的 redisson 节点宕机后，这个锁正好处于锁定状态，会出现锁死的情况，为了避免这种情况的发生，redission 对锁设置了默认的过期时间为30s，可以通过修改Config.lockWatchdogTimeout指定，所以就算宕机，锁也会在指定时间内过期，也就不会出现死锁的情况。

此外，redisson 内部提供了一个监控锁的看门狗，它的作用是在 redisson 实例被关闭前，不断对锁进行续期，如果你的业务执行时间较长，它可以为锁自动续期，保证业务执行完毕后再释放锁，不被其他进程干扰，但一般情况下，我们会事先评估业务完成所需要的时间，设置锁的过期时间>=业务完成的时间，因为自动续期会消耗一定的性能，不过最终采用何种方式加锁具体还要根据项目需求而定。

redisson 还支持读写锁 ReadWriteLock ，闭锁 countDownLatch，信号量 Semaphore 等，用法跟 java.util.concurrent 包下对应类的用法基本是一样的，这里就不再一一列举啦，同学们可以自行了解。