缓存实战小项目

2024-01-29 19:48:58

高性能缓存项目

目标：从0开始迭代，手把手一步步设计并实现

概述

缓存的用处：缓存在实际生产中是非常重要的工具，有了缓存之后，我们可以避免重复计算，提高吞吐量

虽然缓存乍一看很简单，不就是一个Map吗？最初级的缓存确实可以用一个Map来实现，不过一个功能晚辈、性能强劲的缓存，需要考虑的店就非常多了，我们从最简单的HashMap入手，一步步提高我们缓存的性能。

最初始版本

从最简单的案例开始

package imooccache;

import java.util.HashMap;
import java.util.concurrent.TimeUnit;

/**
 * 最简单的形式：HashMap
 */
public class IMoocCache {

		//v1.0
  //  private static  HashMap<String,Integer> cache = new HashMap<>();
		//v2.0  加入final后，该变量只能被被赋值一次
     private static final  HashMap<String,Integer> cache = new HashMap<>();

 //   public static Integer computer(String userId) throws InterruptedException {
 public synchronized Integer computer(String userId) throws InterruptedException {
        Integer result = cache.get(userId);
        //先检查HshMap里面有没有保存过之前的计算结果
        if(result==null){

                result = doCompute(userId);


            cache.put(userId,result);
        }

        return result;

    }


    private static Integer doCompute(String userId) throws InterruptedException {
        TimeUnit.SECONDS.sleep(5);

        return new Integer(userId);
    }


    public static void main(String[] args) throws InterruptedException {

        IMoocCache iMoocCache = new IMoocCache();
        System.out.println("开始计算了");
        Integer result = computer("13");

        System.out.println("第一次计算结果"+result);
      result =  IMoocCache.computer("13");

        System.out.println("第二次计算结果"+result);
    }
}

程序第二次计算结果将会远远快于第一次，因为程序将第一次计算结果存在了HashMap中，计算时先查询HashMap

但是V1.0版本的代码存在问题：程序中使用了hashmap,这是线程不安全的，这时候我们很容易想到利用synchronize关键字来修饰hashmap,以保证其线程安全性。但是使用synchronize关键字的V2.0同样会带来问题，如上所示，整个synchronize锁的范围过大，业务代码和非业务代码全部锁住了。这会导致程序运行效率低下，代码的复用性差，这时候考虑给hashmap加final关键字，并且使用装饰者模式把代码侵入的问题解决。

解决代码复用性差问题：

使用装饰者模式重构代码

这里我们假设ExpensiveFunction类是耗时计算的实现类，实现了Computable接口，但是其本身不具备缓存功能，也不需要考虑缓存的事情。

为了将业务代码compute方法解耦，写一个接口

package imooccache.computable;

/**
 * 有一个计算函数computer,用来代表耗时计算，每个计算器都要事先这个接口，这样就可以无情如实现缓存功能
 */

public interface Computable<A,V> {
    V compute(A arg) throws Exception; 

}

其实现类为：

package imooccache.computable;

public class ExpensiveFunction implements Computable<String, Integer> {


    @Override
    public Integer compute(String arg) throws Exception {
        Thread.sleep(5000);
        return Integer.valueOf(arg);
    }
}

具体代码如下：

package imooccache;

import imooccache.computable.Computable;
import imooccache.computable.ExpensiveFunction;

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

public class ImoocCache2<A,V> implements Computable<A,V> {


    private final Map<A,V> cache = new HashMap<>();

    private final Computable<A,V> c;

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

    @Override
    public synchronized V compute(A arg) throws Exception {
        System.out.println("进入缓存机制");
        //一旦进入缓存机制，那么需要查找值是否已经被计算过
        V result = cache.get(arg);

        if(result==null){
            //如果result没有被计算
            result = c.compute(arg);

            //计算了结果后就需要将它放入缓存中

            cache.put(arg,result);
        }

        return result;
    }


