快捷導(dǎo)航

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

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

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

概述

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

1.緩存實(shí)現(xiàn)

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

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

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

第一種方式是我們使用HashMap做緩存的容器，因?yàn)镠ashMap不是線程安全的，所以我們需要加上synchronized同步機(jī)制來保證數(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保存之前的計(jì)算結(jié)果，我們每次在計(jì)算結(jié)果時(shí)，先去檢查緩存中是否存在，如果存在則返回緩存中的結(jié)果，否則重新計(jì)算結(jié)果并將其放到緩存中，然后再返回結(jié)果。由于HashMap不是線程安全的，所以我們無法確保兩個線程不會同時(shí)訪問HashMap,所以我們對整個compute方法添加synchronized關(guān)鍵字對方法進(jìn)行同步。這種方法可以保證線程安全型，但是會有一個明顯的問題，那就是每次只有一個線程能夠執(zhí)行compute，如果另一個線程正在計(jì)算結(jié)果，由于計(jì)算是很耗時(shí)的，那么其他調(diào)用compute方法的線程可能會被阻塞很長時(shí)間。如果多個線程在排隊(duì)等待還未計(jì)算出的結(jié)果，那么compute方法的計(jì)算時(shí)間可能比沒有緩存操作的計(jì)算時(shí)間更長，那么緩存就失去了意義。

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

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

改進(jìn)后的代碼如下所示：

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ā)的使用它，但是它在做緩存時(shí)仍然存在一些不足，這個不足就是當(dāng)兩個線程同時(shí)調(diào)用compute方法時(shí)，可能會導(dǎo)致計(jì)算得到相同的值。因?yàn)榫彺娴淖饔镁褪潜苊庀嗤臄?shù)據(jù)被計(jì)算多次。對于更通用的緩存機(jī)制來說，這種情況將更嚴(yán)重。而假設(shè)用于只提供單次初始化的對象來說，這個問題就會帶來安全風(fēng)險(xiǎn)。

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

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

我們將前面用于緩存值的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);
            }
        }
    }
 }

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

注意：我們在代碼中用到了ConcurrentHashMap的putIfAbsent(arg, ft)方法，為啥不能直接用put方法呢？因?yàn)槿绻褂胮ut方法，那么仍然會出現(xiàn)兩個線程計(jì)算出相同的值的問題。我們可以看到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();
                }
            }

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

1.4 測試代碼

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

   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 第一次計(jì)算100的結(jié)果========: " 
                    + Arrays.toString(compute.toArray()));
                    compute = memoizer.compute(100);
                    System.out.println("zxj 第二次計(jì)算100的結(jié)果: " + Arrays.toString(compute.toArray()));
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }).start();
        System.out.println("zxj====>start===>");
    }

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