构建高效且可伸缩的结果缓存（中）-阿里云开发者社区

构建高效且可伸缩的结果缓存（中）

2022-04-29 166

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 构建高效且可伸缩的结果缓存（中）

继续演进

FutureTask 是异步编程里面的一个非常重要的组成部分。

反正基于 Future 这个东西，可以玩出花儿来。

比如我们的这个场景中，如果要用到 FutureTask，那么我们的 Map 就需要修改为这样：

Map<String, Future> SCORE_CACHE = new ConcurrentHashMap<>();

通过维护姓名和 Future 的关系来达到我们的目的。

Future 本身就代表一个任务，对于缓存维护这个需求来说，这个任务到底是在执行中还是已经执行完成了它并不关心，这个“它”指的是 SCORE_CACHE 这个 Map。

对于 Map 来说，只要有个任务放进来就行了。

而任务到底执行完成没有，应该是从 Map 里面 get 到对应 Future 的这个线程关心的。

它怎么关心？

通过调用 Future.get() 方法。

整个代码写出来就是这样的：

public class ScoreQueryService {
    private final Map<String, Future<Integer>> SCORE_CACHE = new ConcurrentHashMap<>();
    public Integer query(String userName) throws Exception {
        Future<Integer> future = SCORE_CACHE.get(userName);
        if (future == null) {
            Callable<Integer> callable = () -> loadFormDB(userName);
            FutureTask futureTask = new FutureTask<>(callable);
            future = futureTask;
            SCORE_CACHE.put(userName, futureTask);
            futureTask.run();
        }
        return future.get();
    }
    private Integer loadFormDB(String userName) throws InterruptedException {
        System.out.println("开始查询userName=" + userName + "的分数");
        //模拟耗时
        TimeUnit.SECONDS.sleep(1);
        return ThreadLocalRandom.current().nextInt(380, 420);
    }
}

怕你不熟悉 futureTask ，所以简单解释一下关于 futureTask 的四行代码，但是我还是强烈建议你把这个东西掌握了，毕竟说它是异步编程的基石之一也不为过。

基石还是得拿捏明白，否则就容易被面试官拿捏。

Callable<Integer> callable = () -> loadFormDB(userName);
FutureTask futureTask = new FutureTask<>(callable);
futureTask.run();
return future.get();

首先我构建了一个 Callable 作为 FutureTask 构造函数的入参。

构造函数上面的描述翻译过来就是：创建一个 FutureTask，运行时将执行给定的 Callable。

“运行时”指的就是 futureTask.run() 这一行代码，而“给定的 Callable ”就是 loadFormDB 任务。

也就是说调用了 futureTask.run() 之后，才有可能会执行到 loadFormDB 方法。

然后调用 future.get() 就是获取 Callable 的结果，即获取 loadFormDB 方法的结果。如果该方法还没有运行结束，就死等。

对于这个方案，书上是这样说的：

主要关注我划线的部分，我一句句的说

它只有一个缺陷，即仍然存在两个线程计算出相同值的漏洞。

这句话其实很好理解，因为代码里面始终有一个“①获取-②判断-③放置”的动作。

这个动作就不是原子性的，所以有一定的几率两个线程都冲进来，然后发现缓存中没有，就都走到 if 分支里面去了。

但是标号为 ① 和 ② 的地方，从需求实现的角度来说，肯定是必不可少的。

能想办法的地方也就只有标号为 ③ 的地方了。

到底啥办法呢？

不着急，下一小节说，我先把后半句话给解释了：

这个漏洞的发生概率要远小于 Memoizer2 中发生的概率。

Memoizer2 就是指前面用 ConcurrentHashMap 替换 HashMap 后的方案。

那么为什么引入 Future 之后的这个方案，触发刚刚说到的 bug 的概率比之前的方案小呢？

答案就藏在这两行代码里面：

之前是要把业务逻辑执行完成，拿到返回值之后才能维护到缓存里面。

现在是先维护缓存，然后再执行业务逻辑，节约了执行业务逻辑的时间。

而一般来说最耗时的地方就是业务逻辑的执行，所以这个“远小于”就是这样来的。

那怎么办呢？

接着演进呀。

最终版

书里面，针对上面那个“若没有则添加”这个非原子性的动作的时候，提到了 map 的一个方法：

首先从标号为 ① 的地方我们可以知道，这个方法传进来的 key 如果还没有与一个值相关联（或被映射为null），则将其与给定的值进行映射并返回 null ，否则返回当前值。

如果我们只关心返回值的话，那就是：如果有就返回对应的值，如果没有就返回 null。

标号为 ② 的地方说的是啥呢？

它说默认的实现没有对这个方法的同步性或原子性做出保证。如果你要提供原子性保证，那么就请覆盖此方法，自己去写。

所以，我们接着就要关注一下 ConcurrentHashMap 的这个方法是怎么搞得了：

还是通过 synchronized 方法来保证了原子性，当操作的是同一个 key 的时候保证只有一个线程去执行 put 的操作。

所以书中给出的最终实现，是这样的：

public class ScoreQueryService {
    public static final Map<String, Future<Integer>> SCORE_CACHE = new ConcurrentHashMap<>();
    public Integer query(String userName) throws Exception {
        while (true) {
            Future<Integer> future = SCORE_CACHE.get(userName);
            if (future == null) {
                Callable<Integer> callable = () -> loadFormDB(userName);
                FutureTask futureTask = new FutureTask<>(callable);
                future = SCORE_CACHE.putIfAbsent(userName, futureTask);
                //如果为空说明之前这个 key 在 map 里面不存在
                if (future == null) {
                    future = futureTask;
                    futureTask.run();
                }
            }
            try {
                return future.get();
            } catch (CancellationException e) {
                System.out.println("查询userName=" + userName + "的任务被移除");
                SCORE_CACHE.remove(userName, future);
            } catch (Exception e) {
                throw e;
            }
        }
    }
    private Integer loadFormDB(String userName) throws InterruptedException {
        System.out.println("开始查询userName=" + userName + "的分数");
        //模拟耗时
        TimeUnit.SECONDS.sleep(5);
        return ThreadLocalRandom.current().nextInt(380, 420);
    }
}