    public static void main(String[] args) {


        ImoocCache2<String, Integer> expensiveComputer = new ImoocCache2<>(
                new ExpensiveFunction());
        try {
            Integer result = expensiveComputer.compute("666");
            System.out.println("第一次计算结果"+result);
            result = expensiveComputer.compute("666");
            System.out.println("第二次计算结果"+result);
        } catch (Exception e) {
            e.printStackTrace();
        }





    }
}

##### 使用ConcurrentHashMap

但是目前程序仍然存在问题：那就是性能差，不能并行计算，一开始尝试把synchronize的范围缩小，但是此时是写安全的，但是读仍然不安全，因此将hashmap换成concurrenthashmap

private final Map<A,V> cache = new HashMap<>();

改为

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

但是ConcurrentHashMap也会带来新的问题，它会导致在计算完成前，另一个要求计算相同值得请求到来的时候，会导致计算两边，这和缓存想避免多次计算的初衷恰恰相反，这是不可接受的。这个时候就要引入Future和Callable来避免重复计算的问题,修改后的代码如下：

package imooccache.computable;

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

/**
 * 利用future避免重复计算
 * @param <A>
 * @param <V>
 */

public class ImoocCache7<A,V> implements Computable<A,V> {


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

    private final Computable<A,V> c;

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

    @Override
    public V compute(A arg) throws Exception {
        Future<V> f = cache.get(arg);

        if(f==null){
          Callable<V> callable= new Callable<V>() {
                @Override
                public V call() throws Exception {
                    return c.compute(arg);
                }
            };

            FutureTask<V> ft = new FutureTask<>(callable);
            //因为此时f为空，所以需要对f赋值
            f = ft;
            cache.put(arg,ft);
            System.out.println("从futureTask调用了计算函数");
            ft.run();

        }

        return f.get();

    }


    public static void main(String[] args) {


        ImoocCache7<String, Integer> expensiveComputer = new ImoocCache7<>(
                new ExpensiveFunction());
        try {
            Integer result = expensiveComputer.compute("666");
            System.out.println("第一次计算结果"+result);
            result = expensiveComputer.compute("666");
            System.out.println("第二次计算结果"+result);
        } catch (Exception e) {
            e.printStackTrace();
        }




    }
}

用原子操作消除重复计算

上面的代码这个时候仍然会存在重复的可能，比如两个线程在时间上十分接近（可以看成同时）的运行相同的操作，那么仍然会创建两个任务去计算相同的值。这时候我们可以使用原子操作putIfAbsent来避免。

将上面代码中的put改成putIfAbsent

package imooccache;

import imooccache.computable.Computable;
import imooccache.computable.ExpensiveFunction;

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

/**
 * 利用future避免重复计算
 * @param <A>
 * @param <V>
 */

public class ImoocCache8<A,V> implements Computable<A,V> {


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

    private final Computable<A,V> c;

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

    @Override
    public V compute(A arg) throws Exception {
        Future<V> f = cache.get(arg);

        if(f==null){
          Callable<V> callable= new Callable<V>() {
                @Override
                public V call() throws Exception {
                    return c.compute(arg);
                }
            };

            FutureTask<V> ft = new FutureTask<>(callable);
           // f = ft;
    /****/      f =  cache.putIfAbsent(arg,ft);

          if(f==null){
              f = ft;
              System.out.println("从futureTask调用了计算函数");
              ft.run();
          }


        }

        return f.get();

    }


