快捷導(dǎo)航

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

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

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

概述

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

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

為了演示，我們定義一個(gè)計(jì)算接口Computable<A,V>,并在接口中聲明一個(gè)函數(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不是線程安全的，所以我們無法確保兩個(gè)線程不會(huì)同時(shí)訪問HashMap,所以我們對整個(gè)compute方法添加synchronized關(guān)鍵字對方法進(jìn)行同步。這種方法可以保證線程安全型，但是會(huì)有一個(gè)明顯的問題，那就是每次只有一個(gè)線程能夠執(zhí)行compute，如果另一個(gè)線程正在計(jì)算結(jié)果，由于計(jì)算是很耗時(shí)的，那么其他調(diào)用compute方法的線程可能會(huì)被阻塞很長時(shí)間。如果多個(gè)線程在排隊(duì)等待還未計(jì)算出的結(jié)果，那么compute方法的計(jì)算時(shí)間可能比沒有緩存操作的計(jì)算時(shí)間更長，那么緩存就失去了意義。

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

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

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

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

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

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

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

1.4 測試代碼

本來想弄一個(gè)動(dòng)態(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)就行了：如下圖所示

結(jié)論：使用緩存時(shí)，計(jì)算結(jié)果會(huì)很快得到，不使用緩存時(shí)，每次計(jì)算都會(huì)耗時(shí)。

2.并發(fā)技巧總結(jié)

至此：一個(gè)可伸縮性的高效緩存就設(shè)計(jì)完了，至此我們可以總結(jié)下并發(fā)編程的技巧，如下所示：

1.盡量將域聲明為final類型，除非它們是可變的，即設(shè)計(jì)域的時(shí)候要考慮是可變還是不可變的

2.不可變的對象一定是線程安全的，可以任意共享而無需使用加鎖或者保護(hù)性復(fù)制等機(jī)制。

3.使用鎖保護(hù)每個(gè)可變變量

4.當(dāng)保護(hù)同一個(gè)不變性條件中的所有變量時(shí)，要使用同一個(gè)鎖

5.在執(zhí)行復(fù)合操作期間，要持有鎖

6.在設(shè)計(jì)過程中要考慮線程安全。

到此這篇關(guān)于基于Java實(shí)現(xiàn)一個(gè)高效可伸縮的計(jì)算結(jié)果緩存的文章就介紹到這了,更多相關(guān)Java計(jì)算結(jié)果緩存內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: