利用Callable,Future以及FutureTask实现一

一,利用Callable,Future以及FutureTask实现一个缓存

利用Callable,Future以及FutureTask实现一个缓存实现一个多线程环境下的缓存
一步一步进行设计分析优化

二,实现

• 1

interface Computable<K, V>{
    V compute(K key) throws InterruptedException;
}

class Memoizer<K, V> {
    private final Map<K,V> cache = new HashMap<K,V>();
    Computable<K, V> computation;
    Memoizer(Computable<K, V> computation){
        this.computation = computation;
    }
    public synchronized V get(K key) throws InterruptedException {
              //从缓存中取结果. 如果结果存在,则直接返回; 若结果不存在,则进行计算
        V result = cache.get(key);
        if(result == null){
            result = computation.compute(key);
            cache.put(key,result);
        }
        return result;
    }
}

上面的代码实现存在一个问题-----伸缩性问题. 由于get方法被synchronized关键字修饰,每次只能有一个线程能够取结果.如果当一个线程结果不存在,需要计算执行的时候,而这个任务执行时间很长,那么其他线程就需要等待很长时间.
我们可以使用线程安全的ConcurrentHashMap进行替换HashMap,且去掉get方法的synchronized关键字

• 2

interface Computable<K, V>{
    V compute(K key) throws InterruptedException;
}

class Memoizer<K, V> {
    private final Map<K,V> cache = new ConcurrentHashMap<K,V>();
    Computable<K, V> computation;
    Memoizer(Computable<K, V> computation){
        this.computation = computation;
    }
    public V get(K key) throws InterruptedException {
        //此处由ConcurrentHashMap来保证线程安全
        V result = cache.get(key);
        //以下代码可以并发执行
        if(result == null){
            result = computation.compute(key);
            cache.put(key,result);
        }
        return result;
    }
}

这种方案解决了伸缩性问题,吞吐量得到了提高.
但是存在一个重复计算的问题. 假设一个任务需要执行很长时间,当第一次通过get取结果的时候, 结果为空,需要执行,且是长时间的执行,那么这时候第二线程也过来取同样key的结果,第一次进行运算的结果(需要长时间运行)还没有出来,那么第二次得到的结果也是空(第二个线程不知道其他线程正在进行计算),那么再次进行了重复的运算.
在这里我们考虑使用FutureTask进行优化,利用他的FutureTask的get方法.

• 3

interface Computable<K, V>{
    V compute(K key) throws InterruptedException;
}

class Memoizer<K, V> {
    private final Map<K,Future<V>> cache = new ConcurrentHashMap<K,Future<V>>();
    Computable<K, V> computation;
    Memoizer(Computable<K, V> computation){
        this.computation = computation;
    }
    public V get(K key) throws InterruptedException {
        //此处由ConcurrentHashMap来保证线程安全
        Future<V> result = cache.get(key);
        
        if(result == null){
            Callable<V> task = new Callable<V>(){
                @Override
                public V call() throws Exception {
                    return computation.compute(key);
                }
            };
            
            FutureTask<V> futureTask = new FutureTask<V>(task);
            //缓存不存在则加入
            result = cache.putIfAbsent(key, futureTask);
            if(result == null){
                result = futureTask;
                futureTask.run();
            }
            
            try {
                //等待运算结果返回
                return result.get();
            } catch (ExecutionException e) {
                e.printStackTrace();
                //遇到任务的终止,或者其他异常则移除缓存
                cache.remove(key,result);
            }
        }
        return null;
        
    }
}

这个缓存实现存在一个问题-----缓存逾期的问题

参考:
<<java编发编程实战>>