java實現數據緩存

摘抄自java並發實戰

有時候需要對數據緩存。用Map緩存數據比較合適。但是由於對吞吐量，一致性，計算性能的要求，對數據進行緩存的設計還是需要慎重考慮的。

一、利用HashMap加同步

（1）說明

把HashMap當作緩存容器。每緩存一個key的時候，都進行同步。

（2）代碼

package memory;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by adrian.wu on 2018/12/12.
 */
public class MemoryFirst<K, V> implements Computable<K, V> {
    private final Map<K, V> cache = new HashMap<>();
    private final Computable<K, V> c;

    public MemoryFirst(Computable<K, V> c) {
        this.c = c;
    }

    @Override
    public synchronized V compute(K arg) throws InterruptedException {
        V result = cache.get(arg);

        if (result == null) {
            result = c.compute(arg);
            cache.put(arg, result);
        }

        return result;
    }
}

（3）缺點

由於HashMap並非線程安全，因此每一次計算都使用同步機制確保線程安全。很明顯，這種方式伸縮性比較差。因為一個線程正在計算結果，其它所有線程都在等待，即使對應的arg是不同的。

 

二、用ConcurrentHashMap代替HashMap

（1）說明

ConcurrentHashMap是線程安全的，並且同步並非對整個Map進行同步而是對每一個分段進行同步，所以並發性也可以大大提升。

（2）代碼

package memory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created by adrian.wu on 2018/12/12.
 */
public class MemorySecond<K, V> implements Computable<K, V> {
    private final Map<K, V> cache = new ConcurrentHashMap<>();
    private final Computable<K, V> c;

    public MemorySecond(Computable<K, V> c) {
        this.c = c;
    }

    @Override
    public V compute(K arg) throws InterruptedException {
        V result = cache.get(arg);

        if (result == null) {
            result = c.compute(arg);
            cache.put(arg, result);
        }

        return result;
    }
}

（3）缺點

相比第一個設計方案。這種方案已經有很大的提升了。但是如果一個compute的計算開銷很大，恰巧有另一個同一個arg的線程同時請求compute，則會造成重復計算，重復put的情況。所以我們希望如果有一個線程正在計算的時候另一個線程正在等待而不是重復計算。

 

三、利用FutureTask解決第二個設計的問題

（1）說明

利用FutrueTask, 如果get到結果則返回，如果正在計算則利用FutureTask的特性阻塞。否則計算。

（2）代碼

package memory;

import org.slf4j.Logger;

import java.util.Map;
import java.util.concurrent.*;

import static memory.ErrorHandler.launderThrowable;

/**
 * Created by adrian.wu on 2018/12/12.
 */
public class MemoryThird<K, V> implements Computable<K, V> {
    private final Map<K, Future<V>> cache = new ConcurrentHashMap<>();
    private final Computable<K, V> c;

    public MemoryThird(Computable<K, V> c) {
        this.c = c;
    }

    @Override
    public V compute(final K arg) throws InterruptedException {
        Future<V> f = cache.get(arg);
        if (f == null) {
            Callable<V> eval = new Callable<V>() {
                @Override
                public V call() throws Exception {
                    return c.compute(arg);
                }
            };

            FutureTask<V> ft = new FutureTask<>(eval);
            f = ft;
            cache.put(arg, ft);
            ft.run(); // start compute
        }
        try {
            return f.get();
        } catch (ExecutionException e) {
            throw launderThrowable(e.getCause());
        }
    }
}

（3）缺點

只有一個缺陷，仍然存在兩個線程計算出相同值的漏洞。就是由於compute方法中的if代碼塊仍然是非原子的“先檢查，再執行”，因此仍然有可能兩個線程在同一時間計算一個不存在的arg。原因是第23行的get方法和34行的put方法是對底層的Map操作，所以無法保證原子性。由於cache里面的是future而不是真正的值，所以將有可能導致緩存污染（cache pollution）問題，即如果某個計算過程被取消或者失敗，那么緩存存入的Future是有缺陷的。

 

四、最終設計方案

（1）說明

使用putIfAbsent代替put，以保證原子性。如果發現Future計算被取消或失敗則刪除，從而緩存不會消耗過多內存。

（2）代碼

package memory;

import java.util.Map;
import java.util.concurrent.*;

import static memory.ErrorHandler.launderThrowable;

/**
 * Created by adrian.wu on 2018/12/12.
 */
public class Memory<K, V> implements Computable<K, V> {
    private Map<K, Future<V>> cache = new ConcurrentHashMap<>();
    private Computable<K, V> c;

    public Memory(Computable<K, V> c) {
        this.c = c;
    }

    @Override
    public V compute(K 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 c.compute(arg);
                    }
                };

                FutureTask<V> ft = new FutureTask<>(eval);
                
                f = cache.putIfAbsent(arg, ft); //double check
                if (f == null) {
                    f = ft;
                    ft.run(); //start compute
                }
            }
            try {
                return f.get();
            } catch (CancellationException e) {
                cache.remove(arg);
            } catch (ExecutionException e) {
                throw launderThrowable(e.getCause());
            }
        }
    }
}