Java并发成神之路-精通JUC并发工具十八般武艺——实现高性能缓存(十)

发表于 2021-12-11 更新于 2022-08-19 分类于多线程本文字数： 12k 阅读时长 ≈ 21 分钟

从最简单版本缓存入手——HashMap

代码如下所示，第一次查询缓存Map，找不到就会查询数据库或者通过一些逻辑计算获取值，然后将获取到的结果保存到map中，并且返回

public class YsCache1 {

    private final HashMap<String, Integer> cache = new HashMap<>();

    public Integer computer(String userId) throws InterruptedException {
        Integer result = cache.get(userId);
        //先检查HashMap里面有没有保存过之前的计算结果
        if (result == null) {
            //如果缓存中找不到，那么需要现在计算一下结果，并且保存到HashMap中
            result = doCompute(userId);
            cache.put(userId, result);
        }
        return result;
    }

    private Integer doCompute(String userId) throws InterruptedException {
        TimeUnit.SECONDS.sleep(5);
        return new Integer(userId);
    }

    public static void main(String[] args) throws InterruptedException {
        YsCache1 ysCache1 = new YsCache1();
        System.out.println("开始获取信息");
        Integer result = ysCache1.computer("13");
        System.out.println("第一次获取信息=" + result);
        result = ysCache1.computer("13");
        System.out.println("第二次获取信息=" + result);
    }
}

问题

线程安全问题
上面的代码写法在并发情况下是不安的，比如两个线程同时去查，然后发现没有，同时去进行计算，同时HashMap也不是线程安全的，也有可能会引起一些HashMap自身线程安全的问题

给HashMap加final关键字

属性被声明未final后，该变量则只能被赋值一次，且一旦被赋值，final的变量就不能在被改变
所以我们把它加上final关键字，增强安全性

并发安全要保证——用synchronized实现

解决办法就是可以在computer()获取数据的方法上面添加synchronized关键字，使用锁保证线程安全

问题

性能差(使用缓存本身就是为了提高性能的，加锁之后，只能但线程访问，会对性能造成影响)
代码复用能力差(缓存逻辑和计算结果的业务代码写在同一个类里面，复用性比较差)

代码又重构空间——用装饰着模式

我们假设ExpensiveFunction类是耗时计算的实现类，实现了Computable接口，但是其本身不具备缓存功能，也不需要考虑缓存的事情
优化后的实现如下所示：
每次如果有不同的缓存需求，只需要使用不同的实现就可以了

/**
 * 有一个计算函数computer，用来代表耗时计算，每个计算器都要实现这个接口，这样就可以实现无侵入的缓存功能
 */
public interface Computable<A,V> {
    V compute(A arg) throws Exception;
}

/**
 * 耗时计算的实现类，实现了computable接口，但是本身不具备缓存能力，不需要考虑缓存的事情
 */
public class ExpensiveFunction implements Computable<String,Integer>{
    @Override
    public Integer compute(String arg) throws Exception {
        Thread.sleep(5000);
        return new Integer(arg);
    }
}

public class YsCache2<A, V> implements Computable<A, V> {
    private final Map<A, V> cache = new HashMap<>();
    private Computable<A, V> computable;

    public YsCache2(Computable<A, V> computable) {
        this.computable = computable;
    }

    @Override
    public synchronized V compute(A arg) throws Exception {
        System.out.println("进入缓存机制");
        V result = cache.get(arg);
        if (result == null) {
            result = computable.compute(arg);
            cache.put(arg, result);
        }
        return result;
    }

    public static void main(String[] args) throws Exception {
        YsCache2<String, Integer> expensiveComputer = new YsCache2<>(new ExpensiveFunction());
        System.out.println("开始获取信息");
        Integer result = expensiveComputer.compute("666");
        System.out.println("第一次获取信息=" + result);
        result = expensiveComputer.compute("666");
        System.out.println("第二次获取信息=" + result);
    }
}

问题

性能差
当多个线程同时向计算的时候，需要慢慢等待，严重时，性能甚至比不用缓存更差
当一个缓存不存在，需要进行计算，需要几秒钟的时间，这个时候如果来了另一个请求，即便它的缓存已经存在，也需要等上一个的计算完毕之后才能获取到值，因为锁直接把它挡在了方法外面了

性能待优化——引出锁性能优化经验：缩小锁的粒度

如下代码所示，修改了锁的范围，这样不同的key就可以同时进行读写操作了

public class YsCache3<A, V> implements Computable<A, V> {
    private final Map<A, V> cache = new HashMap<>();
    private Computable<A, V> computable;

    public YsCache3(Computable<A, V> computable) {
        this.computable = computable;
    }

    @Override
    public V compute(A arg) throws Exception {
        System.out.println("进入缓存机制");
        V result = cache.get(arg);
        if (result == null) {
            result = computable.compute(arg);
            synchronized(this){
                cache.put(arg, result);
            }
        }
        return result;
    }

