AOP+Redisson 延时队列，实现缓存延时双删策略

一、缓存延时双删

关于缓存和数据库中的数据保持一致有很多种方案，但不管是单独在修改数据库之前，还是之后去删除缓存都会有一定的风险导致数据不一致。而延迟双删是一种相对简单并且收益比较高的实现最终一致性的方式，即在删除缓存之后，间隔一个短暂的时间后再删除缓存一次。这样可以避免并发更新时，假如缓存在第一次被删除后，被其他线程读到旧的数据更新到了缓存，第二次删除还可以补救，从而时间最终一致性。

实现延时双删的方案也有很多，有本地用 Thread.sleep(); 睡眠的方式做延时，也有借助第三方消息中间件做延时消息等等，本文基于 Redisson 中的延时队列进行实验。

在 Redisson 中提供了 RDelayedQueue 可以迅速实现延时消息，本文所使用的 Redisson 版本为 3.19.0 。

二、Redisson 实现延时消息

新建 SpringBoot 项目，在 pom 中加入下面依赖：

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId>
</dependency><dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.19.0</version>
</dependency>

在 yum 配置中，增加 redis 的信息：

spring:redis:timeout: 6000password:cluster:max-redirects:nodes:- 192.168.72.120:7001- 192.168.72.121:7001- 192.168.72.122:7001

声明 RedissonClient ：

@Configuration
public class RedissonConfig {@Beanpublic RedissonClient getRedisson(RedisProperties redisProperties) {Config config = new Config();String[] nodes = redisProperties.getCluster().getNodes().stream().filter(StringUtils::isNotBlank).map(node -> "redis://" + node).collect(Collectors.toList()).toArray(new String[]{});ClusterServersConfig clusterServersConfig = config.useClusterServers().addNodeAddress(nodes);if (StringUtils.isNotBlank(redisProperties.getPassword())) {clusterServersConfig.setPassword(redisProperties.getPassword());}clusterServersConfig.setConnectTimeout((int) (redisProperties.getTimeout().getSeconds() * 1000));clusterServersConfig.setScanInterval(2000);return Redisson.create(config);}
}

延时队列实现延时消息：

@Slf4j
@Component
public class MsgQueue {@ResourceRedissonClient redissonClient;public static final String QUEUE_KEY = "DELAY-QUEUE";// 发送消息public void send(String msg, Long time, TimeUnit unit) {// 获取队列RBlockingQueue<String> blockingQueue = redissonClient.getBlockingQueue(QUEUE_KEY);// 延时队列RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(blockingQueue);// 添加数据delayedQueue.offer(msg, time, unit);}// 消息监听@PostConstructpublic void listen() {CompletableFuture.runAsync(() -> {RBlockingQueue<String> blockingQueue = redissonClient.getBlockingQueue(MsgQueue.QUEUE_KEY);log.info("延时消息监听！");while (true) {try {consumer(blockingQueue.take());} catch (InterruptedException e) {throw new RuntimeException(e);}}});}// 消费消息public void consumer(String msg) {log.info("收到延时消息: {} , 当前时间: {} ", msg, LocalDateTime.now().toString());}}

测试延时消息：

@Slf4j
@RestController
@RequestMapping("/msg")
public class MsgController {@ResourceMsgQueue queue;@GetMapping("/test")public void test() {String msg = "你好";queue.send(msg, 5L, TimeUnit.SECONDS);}}

上面发送了延时5秒的消息，运行后可以看到日志：

三、AOP+延时队列，实现延时双删策略

缓存延时删除队列：

@Slf4j
@Component
public class CacheQueue {@ResourceRedissonClient redissonClient;public static final String QUEUE_KEY = "CACHE-DELAY-QUEUE";// 延时删除public void delayedDeletion(String key, Long time, TimeUnit unit) {RBlockingQueue<String> blockingQueue = redissonClient.getBlockingQueue(QUEUE_KEY);RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(blockingQueue);log.info("延时删除key: {} , 当前时间: {} ", key, LocalDateTime.now().toString());delayedQueue.offer(key, time, unit);}// 消息监听@PostConstructpublic void listen() {CompletableFuture.runAsync(() -> {RBlockingQueue<String> blockingQueue = redissonClient.getBlockingQueue(CacheQueue.QUEUE_KEY);while (true) {try {consumer(blockingQueue.take());} catch (InterruptedException e) {throw new RuntimeException(e);}}});}// 消费消息public void consumer(String key) {log.info("删除key: {} , 当前时间: {} ", key, LocalDateTime.now().toString());redissonClient.getBucket("key").delete();}
}

定义缓存和删除缓存注解：

@Retention(RetentionPolicy.RUNTIME)
@Documented
@Target(ElementType.METHOD)
public @interface Cache {String name() default "";
}

@Retention(RetentionPolicy.RUNTIME)
@Documented
@Target(ElementType.METHOD)
public @interface DeleteCache {String name() default "";
}

缓存AOP逻辑：

@Aspect
@Component
public class CacheAspect {@ResourceRedissonClient redissonClient;private final Long validityTime = 2L;@Pointcut("@annotation(com.bxc.retrydemo.anno.Cache)")public void pointCut() {}@Around("pointCut()")public Object aroundAdvice(ProceedingJoinPoint pjp) throws Throwable {Cache ann = ((MethodSignature) pjp.getSignature()).getMethod().getDeclaredAnnotation(Cache.class);if (Objects.nonNull(ann) && StringUtils.isNotBlank(ann.name())) {Object proceed = redissonClient.getBucket(ann.name()).get();if (Objects.nonNull(proceed)){return proceed;}}Object proceed = pjp.proceed();if (Objects.nonNull(ann) && StringUtils.isNotBlank(ann.name())) {redissonClient.getBucket(ann.name()).set(proceed, validityTime, TimeUnit.HOURS);}return proceed;}
}

延时双删 AOP 逻辑：

@Aspect
@Component
public class DeleteCacheAspect {@ResourceRedissonClient redissonClient;@ResourceCacheQueue cacheQueue;private final Long delayedTime = 3L;@Pointcut("@annotation(com.bxc.retrydemo.anno.DeleteCache)")public void pointCut() {}@Around("pointCut()")public Object aroundAdvice(ProceedingJoinPoint pjp) throws Throwable {// 第一次删除缓存DeleteCache ann = ((MethodSignature) pjp.getSignature()).getMethod().getDeclaredAnnotation(DeleteCache.class);if (Objects.nonNull(ann) && StringUtils.isNotBlank(ann.name())) {redissonClient.getBucket(ann.name()).delete();}// 执行业务逻辑Object proceed = pjp.proceed();//延时删除if (Objects.nonNull(ann) && StringUtils.isNotBlank(ann.name())) {cacheQueue.delayedDeletion(ann.name(), delayedTime, TimeUnit.SECONDS);}return proceed;}
}

伪业务逻辑使用：

@Slf4j
@Service
public class MsgServiceImpl implements MsgService {@Cache(name = "you key")@Overridepublic String find() {// 数据库操作// ....// 数据库操作结束return "数据结果";}@DeleteCache(name = "you key")@Overridepublic void update() {// 数据库操作// ....// 数据库操作结束}
}