如何使用Future和ConcurrentHashMap实现一个线程安全，高效，无缓存污染的缓存

小灰灰 2021-09-16 03:04 325阅读 0赞

几乎所有的服务器应用程序都会使用某种形式的缓存。重用之前的结果，以及来降低延迟，提高吞吐量。   
像许多重复发明的轮子一样，缓存看上去非常简单。然而，简单的缓存可能会大大提升系统的瓶颈。我们现在首先来实现一个简单的缓存，通过一步步深入来构建一个并发安全且高效的缓存。   
构建场景：我们的系统中有一个比较耗时的查询操作，定义为searchUser，如果不使用缓存那么我们会每次都去数据库进行一个耗时的查询。不要问为什么查询一个用户会耗时，因为这是我定义的耗时的操作。所以我们现在需要构建一个缓存，用来存储已经有的查询结果。于是我定义一个查询接口：

package com.wsy.cache;
    
    import java.util.concurrent.ExecutionException;
    
    /**
     * 查询用户的接口
     * @author shuyweng
     *
     */
    public interface SeachUser<T, V> {
        public V searchUser(T t) throws InterruptedException, ExecutionException;
    }

1 一个简单的缓存   
分析：很显然针对此问题，HashMap是一个非常不错的选择。

package com.wsy.cache;
    
    import java.util.HashMap;
    import java.util.Map;
    
    public class SimpleCache<V, T> implements SeachUser<V, T>{
        // 用于缓存已查询结果的hashMap
        private final Map<V, T> cache = new HashMap<V, T>();
        @Override
        public T searchUser(V v) {
            T result = cache.get(v);
            if(result == null){
                result = searchUserByDatabase(v);
                cache.put(v, result);
                return result;
            }
            return result;
        }
    
        // 自定义一个耗时的查询操作
        private T searchUserByDatabase(V v) {
            try {
                // 让线程睡眠3秒，模仿一个耗时的查询
                Thread.currentThread().sleep(3000);
                return (T)new Object();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            return null;
        }
    
    }

在这个demo中，如果cache中不存在该结果，就去执行耗时的操作查询。如果存在就使用cache中的值。那么这个缓存存在什么问题呢？   
问题是HashMap并不是线程安全的，如果有多个线程同时访问就会出现很快的大家都会想到加锁，可是加锁会导致这个查询接口阻塞，所以并不是一个比较好的方式。另一种方式是使用concurrentHashMap替换HashMap。   
2 并发安全的缓存

package com.wsy.cache;
    
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    
    public class SafeCache<V, T> implements SeachUser<V, T>{
        // 用于缓存已查询结果的ConcurrentHashMap
        private final Map<V, T> cache = new ConcurrentHashMap<V, T>();
        @Override
        public T searchUser(V v) {
            T result = cache.get(v);
            if(result == null){
                result = searchUserByDatabase(v);
                cache.put(v, result);
                return result;
            }
            return result;
        }
    
        // 自定义一个耗时的查询操作
        private T searchUserByDatabase(V v) {
            try {
                // 让线程睡眠3秒，模仿一个耗时的查询
                Thread.currentThread().sleep(3000);
                return (T)new Object();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            return null;
        }
    
    }

这个方案还存在什么问题呢？   
仔细想一下，这种方案会存在重复计算，比如A,B,C3个用户按先后查询某个结果，在A还没有查询得到结果的时候，B,C会去执行这个耗时查询。而这显然是重复的，最优的方案显然是让B,C进行等待，当A查询出结果之后再从缓存中取数据，从而避免重复查询。这就是所谓的高效，我们可以想到java提供了一种阻塞等待的类：FutureTask。   
3 一个线程安全且高效的缓存

package com.wsy.cache;
    
    import java.util.Map;
    import java.util.concurrent.Callable;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Future;
    
    public class SafeEffectiveCache<V, T> implements SeachUser<V, T>{
        // 用于缓存已查询结果的ConcurrentHashMap
        private final Map<V, Future<T>> cache = new ConcurrentHashMap<V, Future<T>>();
        // 用于管理查询线程的线程池
        private ExecutorService pool = Executors.newFixedThreadPool(20);
        @Override
        public T searchUser(final V v) throws InterruptedException, ExecutionException {
            Future<T> f = cache.get(v);
            if(f == null){
                f = pool.submit(new Callable<T>() {
    
                    @Override
                    public T call() throws Exception {
                        return searchUserByDatabase(v);
                    }
                });
                cache.put(v, f);
                return f.get();
            }
            return f.get();
        }
    
        // 自定义一个耗时的查询操作
        private T searchUserByDatabase(V v) {
            try {
                // 让线程睡眠3秒，模仿一个耗时的查询
                Thread.currentThread().sleep(3000);
                return (T)new Object();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            return null;// 可定义成空对象
        }
    
    }

这是一个并发安全且高效的缓存demo，我们不妨想一想这个还存在什么问题?   
存在缓存污染，当缓存的是Future而不是值时，将导致缓存污染：如果某个查询失败了。会导致Future存在，但是Future的get方法并不能获得我们期望的值。所以代码还需要进行修改：   
4 一个线程安全，高效，无缓存污染的缓存

package com.wsy.cache;
    
    import java.util.Map;
    import java.util.concurrent.Callable;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Future;
    
    public class SafeEffectiveNiceCache<V, T> implements SeachUser<V, T>{
        // 用于缓存已查询结果的ConcurrentHashMap
        private final Map<V, Future<T>> cache = new ConcurrentHashMap<V, Future<T>>();
        // 用于管理查询线程的线程池
        private ExecutorService pool = Executors.newFixedThreadPool(20);
        @Override
        public T searchUser(final V v) throws InterruptedException, ExecutionException {
            while(true){
            Future<T> f = cache.get(v);
            if(f == null){
                f = pool.submit(new Callable<T>() {
    
                    @Override
                    public T call() throws Exception {
                        return searchUserByDatabase(v);
                    }
                });
                cache.put(v, f);
                try{
                return f.get();
                }catch(Exception e){
                    cache.remove(v);
                    e.printStackTrace();
                }
            }
            return f.get();
            }
        }
    
        // 自定义一个耗时的查询操作
        private T searchUserByDatabase(V v) {
            try {
                // 让线程睡眠3秒，模仿一个耗时的查询
                Thread.currentThread().sleep(3000);
                return (T)new Object();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            return null;
        }
    }

当出现异常的时候就清楚缓存中相应的数据，从而避免缓存污染。而此处为什么要加一个while循环呢，明明不加while循环也能防止污染啊？仔细想想就会发现，如果有了while循环会解决另外一个并发问题？在代码中的if(f == null)并不是线程安全的，假如2个用户几乎在同一时间执行查询方法，那么2个用户都会去执行耗时操作，而谁执行的更加快我们也不知道。所以此处加了一个while循环，当有一个用户执行查询成功的时候另一个用户也可能使用到缓存中的值。此处依然存在问题：如果B在Future阻塞的时候，A执行完了，过了一会B也执行完了，而此刻A就会重复对缓存赋值，这显然是多余的操作。所以我们还可以将cache.put改成cache.putIfAbsent。这表示只有在key不存在的时候才进行赋值。

[https://blog.csdn.net/wengsy\_5041/article/details/78105315][https_blog.csdn.net_wengsy_5041_article_details_78105315]

[https_blog.csdn.net_wengsy_5041_article_details_78105315]: https://blog.csdn.net/wengsy_5041/article/details/78105315