基于Java實現(xiàn)一個高效可伸縮的計算結果緩存

更新時間：2023年06月20日 10:22:21 作者：海塔燈

這篇文章將通過對一個計算結果緩存的設計迭代介紹，分析每個版本的并發(fā)缺陷，并分析如何修復這些缺陷，最終完成一個高效可伸縮的計算結果緩存，感興趣的小伙伴可以了解一下

概述

現(xiàn)在的軟件開發(fā)中幾乎所有的應用都會用到某種形式的緩存，重用之前的計算結果能夠降低延遲，提高系統(tǒng)吞吐量，但是需要消耗更多的內(nèi)存，是一種以空間換時間的方法。和許多重復造的輪子一樣，緩存看起來很簡單，無非就是把所有的計算結果保存下來，下次使用的時候優(yōu)先使用緩存中已經(jīng)保存的結果，沒有的情況下才去重新計算。但是不合理的緩存機制設計卻會讓程序的性能受到影響，本文就通過對一個計算結果緩存的設計迭代介紹，分析每個版本的并發(fā)缺陷，并分析如何修復這些缺陷，最終完成一個高效可伸縮的計算結果緩存。

1.緩存實現(xiàn)

為了演示，我們定義一個計算接口Computable<A,V>,并在接口中聲明一個函數(shù)compute(A arg),其輸入的值為A類型的，返回的值為V類型的，接口定義如下所示：

public interface Computable<A,V> {
    V compute(A arg) throws InterruptedException;
}

1.1 使用HashMap+Synchronized實現(xiàn)緩存

第一種方式是我們使用HashMap做緩存的容器，因為HashMap不是線程安全的，所以我們需要加上synchronized同步機制來保證數(shù)據(jù)的存取安全。

代碼如下：

public class HashMapMemoizer<A,V> implements Computable<A,V>{
    private final Map<A,V> cache = new HashMap<>();
    private final Computable<A,V> computable;
    private HashMapMemoizer(Computable<A,V> computable){
        this.computable = computable;
    }
    @Override
    public synchronized V compute(A arg) throws InterruptedException {
        V res = cache.get(arg);
        if (res == null) {
            res = computable.compute(arg);
            cache.put(arg,res);
        }
        return res;
    }
}

如上面的代碼所示，我們使用HashMap保存之前的計算結果，我們每次在計算結果時，先去檢查緩存中是否存在，如果存在則返回緩存中的結果，否則重新計算結果并將其放到緩存中，然后再返回結果。由于HashMap不是線程安全的，所以我們無法確保兩個線程不會同時訪問HashMap,所以我們對整個compute方法添加synchronized關鍵字對方法進行同步。這種方法可以保證線程安全型，但是會有一個明顯的問題，那就是每次只有一個線程能夠執(zhí)行compute，如果另一個線程正在計算結果，由于計算是很耗時的，那么其他調(diào)用compute方法的線程可能會被阻塞很長時間。如果多個線程在排隊等待還未計算出的結果，那么compute方法的計算時間可能比沒有緩存操作的計算時間更長，那么緩存就失去了意義。

1.2 使用ConcurrentHashMap代替HashMap改進緩存的并發(fā)

由于ConcurrentHashMap是線程安全的，因此在訪問底層Map時就不需要進行同步，因此可以避免在對compute方法進行同步時帶來的多個線程排隊等待還未計算出的結果的問題

改進后的代碼如下所示：

public class ConcurrentHashMapMemoizer<A,V> implements Computable<A,V>{
    private final Map<A,V> cache = new ConcurrentHashMap<>();
    private final Computable<A,V> computable;
    private ConcurrentHashMapMemoizer(Computable<A,V> computable){
        this.computable = computable;
    }
    @Override
    public V compute(A arg) throws InterruptedException {
        V res = cache.get(arg);
        if (res == null) {
            res = computable.compute(arg);
            cache.put(arg,res);
        }
        return res;
    }
}

注意：這種方式有著比第一種方式更好的并發(fā)行為，多個線程可以并發(fā)的使用它，但是它在做緩存時仍然存在一些不足，這個不足就是當兩個線程同時調(diào)用compute方法時，可能會導致計算得到相同的值。因為緩存的作用就是避免相同的數(shù)據(jù)被計算多次。對于更通用的緩存機制來說，這種情況將更嚴重。而假設用于只提供單次初始化的對象來說，這個問題就會帶來安全風險。

1.3 完成可伸縮性高效緩存的最終方案

