并发与缓存——读《JCP》

缓存方法在我们编程中经常遇到。例如一个通过很复杂计算的值，但是一旦计算以后，就不再变化，我们可以用缓存存放。最简单的写法如下： 

Object value = null;
if ( (value = cache.get(key)) == null ) {
   value = compteValue(key);
  }
cache.put(key, value);

如果不考虑多线程，这看起来也没什么问题。实际上的应用，则不太可能这样写，因为 
首先：缓存的出现，正是为了减少cpu的消耗，也就是说，compteValue这个方法应该是很耗时的（如果很简单就获取了，也没必要使用缓存）。其次，可能有多个线程并发访问这个缓存，也就是频繁使用，所以这里要说的也就是并发下的缓存。 
《JCP》中给出的例子循序渐进，非常好。我们也可以看到，concurrent包中的一些类的使用。下面我把这些例子放在这里，以供学习。 

public class Memoier1<K, V> implements Computable<K, V> {
    private final Map<K, V> cache = new HashMap<K, V>();

    private final Computable<K, V> c;

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

    @Override
    public synchronized V compute(K key) throws InterruptedException {

        V result = cache.get(key);
        if (result == null) {
            result = c.compute(key);
            cache.put(key, result);
        }
        return result;

    }
}

很眼熟吧。虽然使用了 synchronized ，但是很不辛，这样的同步只会造成更低下的性能。因为synchronized 此时把整个方法锁住了。也就是每个调用线程只能排队调用吧。为了改善性能，我们想到了ConcurrentHashMap 

public class Memoier2<K, V> implements Computable<K, V> {
    private final Map<K, V> cache = new ConcurrentHashMap<K, V>();

    private final Computable<K, V> c;

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

    @Override
    public V compute(K key) throws InterruptedException {

        V result = cache.get(key);
        if (result == null) {
            result = c.compute(key);
            cache.put(key, result);
        }
        return result;

    }
}

这下我们利用了ConcurrentHashMap的特性，compute方法可以并行了。可是，这依然不完美，因为我们说了，compute方法可能很耗时。也就是说，当一个线程在调用compute方法时，另外一个线程也开始调用，但是他不知道前一个线程的状态。所以他也调用了compute方法，这样，缓存就变得无意义了。我们只希望compute一次。问题就出在线程之间没有一个状态可共享。后面的线程如何知道这个key上又一个相同的线程正在执行呢？ 
这是Future就派上用场了。Future可以捕获到线程执行的结果。 

import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;

public class Memoier3<K,V> implements Computable<K,V>{
    private final Map<K,Future<V>> cache = new ConcurrentHashMap<K,Future<V>>();
    private final Computable<K, V> c;

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

    @SuppressWarnings("rawtypes")
    @Override
    public V compute(final K k) throws InterruptedException {
        // TODO Auto-generated method stub
        Future<V> f = cache.get(k);
        if(f == null){
             FutureTask<V> ft = new FutureTask<V>(new Callable<V>(){

                @Override
                public V call() throws Exception {
                    return c.compute(k);
                }

             });
             cache.put(k, ft);
             f = ft;
             ft.run();
        }
        try{
            return f.get();
        }catch(ExecutionException e){
            throw lanuderThrowable(e.getCause());
        }
    }

    public static RuntimeException lanuderThrowable(Throwable t){
        if( t instanceof RuntimeException ){
            return (RuntimeException)t;
        }else if(t instanceof Error){
            throw new Error();
        }else
            throw new IllegalStateException("Not Unchecked", t);

    }

}

这下看上去好多了。但是作者还是找出了一些问题来。首先，如果两个线程都访问compute函数，并且都处于 

if ( f == null ){
   ...
 put(k,v)
}

这样的（check-then-act)的线程不安全中。其次，如果Future失败了，那么每次我们获取到的都是错误的结果（cache pollution)，最后还有cache超时，新旧替换等问题（作者只是提一提，我们可以自己思考如何实现） 
下面是“最终”代码 

package book.jcp.basic;

import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;

public class Memoier<K, V> implements Computable<K, V> {
    private final ConcurrentHashMap<K, Future<V>> cache = new ConcurrentHashMap<K, Future<V>>();
    private final Computable<K, V> c;

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

    @SuppressWarnings("rawtypes")
    @Override
    public V compute(final K k) throws InterruptedException {
        while (true) {
            Future<V> f = cache.get(k);
            if (f == null) {
                FutureTask<V> ft = new FutureTask<V>(new Callable<V>() {

                    @Override
                    public V call() throws Exception {
                        // TODO Auto-generated method stub
                        return c.compute(k);
                    }

                });
                f = cache.putIfAbsent(k, ft);
                if (f == null) {
                    f = ft;
                    ft.run();
                }
            }
            try {
                return f.get();
            } catch (CancellationException e) {
                cache.remove(k);
            } catch (ExecutionException e) {
                throw lanuderThrowable(e.getCause());
            }
        }
    }

    public static RuntimeException lanuderThrowable(Throwable t) {
        if (t instanceof RuntimeException) {
            return (RuntimeException) t;
        } else if (t instanceof Error) {
            throw new Error();
        } else
            throw new IllegalStateException("Not Unchecked", t);

    }

}

我们看到 ，解决第一个问题，使用了 putIfAbsent，使得这个点不会出现两次计算。而对于被取消（失败）的Future，则是采用捕获异常后移除Future来实现。 
最后提到的，缓存过期，就只能大家自己想办法了。 
我认为只能是添加一个租期参数，然后使用一个专门的线程来扫描缓存。