java并发（二十九）构建高效且可伸缩的结果缓存

概述
    几乎所有应用程序，都会使用某种形式的缓存。重用之前的计算结果，能降低延时，提高吞吐量，但却要消耗更多的内存。用内存“换”CPU。缓存看上去非常简单，然而简单的缓存可能会将性能瓶颈装变为可伸缩性瓶颈，即使缓存是用于提升单线程的性能。笔者会循序渐进的介绍缓存的使用方法演进。

模拟定义接口和功能
声明一个计算函数，使用泛型，输入是A，输出是V。然后我们实现这个接口，再开发一个包装器，可以缓存计算的结果。
接口：

package com.chinaso.phl;

/**
 * @author piaohailin
 * @date 2014-4-23
 */
public interface Computable<A, V> {
    V compute(A arg) throws InterruptedException;
}

实现：

package com.chinaso.phl;

import java.math.BigInteger;
/**
 * @author piaohailin
 * @date 2014-4-23
 */
public class ExpensiveFunction implements Computable<String, BigInteger> {

    @Override
    public BigInteger compute(String arg) throws InterruptedException {
        return new BigInteger(arg);
    }

}

使用HashMap和同步机制来初始化缓存

package com.chinaso.phl;

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

import net.jcip.annotations.GuardedBy;

/**
 * @author piaohailin
 * @date 2014-4-23
 */
public class Memorizer1<A, V> implements Computable<A, V> {
    @GuardedBy("this")
    private final Map<A, V>        cache = new HashMap<A, V>();
    private final Computable<A, V> c;

    public Memorizer1(Computable<A, V> c) {
        this.c = c;
    }

    @Override
    public synchronized V compute(A arg) throws InterruptedException {
        V result = cache.get(arg);
        if (result == null) {
            result = c.compute(arg);
            cache.put(arg, result);
        }
        return result;
    }
}

    这种方法是最基本的缓存用法，是安全的。但是有一个明显的可伸缩性问题：每次只有一个线程能够执行compute。如果另一个线程正在计算结果，那么其他调用coumpute的线程可能被阻塞很长时间。如果有多个线程在排队等待还未计算出的结果，那么compute方法的计算时间可能比没有“记忆”操作的计算时间更长。


用ConcurrentHashMap替换HashMap

package com.chinaso.phl;

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

/**
 * @author piaohailin
 * @date 2014-4-23
 */
public class Memorizer2<A, V> implements Computable<A, V> {

    private final Map<A, V>        cache = new ConcurrentHashMap<A, V>();
    private final Computable<A, V> c;

    public Memorizer2(Computable<A, V> c) {
        this.c = c;
    }

    @Override
    public V compute(A arg) throws InterruptedException {
        V result = cache.get(arg);
        if (result == null) {
            result = c.compute(arg);
            cache.put(arg, result);
        }
        return result;
    }
}

    Memorizer2比Memorizer1有着更好的并发行为，ConcurrentHashMap是线程安全的，所以不需要同步compute方法。但是作为缓存仍然有问题----2个线程同时调用的compute的时候，可能会导致计算得到相同的值。因为缓存是用来避免相同的数据被计算多次，但对于更通用的缓存机制来说，这种情况是更糟糕的，对于提供单词初始化对象缓存来说，这个漏洞会存在安全风险。

    Memorizer2问题在于，如果某个线程启动了一个开销很大的计算，而其他线程并不知道这个计算正在进行，那么很可能会重复这个计算。我们希望通过某种方法来表达“线程X正在计算f(1226)”这种情况，这样当另一个线程查找f(1226)时，他能够知道最高效的方法是等待线程X计算结束，然后去查询缓存“f(1226)的结果是多少”。


基于FutureTask的Memorizer封装器
    FutureTask表示一个计算过程，这个过程可能已经计算完成，也可能正在进行。如果有结果可用，那么FutureTask.get将立即返回结果，否则它会一直阻塞，直到结果计算出来再将其返回。

package com.chinaso.phl;

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

/**
 * @author piaohailin
 * @date 2014-4-23
 */
public class Memorizer3<A, V> implements Computable<A, V> {

    private final Map<A, FutureTask<V>> cache = new ConcurrentHashMap<A, FutureTask<V>>();
    private final Computable<A, V>      c;

    public Memorizer3(Computable<A, V> c) {
        this.c = c;
    }

    @Override
    public V compute(final A arg) throws InterruptedException {
        FutureTask<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<V>(eval);
            f = ft;
            cache.put(arg, ft);
            ft.run(); // 这里调用的是c.compute(arg);
        }
        try {
            return f.get();
        } catch (ExecutionException e) {
            throw new InterruptedException(e.getMessage());
        }
    }
}

    Memorizer3进一步改进了代码。基于ConcurrentHashMap表现出了更好的并发性。但是他仍然有一个漏洞，就是2个线程计算相同值的漏洞。这个漏洞的概率远远小于Memorizer2的情况。但是compute方法中的if代码块仍然是非原子的“先检查再执行”操作。


基于原子操作putIfAbsent的改进

package com.chinaso.phl;

import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

/**
 * @author piaohailin
 * @date 2014-4-23
 */
public class Memorizer4<A, V> implements Computable<A, V> {

    private final ConcurrentMap<A, FutureTask<V>> cache = new ConcurrentHashMap<A, FutureTask<V>>();
    private final Computable<A, V>                c;

    public Memorizer4(Computable<A, V> c) {
        this.c = c;
    }

    @Override
    public V compute(final A arg) throws InterruptedException {
        FutureTask<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<V>(eval);
            // 只有第一个线程添加的时候才会为空，第二个线程此处会获取之前的FutureTask
            f = cache.putIfAbsent(arg, ft);
            if (f == null) {
                f = ft;
                ft.run(); // 这里调用的是c.compute(arg);
            }
        }
        try {
            return f.get();
        } catch (CancellationException e) {
            cache.remove(arg, f);
            return null;
        } catch (ExecutionException e) {
            throw new InterruptedException(e.getMessage());
        }
    }
}

    Memorizer4使用了putIfAbsent的原子方法，从而有效避免了Memorizer3的漏洞。但是这个缓存仍然存在问题。
    缓存污染
    缓存逾期
    缓存清理