使用ConcurrentHashMap的問題在于如果某個線程啟動了一個開銷很大的計算，而其他線程并不知道這個計算正在進行,那么就很有可能重復這個計算。所以我們希望能通過某種方法來表達“線程X正在進行f(10)這個耗時計算”，這樣當另外一個線程查找f(10)時，它能夠知道目前已經(jīng)有線程在計算它想要的值了，目前最高效的辦法是等線程X計算結束，然后再去查緩存找到f(10)的結果是多少。而FutureTask正好可以實現(xiàn)這個功能。我們可以使用FutureTask表示一個計算過程，這個過程可能已經(jīng)計算完成，也可能正在進行。如果有結果可以用，那么FutureTask.get()方法將會立即返回結果，否則它會一直阻塞，知道結果計算出來再將其返回

我們將前面用于緩存值的Map重新定義為ConcurrentHashMap<A, Future<V>>,替換原來的ConcurrentHashMap<A, V>,代碼如下所示：

public class PerfectMemoizer<A, V> implements Computable<A, V> {
    private final ConcurrentHashMap<A, Future<V>> cache
            = new ConcurrentHashMap<>();
    private final Computable<A, V> computable;
    public PerfectMemoizer(Computable<A, V> computable) {
        this.computable = computable;
    }
    @Override
    public V compute(final A arg) throws InterruptedException {
        while (true) {
            Future<V> f = cache.get(arg);
            if (f == null) {
                Callable<V> eval = new Callable<V>() {
                    @Override
                    public V call() throws Exception {
                        return computable.compute(arg);
                    }
                };
                FutureTask<V> ft = new FutureTask<>(eval);
                f = cache.putIfAbsent(arg, ft);
                if (f == null) {
                    f = ft;
                    ft.run();
                }
            }
            try {
                return f.get();
            } catch (CancellationException e) {
                cache.remove(arg);
            } catch (ExecutionException e) {
                throw new RuntimeException(e);
            }
        }
    }
 }

如上面代碼所示，我們首先檢測某個相應的計算是否已經(jīng)開始，如果還沒開始，就創(chuàng)建一個FutureTask并注冊到Map中，然后啟動計算，如果已經(jīng)開始計算，那么就等待計算的結果。結果可能很快得到，也可能還在運算過程中。但是對于Future.get()方法來說是透明的。

注意：我們在代碼中用到了ConcurrentHashMap的putIfAbsent(arg, ft)方法，為啥不能直接用put方法呢？因為如果使用put方法，那么仍然會出現(xiàn)兩個線程計算出相同的值的問題。我們可以看到compute方法中的if代碼塊是非原子的，如下所示：

// compute方法中的if部分代碼
   if (f == null) {
                Callable<V> eval = new Callable<V>() {
                    @Override
                    public V call() throws Exception {
                        return computable.compute(arg);
                    }
                };
                FutureTask<V> ft = new FutureTask<>(eval);
                f = cache.putIfAbsent(arg, ft);
                if (f == null) {
                    f = ft;
                    ft.run();
                }
            }

因此兩個線程仍有可能在同一時間調(diào)用compute方法來計算相同的值，只是概率比較低。即兩個線程都沒有在緩存中找到期望的值，因此都開始計算。而引起這個問題的原因復合操作（若沒有則添加）是在底層的Map對象上執(zhí)行的，而這個對象無法通過加鎖來確保原子性，所以需要使用ConcurrentHashMap中的原子方法putIfAbsent，避免這個問題

1.4 測試代碼

本來想弄一個動態(tài)圖展示使用緩存和不使用緩存的速度對比的，但是弄出來的圖太大，傳不上來，所以給測試代碼讀者自己驗證下：

   public static void main(String[] args) throws InterruptedException {
        Computable<Integer, List<String>> cache = arg -> {
            List<String> res = new ArrayList<>();
            for (int i = 0; i < arg; i++) {
                Thread.sleep(50);
                res.add("zhongjx==>" + i);
            }
            return res;
        };
        PerfectMemoizer<Integer, List<String>> memoizer = new PerfectMemoizer<>(cache);
        new Thread(new Runnable() {
            @Override
            public void run() {
                List<String> compute = null;
                try {
                    compute = memoizer.compute(100);
                    System.out.println("zxj 第一次計算100的結果========: " 
                    + Arrays.toString(compute.toArray()));
                    compute = memoizer.compute(100);
                    System.out.println("zxj 第二次計算100的結果: " + Arrays.toString(compute.toArray()));
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }).start();
        System.out.println("zxj====>start===>");
    }

測試代碼中我們使用Thread.sleep()方法模擬耗時操作。我們要測試不使用緩存的情況就是把 f = cache.putIfAbsent(arg, ft);這句代碼注釋調(diào)就行了：如下圖所示