《玩转Java并发工具、精通JUC、成为并发多面手》构建高性能缓存
原创《玩转Java并发工具、精通JUC、成为并发多面手》构建高性能缓存
原创
阿东
发布于 2023-06-27 09:58:25
发布于 2023-06-27 09:58:25
引言
《玩转Java并发工具、精通JUC、成为并发多面手》构建高性能缓存这部分的个人笔记。本节为单纯的实战，主要是把之前学习并发编程的知识点串起来。
挺有意思的一个demo，可以快速了解到一些并发编程的时候需要注意的一些问题。
目录
整个高性能构建的梳理思路如下：
- 使用最简单的HashMap
- 分析HashMap实现的问题
- 高并发访问重复计算性能问题
- 复用性能较差的问题
- 装饰模式抽象计算业务
- 解决复用性能较差的问题
- 使用Future改写计算实现接口
- 防止业务重复计算
- 如何处理高并发访问重复计算性能问题
- 增加泛型
- 缓存过期和增加随机性
- 为什么需要缓存过期？
- 防止缓存污染
- 为什么要增加随机性？
- 防止同一个时刻大量缓存过期增加系统压力
- 整体测试
代码
一个简单的小demo，可以直接拷贝下面的包所属的各个版本代码到自己的项目阅读即可：
https://gitee.com/lazyTimes/interview/tree/master/src/main/java/com/zxd/interview/mycache
一、构建步骤
1. 使用最简单的HashMap
最基础的版本实现非常简单，这是我们通常会想到的应用缓存实现方案，这里使用了Lombok的@Slf4j注解进行日志打印。
整个逻辑非常简单，首先通过计算方法匹配缓存，如果有就取缓存内容，否则就调用计算方法然后把结果缓存到HashMap当中。
/**
* 初版高速缓存实现
* 1. 使用简单的HashMap
* 2. 在高速缓存中进行计算
*
* 暴露问题：
* 1. 复用性能查
* 2. HashMap线程不安全，有可能重复判断
*/
@Slf4j
public class MyCacheVersion1 {
private final Map<String, Integer> cache = new HashMap<>();
/**
* 根据参数计算结果，此参数对于同一个参数会计算同样的结果
* 1. 如果缓存存在结果，直接返回
* 2. 如果缓存不存在，则需要计算添加到map之后才返回
* @param userId
* @return
*/
public Integer compute(String userId) throws InterruptedException {
if(cache.containsKey(userId)){
log.info("cached => {}", userId);
return cache.get(userId);
}
log.info("doCompute => {}", userId);
Integer result = doCompute(userId);
//不存在缓存就加入
cache.put(userId, result);
return result;
}
private Integer doCompute(String userId) throws InterruptedException {
TimeUnit.SECONDS.sleep(5);
return Integer.parseInt(userId);
}
}初版存在较多的问题，比较显著的问题是compute这个方法在多线程的环境是不安全的，我们可以编写测试程序验证。
在测试程序中，我们使用线程池构建100个线程处理1000个计算任务。从打印结果中我们随机抽取其中一个数字很容易出现计算两次的情况，比如下面的情况：
/**
* 对应当前版本的测试程序
* 1. HashMap线程不安全
* 2. 此程序验证线程安全问题
*/
@Slf4j
public class Test {
public static void main(String[] args) throws InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(100);
MyCacheVersion1 objectObjectMyCacheVersion1 = new MyCacheVersion1();
Random random = new Random(100);
for (int i = 0; i < 1000; i++) {
executorService.execute(() -> {
int randomInt = random.nextInt(100);
try {
Integer user = objectObjectMyCacheVersion1.compute(String.valueOf(randomInt));
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
}
executorService.shutdown();
}/**运行结果
测试结果：可以看到高并发的情况下非常有可能出现重复计算然后cache的情况
10:56:41.437 [pool-1-thread-53] INFO c.z.i.m.version1.MyCacheVersion1 - doCompute => 59
10:56:41.447 [pool-1-thread-97] INFO c.z.i.m.version1.MyCacheVersion1 - doCompute => 59
*/
}可以发现doCompute => 59算了多次。
处理线程不安全问题
线程不安全问题最简单的处理方式就是方法串行：
public synchronized Integer compute(String userId) throws InterruptedException如果加入synchronized，则整个线程的处理会串行执行，但是效率极低。
/**运行结果
串行之后执行效率极低，基本无法使用
11:14:15.851 [pool-1-thread-1] INFO c.z.i.m.version1.MyCacheVersion1 - doCompute => 15
11:14:20.862 [pool-1-thread-100] INFO c.z.i.m.version1.MyCacheVersion1 - doCompute => 52 11:14:25.874 [pool-1-thread-99] INFO c.z.i.m.version1.MyCacheVersion1 - doCompute => 55
*/2. 分析HashMap实现的问题
简单分析HashMap的缓存实现，主要的问题如下：
- 高并发访问重复计算性能问题
- 复用性能较差的问题
3. 装饰模式抽象计算业务
我们先解决复用性能较差的问题,，这里需要使用装饰模式进行改写。首先定义ComputeAble<A, V>接口，这个接口定义了抽象计算行为。
/**
* 可计算接口
* 装饰接口
*/
public interface ComputeAble<A, V>{
/**
* 根据指定参数 A 进行计算，计算结果为 V
* @description 根据指定参数 A 进行计算，计算结果为 V
* @param arg 泛型参数
* @return V 返回计算后结果
* @author xander
* @date 2023/6/15 15:42
*/
V doCompute(A arg) throws Exception;
}定义一个实现类实现计算接口。
/**
* 装饰模式改写接口
*/
public class ExpensiveCompute implements ComputeAble<String, Integer>{
@Override
public Integer doCompute(String arg) throws InterruptedException {
TimeUnit.SECONDS.sleep(5);
return Integer.parseInt(arg);
}
}第二个版本的缓存实现区别是使用了装饰模式封装计算方法，其他方法暂时不做调整。
/**
* 第二版，使用装饰模式进行改造
* synchronized 同步加锁
* @author
* @version v1.0.0
* @Package : version2
* @Description : 使用装饰模式进行改造
* @Create on : 2023/6/15 16:29
**/@Slf4j
public class MyCacheVersion2 {
/**
* 缓存
*/
private final Map<String, Integer> cache = new HashMap<>();
/**
* 计算方法实现对象
*/
private final static ComputeAble<String, Integer> COMPUTE = new ExpensiveCompute();
public synchronized Integer compute(String userId) throws Exception {
if(cache.containsKey(userId)){
log.info("cached => {}", userId);
return cache.get(userId);
}
log.info("doCompute => {}", userId);
Integer result = doCompute(userId);
// 不存在缓存就加入
cache.put(userId, result);
return result;
}
/**
* 计算方法由具体的类实现封装
* @param userId
* @return
* @throws InterruptedException
*/
private Integer doCompute(String userId) throws Exception {
return COMPUTE.doCompute(userId);
}
}通过上面的代码了解到，MyCacheVersion2 缓存实现类的具体计算逻辑抽象到具体的实现类当中，如果想要切换新的逻辑，可以改写COMPUTE的实现类。
4. 使用Future改写计算实现接口处理重复计算问题
我们简单分析HashMap的缓存实现，主要问题如下：
- 高并发访问重复计算性能问题
- <s>复用性能较差的问题</s>，通过装饰模式改写。
复用较差的问题处理了，下面介绍高并发访问重复计算问题处理办法，我们一步步介绍改写过程。为了方便理解这里暂时先用具体类型代替泛型。
- 首先把Map的value存储改为存储Future< V > 结果，并且把HashMap改为线程安全的ConcurrentHashMap。
private final Map<String, Future<Integer>> concurrentHashMap = new ConcurrentHashMap<>();- 构建 FutureTask 对象，这里使用 Lambda 表达式直接调用 doCompute 方法。
FutureTask<Integer> future = new FutureTask<>(() -> doCompute(userId));- 在计算函数中，大致的逻辑并没有改变，但是需要注意下面的细节：
- future.get() 在获取到结果之前会进行阻塞。
- ConcurrentHashMap 在类似如果不存在就加入的复合操作情况下需要考虑重复设置缓存的问题。
- putIfAbsent 可以做一些复合操作，如果设置缓存失败，说明有其他线程做过同样的操作，此时就可以重新操作一次获取结果即可。
- putIfAbsent不成功为什么不直接获取结果，而是要再计算一次，这是为了防止缓存刚好获取获取到一个null的值。
/**
* 第三个版本
* 1. 优化多线程访问重复计算问题
*
* * @author
* @version v1.0.0
* @Package : version3
* @Description : 第三个版本
* @Create on : 2023/6/15 16:47
**/@Slf4j
public class MyCacheVersion3 {
/**
* 改造，并发不安全集合改为并发安全集合
* value 存储为 future 的值
*/
private final Map<String, Future<Integer>> concurrentHashMap = new ConcurrentHashMap<>();
/**
* 计算实现类
*/
private static final ComputeAble<String, Integer> COMPUTEABLE = new ExpensiveCompute();
/**
* 先使用具体类型实现，后续改为使用泛型实现
* 1. 使用 FutureTask 对于要计算的值进行封装，根据 FutureTask 特性，获取到结果之前单个线程会一直等待
* 2. 由于计算方法变动，所有的代码需要调整
* 3. concurrentHashMap.get() 在 if 判断的时候依然存在非原子行为，所以在设置的时候使用 putIfAbsent 原子操作
*
* @param userId
* @return
* @throws InterruptedException
* @throws ExecutionException
*/ public Integer compute(String userId) throws InterruptedException, ExecutionException {
Future<Integer> result = concurrentHashMap.get(userId);
// 如果获取不到内容，说明不在缓存当中
if(Objects.isNull(result)){
// 此时利用callAble 线程任务指定任务获取，在获取到结果之前线程会阻塞
FutureTask<Integer> future = new FutureTask<>(() -> doCompute(userId));
//把新的future覆盖之前获取的future
result = future;
// 执行
future.run();
log.info("FutureTask 调用计算函数");
result = concurrentHashMap.putIfAbsent(userId, result);
// 如果返回null，说明这个记录被添加过了
if(Objects.isNull(result)){
log.info("其他线程进行设置,重新执行计算");
// 说明其他线程已经设置过值，这里重新跑一次计算方法即可直接获取
result = future;
// 再重新跑一次
future.run();
return result.get();
}else{
return result.get();
}
}
return result.get();
}
/**
* 计算方法由具体的类实现封装
* @param userId
* @return
* @throws InterruptedException
*/ private Integer doCompute(String userId) throws Exception {
return COMPUTEABLE.doCompute(userId);
}
}5. 泛型接口改写
有了上面的实现基础，改为为泛型就容易很多了，泛型的写法实际上就是把之前的具体类型转为泛型即可。
这里的代码可能不完整，建议把开头部分的仓库代码放到本地验证。
/**
* 第四个版本，方法改为泛型实现
* @author
* @version v1.0.0
* @Package : version3
* @Description : 第三个版本
* @Create on : 2023/6/15 16:47
**/@Slf4j
public class MyCacheVersion4<A, V> {
/**
* 改造，并发不安全集合改为并发安全集合
* value 存储为 future的值
*/
private final Map<A, Future<V>> concurrentHashMap = new ConcurrentHashMap<>();
private final ComputeAble computeAble = new ExpensiveCompute<>();
/**
* 先使用具体类型实现，后续改为使用泛型实现
* 1. 使用 FutureTask 对于要计算的值进行封装，根据 FutureTask 特性，获取到结果之前单个线程会一直等待
* 2. 由于计算方法变动，所有的代码需要调整
* 3. concurrentHashMap.get() 在 if 判断的时候依然存在非原子行为，所以在设置的时候使用 putIfAbsent 原子操作
* 4. 重构，使用泛型参数
* @param arg
* @return
* @throws InterruptedException
* @throws ExecutionException
*/
public V compute(A arg) throws InterruptedException, ExecutionException {
Future<V> result = concurrentHashMap.get(arg);
// 如果获取不到内容，说明不在缓存当中
if(Objects.isNull(result)){
// 此时利用callAble 线程任务指定任务获取，在获取到结果之前线程会阻塞
FutureTask<V> future = new FutureTask<>(new Callable<V>() {
@Override
@SuppressWarnings("unchecked")
public V call() throws Exception {
return (V) computeAble.doCompute(arg);
}
});
//把新的future覆盖之前获取的future
result = future;
// 执行
future.run();
log.info("FutureTask 调用计算函数");
result = concurrentHashMap.putIfAbsent(arg, result);
// 如果返回null，说明这个记录被添加过了
if(Objects.isNull(result)){
log.info("其他线程进行设置,重新执行计算");
// 说明其他线程已经设置过值，这里重新跑一次计算方法即可直接获取
result = future;
// 再重新跑一次
future.run();
return result.get();
}else{
return result.get();
}
}
return result.get();
}
}6. 可能失败的计算导致缓存污染问题处理
观察另一个计算实现，当我们的使用下面的方式会有什么样的效果？
/**
* 可能会出现失败的计算方法
* @author Xander
* @version v1.0.0
* @Package : compute
* @Description : 可能会出现失败的计算方法
* @Create on : 2023/6/19 10:40
**/public class MayFailCompute implements ComputeAble<String, Integer>{
/**
* 触发失败阈值
*/
private static final int FAILURE_THRESHOLD = 50;
/**
* 随机数生成器
*/
private static final Random RANDOM = new Random(100);
/**
* 有可能会出现失败的计算方法
* @description 有可能会出现失败的计算方法
* @param arg
* @return java.lang.Integer
* @author xander
* @date 2023/6/19 10:41
*/ @Override
public Integer doCompute(String arg) throws Exception {
if(RANDOM.nextInt() < FAILURE_THRESHOLD){
throw new Exception("自定义异常");
}
TimeUnit.MILLISECONDS.sleep(5);
return Integer.parseInt(arg);
}
}由于一开始我们就使用装饰模式改写过代码，所以要替换实现类非常简单：
private final ComputeAble computeAble = new MayFailCompute();测试的结果毫不意外的出现大量的失败。这样结果不符合预期，虽然 50%的失败率相当高，但实际上更多的是从缓存中获取的结果就是异常信息，这种情况就是缓存污染问题。
为了解决缓存污染问题，我们需要在try/catch中对于不同的情况进行不同的处理。在之前计算处理逻辑中一共会出现下面三种情况：
- CancellationException：线程被取消抛出的异常。
- InterruptedException：线程被中断时候抛出的异常。
- ExecutionException：试图检索一个因抛出异常而中止的任务的结果时抛出的异常。
对于不同的异常要对应不同的处理态度：
- CancellationException 和 InterruptedException 基本都是人为操作，这时候应该立即终止任务。
- 根据方法逻辑我们知道方法是有可能计算成功的，只不过需要多重试几次。
- while(true) 的加入可以让出错之后自动重新进行计算直到成功为止，但是如果是人为取消，就需要抛出异常并且手动结束任务。
我们把上面的处理思路转化为代码，相关注释已经加入，可以看下面的结果：
/**
* <pre>
* 第五个版本，当碰到会抛出异常的计算方法的情况这时候应该重新计算
* 对于不同的异常，也要对应不同的处理态度：
*
* - CancellationException 和 InterruptedException 基本都是人为操作，这时候应该立即终止任务。
* - 根据方法逻辑，我们可以知道方法是有可能计算成功的，只不过需要多重试几次。
* - while(true) 的加入可以让出错之后自动重新进行计算直到成功为止，但是如果是人为取消，就需要抛出异常并且结束。
* </pre>
* @author
* @version v1.0.0
* @Package : version3
* @Description : 第五个版本
* @Create on : 2023/6/15 16:47
**/@Slf4j
public class MyCacheVersion5<A, V> {
/**
* 改造，并发不安全集合改为并发安全集合
* value 存储为 future的值
*/
private final Map<A, Future<V>> concurrentHashMap = new ConcurrentHashMap<>();
private final ComputeAble computeAble = new MayFailCompute();
public V compute(A arg) {
return doCompute(arg);
}
private V doCompute(A arg) {
// 对于重复计算进行处理
while (true) {
Future<V> result = concurrentHashMap.get(arg);
try {
// 如果获取不到内容，说明不在缓存当中
if (Objects.isNull(result)) {
// 此时利用callAble 线程任务指定任务获取，在获取到结果之前线程会阻塞
FutureTask<V> future = new FutureTask<>(new Callable<V>() {
@Override
@SuppressWarnings("unchecked")
public V call() throws Exception {
return (V) computeAble.doCompute(arg);
}
});
//把新的future覆盖之前获取的future
result = future;
// 执行
future.run();
System.out.println("FutureTask 调用计算函数");
result = concurrentHashMap.putIfAbsent(arg, result);
// 如果返回null，说明这个记录被添加过了
if (Objects.isNull(result)) {
System.out.println("其他线程进行设置,重新执行计算");
// 说明其他线程已经设置过值，这里重新跑一次计算方法即可直接获取
result = future;
// 再重新跑一次
future.run();
return result.get();
} else {
return result.get();
}
}
return result.get();
} catch (CancellationException cancellationException) {
log.warn("CancellationException result => {}", result);
// 线程在执行过程当中有可能被取消
// 被取消的时候不管如何处理，首先需要先从缓存中移除掉污染缓存
concurrentHashMap.remove(arg);
throw new RuntimeException(cancellationException);
} catch (InterruptedException e) {
log.warn("InterruptedException result => {}", result);
// 线程被中断的异常处理
concurrentHashMap.remove(arg);
throw new RuntimeException(e);
} catch (ExecutionException e) {
// log.warn("ExecutionException result => {}", result);
log.info("移除缓存Key => {}，重新计算", arg);
concurrentHashMap.remove(arg);
// 不会抛出异常，而是重新在下一次循环中计算
// throw new RuntimeException(e);
/* 打印结果如下：
FutureTask 调用计算函数
FutureTask 调用计算函数
result => 65 其他线程进行设置,重新执行计算
result => 33 result => 65 result => 26 15:59:56.584 [pool-1-thread-30] INFO version5.MyCacheVersion5 - 移除缓存Key => 75，重新计算
result => 75 15:59:56.584 [pool-1-thread-3] INFO version5.MyCacheVersion5 - 移除缓存Key => 35，重新计算
15:59:56.584 [pool-1-thread-42] INFO version5.MyCacheVersion5 - 移除缓存Key => 67，重新计算
15:59:56.584 [pool-1-thread-36] INFO version5.MyCacheVersion5 - 移除缓存Key => 75，重新计算
15:59:56.584 [pool-1-thread-90] INFO version5.MyCacheVersion5 - 移除缓存Key => 40，重新计算
15:59:56.585 [pool-1-thread-31] INFO version5.MyCacheVersion5 - 移除缓存Key => 13，重新计算
15:59:56.586 [pool-1-thread-94] INFO version5.MyCacheVersion5 - 移除缓存Key => 60，重新计算
Disconnected from the target VM, address: '127.0.0.1:11054', transport: 'socket'
Process finished with exit code 0
* */
} catch (Exception e) {
log.warn("Exception result => {}", result);
concurrentHashMap.remove(arg);
// 无法处理的未知异常，直接抛出运行时异常不做任何处理。
throw new RuntimeException(e);
}
}
}
}最后是测试部分。
/**
* 可能失败的计算导致缓存污染问题处理
* 1. 解决缓存污染问题。
* 2. 异常情况尝试一直重复计算。
*
* @author Xander
* @version v1.0.0
* @Package : version5
* @Description :
* @Create on : 2023/6/19 10:49
**/public class Test {
public static void main(String[] args) throws InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(100);
MyCacheVersion5<String, Integer> myCacheVersion5 = new MyCacheVersion5<>();
Random random = new Random(100);
for (int i = 0; i < 1000; i++) {
executorService.submit(() -> {
int randomInt = random.nextInt(100);
Integer user = myCacheVersion5.compute(String.valueOf(randomInt));
System.out.println("result => " + user);
});
}
executorService.shutdown();
}/**
运行结果：
短期内会有海量异常，这不符合预期情况。根本原因是缓存不存在过期时间，会存在无效的内容缓存计算
其他线程进行设置,重新执行计算
其他线程进行设置,重新执行计算
FutureTask 调用计算函数
其他线程进行设置,重新执行计算
FutureTask 调用计算函数
其他线程进行设置,重新执行计算
FutureTask 调用计算函数
其他线程进行设置,重新执行计算
FutureTask 调用计算函数
其他线程进行设置,重新执行计算
其他线程进行设置,重新执行计算
FutureTask 调用计算函数
java.util.concurrent.ExecutionException: java.lang.Exception: 自定义异常
at java.base/java.util.concurrent.FutureTask.report(FutureTask.java:122)
at java.base/java.util.concurrent.FutureTask.get(FutureTask.java:191) at interview/version5.MyCacheVersion5.compute(MyCacheVersion5.java:63) at interview/version5.Test.lambda$main$0(Test.java:30) at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) at java.base/java.lang.Thread.run(Thread.java:829) Caused by: java.lang.Exception: 自定义异常
at interview/compute.MayFailCompute.doCompute(MayFailCompute.java:37)
at interview/compute.MayFailCompute.doCompute(MayFailCompute.java:14) at interview/version5.MyCacheVersion5$1.call(MyCacheVersion5.java:47) at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java) at interview/version5.MyCacheVersion5.compute(MyCacheVersion5.java:53) ... 7 more java.util.concurrent.ExecutionException: java.lang.Exception: 自定义异常
at java.base/java.util.concurrent.FutureTask.report(FutureTask.java:122)
at java.base/java.util.concurrent.FutureTask.get(FutureTask.java:191) at interview/version5.MyCacheVersion5.compute(MyCacheVersion5.java:63) at interview/version5.Test.lambda$main$0(Test.java:30) at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) at java.base/java.lang.Thread.run(Thread.java:829) Caused by: java.lang.Exception: 自定义异常
at interview/compute.MayFailCompute.doCompute(MayFailCompute.java:37)
at interview/compute.MayFailCompute.doCompute(MayFailCompute.java:14) at interview/version5.MyCacheVersion5$1.call(MyCacheVersion5.java:47) at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java) at interview/version5.MyCacheVersion5.compute(MyCacheVersion5.java:53) ... 7 more java.util.concurrent.ExecutionException: java.lang.Exception: 自定义异常
at java.base/java.util.concurrent.FutureTask.report(FutureTask.java:122)
at java.base/java.util.concurrent.FutureTask.get(FutureTask.java:191) at interview/version5.MyCacheVersion5.compute(MyCacheVersion5.java:63) at interview/version5.Test.lambda$main$0(Test.java:30) at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) at java.base/java.lang.Thread.run(Thread.java:829) Caused by: java.lang.Exception: 自定义异常
at interview/compute.MayFailCompute.doCompute(MayFailCompute.java:37)
at interview/compute.MayFailCompute.doCompute(MayFailCompute.java:14) at interview/version5.MyCacheVersion5$1.call(MyCacheVersion5.java:47) at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java) at interview/version5.MyCacheVersion5.compute(MyCacheVersion5.java:53) ... 7 more
经过修复之后：
16:09:07.705 [pool-1-thread-83] INFO version5.MyCacheVersion5 - 移除缓存Key => 9，重新计算
FutureTask 调用计算函数
result => 37
16:09:07.705 [pool-1-thread-84] INFO version5.MyCacheVersion5 - 移除缓存Key => 84，重新计算
16:09:07.705 [pool-1-thread-66] INFO version5.MyCacheVersion5 - 移除缓存Key => 84，重新计算
FutureTask 调用计算函数
FutureTask 调用计算函数
其他线程进行设置,重新执行计算
其他线程进行设置,重新执行计算
FutureTask 调用计算函数
16:09:07.705 [pool-1-thread-91] INFO version5.MyCacheVersion5 - 移除缓存Key => 9，重新计算
16:09:07.706 [pool-1-thread-2] INFO version5.MyCacheVersion5 - 移除缓存Key => 84，重新计算
16:09:07.706 [pool-1-thread-3] INFO version5.MyCacheVersion5 - 移除缓存Key => 40，重新计算
16:09:07.706 [pool-1-thread-5] INFO version5.MyCacheVersion5 - 移除缓存Key => 84，重新计算
FutureTask 调用计算函数
其他线程进行设置,重新执行计算
FutureTask 调用计算函数
FutureTask 调用计算函数
其他线程进行设置,重新执行计算
16:09:07.706 [pool-1-thread-91] INFO version5.MyCacheVersion5 - 移除缓存Key => 9，重新计算
其他线程进行设置,重新执行计算
16:09:07.706 [pool-1-thread-3] INFO version5.MyCacheVersion5 - 移除缓存Key => 40，重新计算
16:09:07.706 [pool-1-thread-5] INFO version5.MyCacheVersion5 - 移除缓存Key => 84，重新计算
FutureTask 调用计算函数
FutureTask 调用计算函数
其他线程进行设置,重新执行计算
FutureTask 调用计算函数
FutureTask 调用计算函数
*/
}7. 缓存过期和增加随机性
上面处理了缓存污染的问题，下面我们尝试为整个高性能缓存添加缓存过期时间，同时为防止“缓存雪崩”，增加过期时间随机性，为了方便理解，这里拆分两个小部分介绍如何处理。
缓存过期
实现缓存过期这里要用到定时线程池 ScheduledExecutorService。我们直接定一个带过期时间的新方法处理。
public V compute(A arg, long expireTime) {
if (expireTime > 0) {
SCHEDULED_EXECUTOR_SERVICE.schedule(new Runnable() {
@Override
public void run() {
// 定期清除缓存的方法
expire(arg);
}
/**
* @description 注意需要同步方法，防止多线程重复添加定时任务
* @param arg
* @return void
* @author xander
* @date 2023/6/20 16:58
*/ private synchronized void expire(A arg) {
// 检查当前 key 是否存在
Future<V> vFuture = concurrentHashMap.get(arg);
// 如果 value 存在，则需要进行
if(Objects.nonNull(vFuture)){
//如果任务被取消，此时需要关闭对应的定时任务
if(vFuture.isDone() ){
log.warn("future 任务被取消");
vFuture.cancel(true);
}
log.warn("过期时间到了，缓存被清除");
concurrentHashMap.remove(arg);
}
}
}, expireTime, TimeUnit.MILLISECONDS);
}
return doCompute(arg);
}增加随机性
增加随机性的目的是防止缓存在同一个时刻大量失效这种情况。增加随机性的最简单方法就是在设置超时时间的时候给一个随机random值。
public V compute(A arg, long expireTime, boolean isRandom){
if(isRandom){
return compute( arg, expireTime);
}else{
return compute( arg, expireTime + RANDOM.nextInt(1000));
}
}8. 完整代码
至此我们整个高性能缓存构建完成，最终的成果代码如下：
/**
* 高性能缓存第六版
*
* @author Xander
* @version v1.0.0
* @Package : version6
* @Description : 高性能缓存第六版
* @Create on : 2023/6/20 16:30
**/@Slf4j
public class MyCacheVersion6<A, V> {
private final Map<A, Future<V>> concurrentHashMap = new ConcurrentHashMap<>();
private static final Random RANDOM = new Random();;
private static final ScheduledExecutorService SCHEDULED_EXECUTOR_SERVICE = new ScheduledThreadPoolExecutor(10);
private final ComputeAble computeAble = new MayFailCompute();
public V compute(A arg) {
return doCompute(arg);
}
public V compute(A arg, long expireTime) {
if (expireTime > 0) {
SCHEDULED_EXECUTOR_SERVICE.schedule(new Runnable() {
@Override
public void run() {
// 定期清除缓存的方法
expire(arg);
}
/**
* @description 注意需要同步方法，防止多线程重复添加定时任务
* @param arg
* @return void
* @author xander
* @date 2023/6/20 16:58
*/ private synchronized void expire(A arg) {
// 检查当前 key 是否存在
Future<V> vFuture = concurrentHashMap.get(arg);
// 如果 value 存在，则需要进行
if(Objects.nonNull(vFuture)){
//如果任务被取消，此时需要关闭对应的定时任务
if(vFuture.isDone() ){
log.warn("future 任务被取消");
vFuture.cancel(true);
}
log.warn("过期时间到了，缓存被清除");
concurrentHashMap.remove(arg);
}
}
}, expireTime, TimeUnit.MILLISECONDS);
}
return doCompute(arg);
}
public V compute(A arg, long expireTime, boolean isRandom){
if(isRandom){
return compute( arg, expireTime);
}else{
return compute( arg, expireTime + RANDOM.nextInt(1000));
}
}
private V doCompute(A arg) {
// 对于重复计算进行处理
while (true) {
Future<V> result = concurrentHashMap.get(arg);
try {
// 如果获取不到内容，说明不在缓存当中
if (Objects.isNull(result)) {
// 此时利用callAble 线程任务指定任务获取，在获取到结果之前线程会阻塞
FutureTask<V> future = new FutureTask<>(new Callable<V>() {
@Override
@SuppressWarnings("unchecked")
public V call() throws Exception {
return (V) computeAble.doCompute(arg);
}
});
//把新的future覆盖之前获取的future
result = future;
// 执行
future.run();
System.out.println("FutureTask 调用计算函数");
result = concurrentHashMap.putIfAbsent(arg, result);
// 如果返回null，说明这个记录被添加过了
if (Objects.isNull(result)) {
System.out.println("其他线程进行设置,重新执行计算");
// 说明其他线程已经设置过值，这里重新跑一次计算方法即可直接获取
result = future;
// 再重新跑一次
future.run();
return result.get();
} else {
return result.get();
}
}
return result.get();
} catch (CancellationException cancellationException) {
log.warn("CancellationException result => {}", result);
// 线程在执行过程当中有可能被取消
// 被取消的时候不管如何处理，首先需要先从缓存中移除掉污染缓存
concurrentHashMap.remove(arg);
throw new RuntimeException(cancellationException);
} catch (InterruptedException e) {
log.warn("InterruptedException result => {}", result);
// 线程被中断的异常处理
concurrentHashMap.remove(arg);
throw new RuntimeException(e);
} catch (ExecutionException e) {
// log.warn("ExecutionException result => {}", result);
log.info("移除缓存Key => {}，重新计算", arg);
concurrentHashMap.remove(arg);
// 不会抛出异常，而是重新在下一次循环中计算
// throw new RuntimeException(e);
} catch (Exception e) {
log.warn("Exception result => {}", result);
concurrentHashMap.remove(arg);
// 无法处理的未知异常，直接抛出运行时异常不做任何处理。
throw new RuntimeException(e);
}
}
}
}测试代码
/**
* 缓存过期功能测试
* @author Xander
* @version v1.0.0
* @Package : version6
* @Description : 缓存过期功能测试
* @Create on : 2023/6/20 17:06
**/
public class Test {
public static void main(String[] args) throws InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(100);
MyCacheVersion6<String, Integer> myCacheVersion5 = new MyCacheVersion6<>();
Random random = new Random(100);
for (int i = 0; i < 1000; i++) {
executorService.submit(() -> {
int randomInt = random.nextInt(100);
Integer user = myCacheVersion5.compute(String.valueOf(randomInt));
System.out.println("result => " + user);
});
}
executorService.shutdown();
}
}二、测试缓存性能
测试缓存性能的点包含下面的部分：
- 使用线程池测试高性能缓存的性能
- 使用CountDownLatch压力测试
- 线程安全类ThreadSafeFormatter验证CountDownLatch
之前我们的Test测试都是使用线程池的模式，这里不过多介绍，这里提一下如何使用CountDownLatch进行”压力测试“，以及使用ThreadSafeFormatter验证CountDownLatch的性能。
/**
* ThreadSafeFormatter * @author Xander
* @version v1.0.0
* @Package : com.zxd.interview.mycache.version7
* @Description : 线程安全 ThreadSafeFormatter
* @Create on : 2023/6/22 11:11
**/public class ThreadSafeFormatter {
public static ThreadLocal<SimpleDateFormat> dateFormatter = new ThreadLocal<SimpleDateFormat>() {
/**
* 每个线程会调用本方法一次，用于初始化
* @description 每个线程会调用本方法一次，用于初始化
* @param
* @return java.text.SimpleDateFormat
* @author xander
* @date 2023/6/22 11:30
*/
@Override
protected SimpleDateFormat initialValue() {
return new SimpleDateFormat("mm:ss");
}
/**
* 首次调用本方法时，会调用initialValue()；后面的调用会返回第一次创建的值
* @description 首次调用本方法时，会调用initialValue()；后面的调用会返回第一次创建的值
* @param
* @return java.text.SimpleDateFormat
* @author xander
* @date 2023/6/22 11:30
*/
@Override
public SimpleDateFormat get() {
return super.get();
}
};
}
/**
* 整体性能测试
* @author Xander
* @version v1.0.0
* @Package : com.zxd.interview.mycache.version6
* @Description : 整体性能测试
* @Create on : 2023/6/20 17:06
**/public class Test {
public static void main(String[] args) throws InterruptedException {
long beginTime = System.currentTimeMillis();
ExecutorService executorService = Executors.newFixedThreadPool(100);
MyCacheVersion6<String, Integer> myCacheVersion5 = new MyCacheVersion6<>();
Random random = new Random(200);
CountDownLatch countDownLatch = new CountDownLatch(1);
for (int i = 0; i < 100; i++) {
executorService.submit(() -> {
int randomInt = random.nextInt(100);
try {
countDownLatch.await();
// 从线程安全的集合当中取出当前时间
SimpleDateFormat simpleDateFormat = ThreadSafeFormatter.dateFormatter.get();
System.out.println("simpleDateFormat => "+ simpleDateFormat.format(new Date()));
} catch (InterruptedException e) {
e.printStackTrace();
}
Integer user = myCacheVersion5.compute(String.valueOf(randomInt), 5000);
System.out.println("result => " + user);
});
}
// 假设此时所有的请求需要5秒时间准备。
Thread.sleep(1000);
countDownLatch.countDown();
executorService.shutdown();
long endTime = System.currentTimeMillis();
// 如果线程池没有停止一直死循环
while(!executorService.isTerminated()){
}
System.out.println("最终时间" + (endTime - beginTime));
}/**
10:59:34.521 [pool-2-thread-3] WARN com.zxd.interview.mycache.version6.MyCacheVersion6 - future 任务被取消
10:59:34.522 [pool-2-thread-3] WARN com.zxd.interview.mycache.version6.MyCacheVersion6 - 过期时间到了，缓存被清除
10:59:34.521 [pool-2-thread-4] WARN com.zxd.interview.mycache.version6.MyCacheVersion6 - future 任务被取消
10:59:34.522 [pool-2-thread-4] WARN com.zxd.interview.mycache.version6.MyCacheVersion6 - 过期时间到了，缓存被清除
10:59:34.521 [pool-2-thread-1] WARN com.zxd.interview.mycache.version6.MyCacheVersion6 - future 任务被取消
10:59:34.522 [pool-2-thread-1] WARN com.zxd.interview.mycache.version6.MyCacheVersion6 - 过期时间到了，缓存被清除
10:59:34.521 [pool-2-thread-8] WARN com.zxd.interview.mycache.version6.MyCacheVersion6 - future 任务被取消
10:59:34.522 [pool-2-thread-8] WARN com.zxd.interview.mycache.version6.MyCacheVersion6 - 过期时间到了，缓存被清除
10:59:34.521 [pool-2-thread-7] WARN com.zxd.interview.mycache.version6.MyCacheVersion6 - 过期时间到了，缓存被清除
10:59:34.521 [pool-2-thread-2] WARN com.zxd.interview.mycache.version6.MyCacheVersion6 - future 任务被取消
10:59:34.522 [pool-2-thread-2] WARN com.zxd.interview.mycache.version6.MyCacheVersion6 - 过期时间到了，缓存被清除
10:59:34.522 [pool-2-thread-5] WARN com.zxd.interview.mycache.version6.MyCacheVersion6 - 过期时间到了，缓存被清除
最终时间146
如果修改为多个时间同时发起请求：
最终时间1074 - 1000 主线程的睡眠时间 = 74
可以看到时间点都是在同一个分秒，可以人为countDownlatch是生效的
simpleDateFormat => 14:50
simpleDateFormat => 14:50 simpleDateFormat => 14:50 simpleDateFormat => 14:50 simpleDateFormat => 14:50 simpleDateFormat => 14:50 simpleDateFormat => 14:50 simpleDateFormat => 14:50 simpleDateFormat => 14:50 simpleDateFormat => 14:50 simpleDateFormat => 14:50 simpleDateFormat => 14:50 simpleDateFormat => 14:50 simpleDateFormat => 14:50 simpleDateFormat => 14:50
*/
}
}三、写在最后
这个小demo在自己实际上手写的时候还是有不少卡壳的地方，很多时候看懂不一定能写出来，如果有疑问或者改进意见欢迎讨论。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
评论
登录后参与评论
推荐阅读
目录