    public static void main(String[] args) {


        ImoocCache8<String, Integer> expensiveComputer = new ImoocCache8<>(
                new ExpensiveFunction());
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Integer result = expensiveComputer.compute("666");
                    System.out.println("第一次的计算结果：" + result);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Integer result = expensiveComputer.compute("666");
                    System.out.println("第三次的计算结果：" + result);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Integer result = expensiveComputer.compute("667");
                    System.out.println("第二次的计算结果：" + result);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

异常处理

对于一个高性能的缓存，我们还需要考虑针对各种异常的处理方法。

我们假设会出现异常

将

ImoocCache10<String, Integer> expensiveComputer = new ImoocCache10<>(new ExpensiveFunction());

改为

ImoocCache10<String, Integer> expensiveComputer = new ImoocCache10<>(new MayFail());

MayFail方法的具体代码如下：

package imooccache.computable;

import java.io.IOException;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 描述：     耗时计算的实现类，有概率计算失败
 *
 */
public class MayFail implements Computable<String, Integer>{

    @Override
    public Integer compute(String arg) throws Exception {
        double random = Math.random();
        if (random > 0.5) {
            throw new IOException("读取文件出错");
        }
        Thread.sleep(3000);
        return Integer.valueOf(arg);
    }
}

将compute方法进行改造

 @Override
    public V compute(A arg) throws InterruptedException, ExecutionException {
        while (true) {
            Future<V> f = cache.get(arg);
            if (f == null) {
                Callable<V> callable = new Callable<V>() {
                    @Override
                    public V call() throws Exception {
                        return c.compute(arg);
                    }
                };
                FutureTask<V> ft = new FutureTask<>(callable);
                f = cache.putIfAbsent(arg, ft);
                if (f == null) {
                    f = ft;
                    System.out.println("从FutureTask调用了计算函数");
                    ft.run();
                }
            }
            try {
                return f.get();
            } catch (CancellationException e) {
                System.out.println("被取消了");
                cache.remove(arg);
                throw e;
            } catch (InterruptedException e) {
                cache.remove(arg);
                throw e;
            } catch (ExecutionException e) {
                System.out.println("计算错误，需要重试");
                cache.remove(arg);
            }
        }
    }

缓存过期

通常情况下，缓存不是数据库，因此，我们需要对缓存设置过期时间。所以我们给缓存添加过期功能

在这里我们写一个compute的重载方法，参数中包含超时时间

   public V compute(A arg, long expire) throws ExecutionException, InterruptedException {
        if (expire>0) {
            executor.schedule(new Runnable() {
                @Override
                public void run() {
                    expire(arg);
                }
            }, expire, TimeUnit.MILLISECONDS);
        }
        return compute(arg);
    }

其中的expire方法的具体代码如下：

 public synchronized void expire(A key) {
        Future<V> future = cache.get(key);
        if (future != null) {
            if (!future.isDone()) {
                System.out.println("Future任务被取消");
                future.cancel(true);
            }
            System.out.println("过期时间到，缓存被清除");
            cache.remove(key);
        }
    }

整体的代码如下：

package imooccache;

import imooccache.computable.Computable;
import imooccache.computable.MayFail;
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.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 描述：     出于安全性考虑，缓存需要设置有效期，到期自动失效，否则如果缓存一直不失效，那么带来缓存不一致等问题
 */
public class ImoocCache10<A, V> implements Computable<A, V> {

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

    private final Computable<A, V> c;

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

    @Override
    public V compute(A arg) throws InterruptedException, ExecutionException {
        while (true) {
            Future<V> f = cache.get(arg);
            if (f == null) {
                Callable<V> callable = new Callable<V>() {
                    @Override
                    public V call() throws Exception {
                        return c.compute(arg);
                    }
                };
                FutureTask<V> ft = new FutureTask<>(callable);
                f = cache.putIfAbsent(arg, ft);
                if (f == null) {
                    f = ft;
                    System.out.println("从FutureTask调用了计算函数");
                    ft.run();
                }
            }
            try {
                return f.get();
            } catch (CancellationException e) {
                System.out.println("被取消了");
                cache.remove(arg);
                throw e;
            } catch (InterruptedException e) {
                cache.remove(arg);
                throw e;
            } catch (ExecutionException e) {
                System.out.println("计算错误，需要重试");
                cache.remove(arg);
            }
        }
    }

    public V computeRandomExpire(A arg) throws ExecutionException, InterruptedException {
        long randomExpire = (long) (Math.random() * 10000);
        return compute(arg, randomExpire);
    }

    public final static ScheduledExecutorService executor = Executors.newScheduledThreadPool(5);

    public V compute(A arg, long expire) throws ExecutionException, InterruptedException {
        if (expire>0) {
            executor.schedule(new Runnable() {
                @Override
                public void run() {
                    expire(arg);
                }
            }, expire, TimeUnit.MILLISECONDS);
        }
        return compute(arg);
    }

    public synchronized void expire(A key) {
        Future<V> future = cache.get(key);
        if (future != null) {
            if (!future.isDone()) {
                System.out.println("Future任务被取消");
                future.cancel(true);
            }
            System.out.println("过期时间到，缓存被清除");
            cache.remove(key);
        }
    }
    public static void main(String[] args) throws Exception {
        ImoocCache10<String, Integer> expensiveComputer = new ImoocCache10<>(
                new MayFail());
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Integer result = expensiveComputer.compute("666",5000L);
                    System.out.println("第一次的计算结果：" + result);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Integer result = expensiveComputer.compute("666");
                    System.out.println("第三次的计算结果：" + result);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Integer result = expensiveComputer.compute("667");
                    System.out.println("第二次的计算结果：" + result);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();


        Thread.sleep(6000L);
        Integer result = expensiveComputer.compute("666");
        System.out.println("主线程的计算结果：" + result);
    }
}

上面代码中，我们还考虑了如果出现大量缓存同时过期，这会导致出现缓存击穿的情况，因此对过期时间进行随机设置。

测试性能

package imooccache;

import imooccache.computable.ExpensiveFunction;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ImoocCache13 {

    static ImoocCache10<String,Integer> expensiveComputer = new ImoocCache10<>(new ExpensiveFunction());
    public static void main(String[] args) {
    //测试20000个线程执行的效率
        ExecutorService executorService = Executors.newFixedThreadPool(20000);


        for (int i = 0; i < 20000; i++) {
            executorService.submit(()->{
                Integer result = null;


                try {
                   result = expensiveComputer.compute("666");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }

                System.out.println(result);
            });
        }

    }
}

性能测试优化

因为现成的提交时间还是不是完全同步的，因此我们需要利用CountDownLatch来实现线程的统一触发

package imooccache;

import imooccache.computable.ExpensiveFunction;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 描述：     TODO
 */
public class ImoocCache12 {

    static ImoocCache10<String, Integer> expensiveComputer = new ImoocCache10<>(
            new ExpensiveFunction());
    public static CountDownLatch countDownLatch = new CountDownLatch(1);

    public static void main(String[] args) throws InterruptedException {
        ExecutorService service = Executors.newFixedThreadPool(100);
        long start = System.currentTimeMillis();
        for (int i = 0; i < 100; i++) {
            service.submit(() -> {
                Integer result = null;
                try {
                    System.out.println(Thread.currentThread().getName()+"开始等待");
                    countDownLatch.await();
                    SimpleDateFormat dateFormat = ThreadSafeFormatter.dateFormatter.get();
                    String time = dateFormat.format(new Date());
                    System.out.println(Thread.currentThread().getName()+"   "+time+"被放行");
                    result = expensiveComputer.compute("666");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
                System.out.println(result);
            });
        }

        Thread.sleep(5000);
        countDownLatch.countDown();
        service.shutdown();
    }
}
class ThreadSafeFormatter {

    public static ThreadLocal<SimpleDateFormat> dateFormatter = new ThreadLocal<SimpleDateFormat>() {

        //每个线程会调用本方法一次，用于初始化
        @Override
        protected SimpleDateFormat initialValue() {
            return new SimpleDateFormat("mm:ss");
        }

        //首次调用本方法时，会调用initialValue()；后面的调用会返回第一次创建的值
        @Override
        public SimpleDateFormat get() {
            return super.get();
        }
    };
}

码农公寓