    public static void main(String[] args) throws Exception {
        YsCache3<String, Integer> expensiveComputer = new YsCache3<>(new ExpensiveFunction());
        System.out.println("开始获取信息");
        Integer result = expensiveComputer.compute("666");
        System.out.println("第一次获取信息=" + result);
        result = expensiveComputer.compute("666");
        System.out.println("第二次获取信息=" + result);
    }
}

问题

看起来是解决问题了，但是同时读写，对HashMap也是不安全的

用并发集合——ConcurrentHashMap

如下代码所示，更改使用并发集合之后，就不会在存在同时读写的并发问题了

public class YsCache4<A, V> implements Computable<A, V> {
    private final Map<A, V> cache = new ConcurrentHashMap<>();
    private Computable<A, V> computable;

    public YsCache4(Computable<A, V> computable) {
        this.computable = computable;
    }

    @Override
    public V compute(A arg) throws Exception {
        System.out.println("进入缓存机制");
        V result = cache.get(arg);
        if (result == null) {
            result = computable.compute(arg);
            cache.put(arg, result);
        }
        return result;
    }

    public static void main(String[] args) throws Exception {
        YsCache4<String, Integer> expensiveComputer = new YsCache4<>(new ExpensiveFunction());
        System.out.println("开始获取信息");
        Integer result = expensiveComputer.compute("666");
        System.out.println("第一次获取信息=" + result);
        result = expensiveComputer.compute("666");
        System.out.println("第二次获取信息=" + result);
    }
}

问题

在计算完成之前，另一个需要计算相同值的请求到来，会导致计算两遍，这和缓存想避免多次计算的初衷恰恰相反，是不可接受的

避免重复计算——Future和Callable的妙用

修改后的代码如下所示，使用Future将最终的结果做一层封装，如果两个请求是前后脚进来，如果获取到值，直接调用get方法返回内容即可，如果获取到的值为空，那么，前一个请求就立刻封装一个FutureTask对象放进缓存，并调用它的run方法开始计算，这样第二次请求进来之后，就会拿到这个Future，最终返回的时候会去调用get方法，如果计算还没有完成，就会进行阻塞，计算完成之后就直接返回，不需要两次请求都进行计算

public class YsCache5<A, V> implements Computable<A, V> {
    private final Map<A, Future<V>> cache = new ConcurrentHashMap<>();
    private Computable<A, V> computable;

    public YsCache5(Computable<A, V> computable) {
        this.computable = computable;
    }

    @Override
    public V compute(A arg) throws Exception {
        Future<V> f = cache.get(arg);
        if (f == null) {
            Callable<V> callable = new Callable<V>() {
                @Override
                public V call() throws Exception {
                    return computable.compute(arg);
                }
            };
            FutureTask<V> ft = new FutureTask<>(callable);
            f = ft;
            cache.put(arg, f);
            ft.run();
        }
        return f.get();
    }

    public static void main(String[] args) throws Exception {
        YsCache5<String, Integer> expensiveComputer = new YsCache5<>(new ExpensiveFunction());
        System.out.println("开始获取信息");
        Integer result = expensiveComputer.compute("666");
        System.out.println("第一次获取信息=" + result);
        result = expensiveComputer.compute("666");
        System.out.println("第二次获取信息=" + result);
    }
}

问题

如果两个请求，不是前后脚，而是同时进行，那么当两个请求同时调用map的get方法的时候，拿到的都是null，然后都会进入判断进行计算，依然会存在重复计算的情况

依然存在重复计算的可能——用原子操作putIfAbsent

代码修改为如下所示。当我们保存最后封装好的Future对象的时候，调用ConcurrentHashMap的putIfAbsent方法，这个方法由ConcurrentHashMap保证原子性，如果当前Key已经存在对应的值了，那么就不会再存放，并且返回之前的值，如果当前Key没有值，那么会将键值对存放进去，返回null，这样就可以保证，即便是两个线程同时请求，并且没有拿到对象，虽然会封装两个Future，但是最终只有一个进行计算

public class YsCache6<A, V> implements Computable<A, V> {
    private final Map<A, Future<V>> cache = new ConcurrentHashMap<>();
    private Computable<A, V> computable;

    public YsCache6(Computable<A, V> computable) {
        this.computable = computable;
    }

    @Override
    public V compute(A arg) throws Exception {
        Future<V> f = cache.get(arg);
        if (f == null) {
            Callable<V> callable = new Callable<V>() {
                @Override
                public V call() throws Exception {
                    return computable.compute(arg);
                }
            };
            FutureTask<V> ft = new FutureTask<>(callable);
            f = cache.putIfAbsent(arg, ft);
            if (f == null){
                f = ft;
                ft.run();
            }
        }
        return f.get();
    }

    public static void main(String[] args) throws Exception {
        YsCache6<String, Integer> expensiveComputer = new YsCache6<>(new ExpensiveFunction());
        System.out.println("开始获取信息");
        Integer result = expensiveComputer.compute("666");
        System.out.println("第一次获取信息=" + result);
        result = expensiveComputer.compute("666");
        System.out.println("第二次获取信息=" + result);
    }
}

