5.6 构建高效且可伸缩的结果缓存

骑猪看日落 2022-05-23 04:55 167阅读 0赞

首先，声明一个计算接口：

package JavaDay6_02.Demo5;
    
    /**
     * @author myvina@qq.com
     * @date 18-6-2  下午12:15
     */
    
    public interface Computable<K, V> {
        V compute(K arg) throws InterruptedException;
    }

一种实现类如下：

package JavaDay6_02.Demo5;
    
    import java.math.BigInteger;
    
    /**
     * @author myvina@qq.com
     * @date 18-6-2  下午12:23
     */
    
    public class ExpensiveFunction implements Computable<String, BigInteger> {
        @Override
        public BigInteger compute(String arg) throws InterruptedException {
            Thread.sleep(5000);
            System.out.println("ExpensiveFunction计算完成，arg = " + arg);
            return new BigInteger(arg);
        }
    }

第一次缓存的尝试：

package JavaDay6_02.Demo5;
    
    import net.jcip.annotations.GuardedBy;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * @author myvina@qq.com
     * @date 18-6-2  下午12:18
     */
    
    public class Memorizer<K, V> implements Computable<K, V> {
        @GuardedBy("this")
        private final Map<K, V> cache = new HashMap<>();
        private final Computable<K, V> computable;
    
        public Memorizer(Computable<K, V> computable) {
            this.computable = computable;
        }
    
        @Override
        public synchronized V compute(K arg) throws InterruptedException {
            V result = cache.get(arg);
            if(result == null) {
                result = computable.compute(arg);
                cache.put(arg, result);
            }
            return result;
        }
    }

这里用synchronized来确保线程安全，但当有多个线程在排队等待还未计算出的结果，则更浪费时间，与预想不符。

下面为第二次尝试：

package JavaDay6_02.Demo5;
    
    import net.jcip.annotations.GuardedBy;
    
    import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    
    /**
     * @author myvina@qq.com
     * @date 18-6-2  下午12:18
     */
    
    public class Memorizer<K, V> implements Computable<K, V> {
        @GuardedBy("this")
        private final Map<K, V> cache = new ConcurrentHashMap<>();
        private final Computable<K, V> computable;
    
        public Memorizer(Computable<K, V> computable) {
            this.computable = computable;
        }
    
        @Override
        public V compute(K arg) throws InterruptedException {
            V result = cache.get(arg);
            if(result == null) {
                result = computable.compute(arg);
                cache.put(arg, result);
            }
            return result;
        }
    }

这样使用线程安全的ConcurrentHashMap来代替HashMap确保线程安全，却有可能出现相同的运算结果被多个线程计算多次，这也与预想不符。

下面为第三次尝试：

package JavaDay6_02.Demo5;
    
    import net.jcip.annotations.GuardedBy;
    
    import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.*;
    
    /**
     * @author myvina@qq.com
     * @date 18-6-2  下午12:18
     */
    
    public class Memorizer<K, V> implements Computable<K, V> {
        @GuardedBy("this")
        private final Map<K, Future<V>> cache = new ConcurrentHashMap<>();
        private final Computable<K, V> computable;
    
        public Memorizer(Computable<K, V> computable) {
            this.computable = computable;
        }
    
        @Override
        public V compute(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 computable.compute(arg);
                    }
                };
                FutureTask<V> futureTask = new FutureTask<>(eval);
                f = futureTask;
                cache.put(arg, futureTask);
            }
            try {
                return f.get();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
    }

这个实现方法几乎是完美的，但由于compute方法中的if代码块仍然是非原子性的“先检查再执行”操作，因此两个线程仍有可能在同一时间内调用compute来计算相同的值。

下面的例子使用了ConcurrentMap中的原子方法putIfAbsent，避免了上次实现中的漏洞。代码如下：

package JavaDay6_02.Demo5;
    
    import net.jcip.annotations.GuardedBy;
    
    import java.util.concurrent.*;
    
    /**
     * @author myvina@qq.com
     * @date 18-6-2  下午12:18
     */
    
    public class Memorizer<K, V> implements Computable<K, V> {
        @GuardedBy("this")
        private final ConcurrentMap<K, Future<V>> cache = new ConcurrentHashMap<>();
        private final Computable<K, V> computable;
    
        public Memorizer(Computable<K, V> computable) {
            this.computable = computable;
        }
    
        @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 computable.compute(arg);
                        }
                    };
                    FutureTask<V> futureTask = new FutureTask<>(eval);
                    f = cache.putIfAbsent(arg, futureTask);
                    if (f == null) {
                        f = futureTask;
                        futureTask.run();
                    }
                }
                try {
                    return f.get();
                } catch (CancellationException e) {
                    cache.remove(arg, f);
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            }
        }
    }

这几乎完美实现了缓存的功能，以下为一个测试类：

package JavaDay6_02.Demo5;
    
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    
    /**
     * @author myvina@qq.com
     * @date 18-6-2  下午12:44
     */
    
    public class TestMemorizer {
        @org.junit.jupiter.api.Test
        void testMemorizer() throws BrokenBarrierException, InterruptedException {
            CyclicBarrier cyclicBarrier = new CyclicBarrier(9);
            ExpensiveFunction ef = new ExpensiveFunction();
            Memorizer memorizer = new Memorizer(ef);
            for(int i = 0; i < 4; i++) {
                new Thread(new Test(memorizer, cyclicBarrier, i)).start();
            }
            for(int i = 0; i < 4; i++) {
                new Thread(new Test(memorizer, cyclicBarrier, i)).start();
            }
            System.out.println("（主线程）等待所有线程缓存完毕...");
            cyclicBarrier.await();
            System.out.println("（主线程）所有线程缓存完毕...");
            System.out.println("（主线程）检查是否缓存完成...");
            for(int i = 0; i < 4; i++) {
                System.out.println(memorizer.compute("" + i));
            }
        }
    
        private class Test implements Runnable {
            private final Memorizer memorizer;
            private final CyclicBarrier cyclicBarrier;
            private int arg;
    
            private Test(Memorizer memorizer, CyclicBarrier cyclicBarrier, int arg) {
                this.memorizer = memorizer;
                this.cyclicBarrier = cyclicBarrier;
                this.arg = arg;
            }
    
            @Override
            public void run() {
                try {
                    System.out.println("(子线程）开始计算compute(" + arg + ")...");
                    memorizer.compute("" + arg);
                    System.out.println("(子线程）计算完毕...");
                    cyclicBarrier.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }
        }
    }

测试结果如下：

![70][]

由测试结果可以看出，即使有多个线程同时获取compute的值，真正的计算方法也不会运行两次，同时也是线程安全的。

[70]: /images/20220523/5897c6af633f429794d3cb46246316e2.png