首页 > 代码库 > Java并发(具体实例)—— 构建高效且可伸缩的结果缓存

Java并发(具体实例)—— 构建高效且可伸缩的结果缓存

      这个例子来自《Java并发编程实战》第五章。本文将开发一个高效且可伸缩的缓存,文章首先从最简单的HashMap开始构建,然后分析它的并发缺陷,并一步一步修复。

hashMap版本

      首先我们定义一个Computable接口,该接口包含一个compute()方法,该方法是一个耗时很久的数值计算方法。Memoizer1是第一个版本的缓存,该版本使用hashMap来保存之前计算的结果,compute方法将首先检查需要的结果是否已经在缓存中,如果存在则返回之前计算的值,否则重新计算并把结果缓存在HashMap中,然后再返回。

interface Computable<A, V> {	V compute(A arg) throws InterruptedException;//耗时计算}public class Memoizer1<A, V> implements Computable<A, V> {	private final Map<A, V> cache = new HashMap<A, V>();	private final Computable<A, V> c;	public Memoizer1(Computable<A, V> c) {		this.c = c;	}	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;	}}

  HashMap不是线程安全的,因此要确保两个线程不会同时访问HashMap,Memoizer1采用了一种保守的方法,即对整个方法进行同步。这种方法能确保线程安全性,但会带来一个明显的可伸缩问题:每次只有一个线程可以执行compute。

ConcurrentHashMap版本

      我们可以用ConcurrentHashMap代替HashMap来改进Memoizer1中糟糕的并发行为,由于ConcurrentHashMap是线程安全的,因此在访问底层Map时就不需要进行同步,因此避免了对compute()方法进行同步带来的串行性:

public class Memoizer2 <A, V> implements Computable<A, V> {    private final Map<A, V> cache = new ConcurrentHashMap<A, V>();    private final Computable<A, V> c;    public Memoizer2(Computable<A, V> c) {        this.c = c;    }    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;    }}

  但是这个版本的缓存还是有问题的,如果线程A启动了一个开销很大的计算,而其他线程并不知道这个线程正在进行,那么很可能会重复这个计算。

FutureTask版本1

      我们可以在map中存放Future对象而不是最终计算结果,Future对象相当于一个占位符,它告诉用户,结果正在计算中,如果想得到最终结果,请调用get()方法。Future的get()方法是一个阻塞方法,如果结果正在计算中,那么它会一直阻塞到结果计算完毕,然后返回;如果结果已经计算完毕,那么就直接返回。

public class Memoizer3<A, V> implements Computable<A, V> {	private final Map<A, Future<V>> cache = new ConcurrentHashMap<A, Future<V>>();	private final Computable<A, V> c;	public Memoizer3(Computable<A, V> c) {		this.c = c;	}	public V compute(final A arg) throws InterruptedException {		Future<V> f = cache.get(arg);		if (f == null) {			Callable<V> eval = new Callable<V>() {				public V call() throws InterruptedException {					return c.compute(arg);				}			};			FutureTask<V> ft = new FutureTask<V>(eval);			f = ft;			cache.put(arg, ft);			ft.run(); // call to c.compute happens here		}		try {			return f.get();		} catch (ExecutionException e) {			cache.remove(arg);		}		return null;	}}

  Memoizer3解决了上一个版本的问题,如果有其他线程在计算结果,那么新到的线程会一直等待这个结果被计算出来,但是他又一个缺陷,那就是仍然存在两个线程计算出相同值的漏洞。这是一个典型的"先检查再执行"引起的竞态条件错误,我们先检查map中是否存在结果,如果不存在,那就计算新值,这并不是一个原子操作,所以两个线程仍有可能在同一时间内调用compute来计算相同的值。

FutureTask版本2

      Memoizer3存在这个问题的原因是,复合操作"若没有则添加"不具有原子性,我们可以改用ConcurrentMap中的原子方法putIfAbsent,避免了Memoizer3的漏洞。

public class Memoizer <A, V> implements Computable<A, V> {    private final ConcurrentMap<A, Future<V>> cache            = new ConcurrentHashMap<A, Future<V>>();    private final Computable<A, V> c;    public Memoizer(Computable<A, V> c) {        this.c = c;    }    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>() {                    public V call() throws InterruptedException {                        return c.compute(arg);                    }                };                FutureTask<V> ft = new FutureTask<V>(eval);                f = cache.putIfAbsent(arg, ft);                if (f == null) {                    f = ft;                    ft.run();                }            }            try {                return f.get();            } catch (CancellationException e) {                cache.remove(arg, f);            } catch (ExecutionException e) {                throw LaunderThrowable.launderThrowable(e.getCause());            }        }    }}

  当缓存对象是Future而不是值的时候,将导致缓存污染问题,因为某个计算可能被取消或失败,当出现这种情况时,我们应该把对象从map中移除,然后再重新计算一遍。这个缓存系统的使用十分简单,只需传入一个Computable对象即可构造缓存,要得到计算结果,调用compute()方法即可,该方法会把计算过的结果缓存起来。

总结

      相较于第一个版本,最终版本在性能上有了很大提升,ConcurrentHashMap的使用避免了整个方法加锁;FutureTask的使用使计算异步化,同时通过一个Future对象告诉当前线程计算正在进行中,避免了大量重复计算。

Java并发(具体实例)—— 构建高效且可伸缩的结果缓存