问题

计算并不一定都是成功的，有可能会因为一些外部原因导致失败，需要重试机制

计算中抛出异常——ExecutionException

抓取异常，并在对应的异常中进行相应的处理，比如直接向上抛出异常，或者打印一些提示，或者做其他处理，如代码所示，在方法中添加while循环语句，在异常之后会进行重试，直到成功
但是会存在内存污染的问题，Future封装的对象一旦异常，那么每次调用get方法都会打印异常，而我们又是先把Future放到map中才进行的计算的，所以需要在异常中将原本的key对应的异常Future删除

public class YsCache7<A, V> implements Computable<A, V> {
    private final Map<A, Future<V>> cache = new ConcurrentHashMap<>();
    private Computable<A, V> computable;

    public YsCache7(Computable<A, V> computable) {
        this.computable = computable;
    }

    @Override
    public V compute(A arg) throws Exception {
        while (true){
            Future<V> f = cache.get(arg);
            if (f == null) {
                Callable<V> callable = new Callable<V>() {
                    @Override
                    public V call() throws Exception {
                        return computable.compute(arg);
                    }
                };
                FutureTask<V> ft = new FutureTask<>(callable);
                f = cache.putIfAbsent(arg, ft);
                if (f == null){
                    f = ft;
                    ft.run();
                }
            }
            try {
                return f.get();
            } catch (CancellationException e){
                cache.remove(arg);
                throw e;
            }catch (InterruptedException e) {
                cache.remove(arg);
                throw e;
            } catch (ExecutionException e) {
                System.out.println("计算异常，请重试");
                cache.remove(arg);
            }
        }
    }

    public static void main(String[] args) throws Exception {
        YsCache7<String, Integer> expensiveComputer = new YsCache7<>(new MayFail());
        System.out.println("开始获取信息");
        Integer result = expensiveComputer.compute("666");
        System.out.println("第一次获取信息=" + result);
        result = expensiveComputer.compute("666");
        System.out.println("第二次获取信息=" + result);
    }
}

问题

缓存并不是数据库，没必要长时间存储，有些数据已经没用了，需要进行删除

缓存过期功能

实现如下，使用线程池，创建一个可以在未来指定时间执行任务的线程池，等超时时间达到之后，就会去清除缓存（方法未做线程安全处理，可以和上面的方法合并使用）

public class YsCache8<A, V> implements Computable<A, V> {
    private final Map<A, Future<V>> cache = new ConcurrentHashMap<>();
    private Computable<A, V> computable;

    public YsCache8(Computable<A, V> computable) {
        this.computable = computable;
    }

    @Override
    public V compute(A arg) throws Exception {
        while (true){
            Future<V> f = cache.get(arg);
            if (f == null) {
                Callable<V> callable = new Callable<V>() {
                    @Override
                    public V call() throws Exception {
                        return computable.compute(arg);
                    }
                };
                FutureTask<V> ft = new FutureTask<>(callable);
                f = cache.putIfAbsent(arg, ft);
                if (f == null){
                    f = ft;
                    ft.run();
                }
            }
            try {
                return f.get();
            } catch (CancellationException e){
                cache.remove(arg);
                throw e;
            }catch (InterruptedException e) {
                cache.remove(arg);
                throw e;
            } catch (ExecutionException e) {
                System.out.println("计算异常，请重试");
                cache.remove(arg);
            }
        }
    }
private static final ScheduledExecutorService EXECUTOR_SERVICE = Executors.newScheduledThreadPool(5);
    public V compute(A arg,long expire) throws Exception {
        if (expire > 0){
            EXECUTOR_SERVICE.schedule(new Runnable() {
                @Override
                public void run() {
                    expire(arg);
                }
            },expire,TimeUnit.MILLISECONDS);
        }
        return computable.compute(arg);
    }

    private synchronized void expire(A key) {
        Future<V> f = cache.get(key);
        if (f != null) {
            if (!f.isDone()){
                System.out.println("Future 任务被取消");
                f.cancel(true);
            }
            System.out.println("过期时间到，缓存被清楚了");
            cache.remove(key);
        }
    }


    public static void main(String[] args) throws Exception {
        YsCache8<String, Integer> expensiveComputer = new YsCache8<>(new ExpensiveFunction());
        System.out.println("开始获取信息");
        Integer result = expensiveComputer.compute("666",6000);
        System.out.println("第一次获取信息=" + result);
        result = expensiveComputer.compute("666");
        System.out.println("第二次获取信息=" + result);
    }
}

问题

指定超时时间，如果同时有大量的请求访问，但又有大量的数据同时缓存过期，会导致雪崩效应

缓存过期时间设置未随机

使用随机数设置缓存过期时间，可以设置一个大概的范围，比如几分钟或者或者几小时，然后生成随机数去指定缓存过期时间，避免缓存会同一时间失效