RedissonClient延时队列延时队列做双删(高并发下保持redis缓存和数据库一致性)
1.导入jar包<!-- redisson --><dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.16.8</version></dependency&
·
使用场景
例如:电商系统,用户下单后超过15分钟未支付则自动关闭订单
1.导入jar包
<!-- redisson -->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.16.8</version>
</dependency>
2.redis yml 配置
spring:
#数据源配置
redis:
database: 6
host: xxxx
password: xxx
port: 6379
ssl: false
jedis:
pool:
max-wait: 30000
max-active: 100
max-idle: 50
min-idle: 10
timeout: 10000
3.CacheConfigEnums 缓存枚举
package com.yjx.cloud.common.enums;
import java.time.Duration;
import java.util.Arrays;
import java.util.stream.Collectors;
/**
* 系统缓存,及基础配置
*
* @author yangjunxiong
* @date 2018/7/19 20:48
*/
public enum CacheConfigEnums {
/**
* 【 延时队列 】
**/
R_BLOCKING_QUEUE(commonCacheKey() + "rBlockingQueue", Duration.ZERO),
/**
* 永久缓存,永不过期 ttl永久 = Duration.ZERO
*/
PERPETUAL_CACHE(commonCacheKey() + "perpetualCH", Duration.ZERO),
/**
* 默认缓存 ,1天
*/
DEFAULT_CACHE(commonCacheKey() + "defaultCH", Duration.ofDays(1)),
SOCO_LONGITUDE_TO_ADDRESS_CACHE(commonCacheKey() + "socoLongitudeToAddressCH", Duration.ofDays(1)),
/**
* 【 测试缓存 】
**/
TEST_DTO_CACHE(commonCacheKey() + "testDtoCH", Duration.ofDays(1)),
;
private final String cacheName; //缓存名
private final Duration ttl; //ttl
static {
//验证【value】必须是唯一的
int size = Arrays.stream(CacheConfigEnums.values())
.map(CacheConfigEnums::getKey)
.collect(Collectors.toSet())
.size();
}
//公共缓存key
private static final String commonCacheKey = "yjxService:";
public static String commonCacheKey() {
return commonCacheKey;
}
CacheConfigEnums(String cacheName, Duration ttl) {
this.cacheName = cacheName;
this.ttl = ttl;
}
/**
* 获取缓存名
*
* @return
*/
public String getKey() {
return cacheName;
}
/**
* 获取TTL
*
* @return
*/
public Duration getTTL() {
return this.ttl;
}
}
- RedissonClient 配置
package com.yjx.cloud.common.config;
import com.yjx.cloud.common.enums.CacheConfigs;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cache.CacheManager;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.cache.RedisCacheConfiguration;
import org.springframework.data.redis.cache.RedisCacheManager;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
@Slf4j
@Configuration
@EnableCaching
public class RedisConfig {
@Bean
public CacheManager redisCacheManager(RedisConnectionFactory redisConnectionFactory) {
Map<String, RedisCacheConfiguration> cacheConfiguration = new HashMap<>();
for (CacheConfigs cacheConfigs : CacheConfigs.values()) {
cacheConfiguration.put(cacheConfigs.getKey(),
RedisCacheConfiguration.defaultCacheConfig().entryTtl(cacheConfigs.getTTL()));
}
return RedisCacheManager.RedisCacheManagerBuilder
.fromConnectionFactory(redisConnectionFactory)
.initialCacheNames(cacheConfiguration.keySet())
.cacheDefaults(RedisCacheConfiguration.defaultCacheConfig().entryTtl(Duration.ofDays(1)))
.withInitialCacheConfigurations(cacheConfiguration)
.disableCreateOnMissingCache() //如果缓存不存在,不允许创建
.build();
}
@Data
public static class RedissonConfigBean {
private String host;
private String port;
private String password;
private Integer timeout;
private Integer database;
}
@Bean
@ConfigurationProperties("spring.redis")
public RedissonConfigBean redissonConfigBean() {
return new RedissonConfigBean();
}
@Bean(destroyMethod = "shutdown")
public RedissonClient redissonClient() {
Config config = new Config();
config.useSingleServer()
.setAddress("redis://" + redissonConfigBean().getHost() + ":" + redissonConfigBean().getPort())
.setPassword(redissonConfigBean().getPassword())
.setTimeout(redissonConfigBean().getTimeout())
.setDatabase(redissonConfigBean().getDatabase())
;
return Redisson.create(config);
}
}
- RedisQueueConfig 配置
/*
* Copyright 2022 Wicrenet, Inc. All rights reserved.
*/
package com.yjx.cloud.test1.config;
import com.yjx.cloud.common.dto.RedisQueueData;
import com.yjx.cloud.common.enums.CacheConfigEnums;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* 【 redis延时队列 】
*
* @author yangjunxiong
* Created on 2022/2/24 13:50
*/
@Slf4j
@Configuration
public class RedisQueueConfig {
private final CacheConfigEnums queueName = CacheConfigEnums.R_BLOCKING_QUEUE;
@Bean("rBlockingQueue")
public RBlockingQueue<RedisQueueData> rBlockingQueue(@Qualifier("redissonClient") RedissonClient redissonClient) {
return redissonClient.getBlockingQueue(queueName.getKey());
}
@Bean("rDelayedQueue")
public RDelayedQueue<RedisQueueData> rDelayedQueue(@Qualifier("redissonClient") RedissonClient redissonClient, @Qualifier("rBlockingQueue") RBlockingQueue<RedisQueueData> blockQueue) {
return redissonClient.getDelayedQueue(blockQueue);
}
}
6.RedisQueueController 给redis队列造数据 延时双删
/*
* Copyright 2022 Wicrenet, Inc. All rights reserved.
*/
package com.yjx.cloud.test1.controller.api;
import com.yjx.cloud.common.core.api.Result;
import com.yjx.cloud.common.enums.CacheConfigEnums;
import com.yjx.cloud.common.enums.RedisQueueEnums;
import com.yjx.cloud.test1.domain.dto.TestDto;
import com.yjx.cloud.test1.schedule.RedisQueueTask;
import io.swagger.annotations.Api;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* 【】
*
* @author yangjunxiong
* Created on 2022/2/24 14:26
*/
@Slf4j
@RestController
@Api(tags = {"test1"})
@RequestMapping("{server-path}/api")
public class RedisQueueController {
@Autowired
private RedisQueueTask redisQueueTask;
@Autowired
private RedisTemplate redisTemplate;
/**
* 【 延迟消息队列 】
**/
@GetMapping("/offer")
public Result<Boolean> offer() {
TestDto testDto = new TestDto();
testDto.setId(11L);
this.redisQueueTask.pullData(RedisQueueEnums.TEST_QUEUE, testDto);
return Result.success(true);
}
/**
* 【 利用延迟消息队列实现双删,保持数据库和缓存一致性 】 先删除redis缓存,再操作数据库(修改或删除数据),然后发消息延迟1秒再次删除redis缓存
**/
@GetMapping("/delete")
public Result<Boolean> delete(TestDto testDto) {
String key = CacheConfigEnums.TEST_DTO_CACHE.getKey() + testDto.getId();
redisTemplate.delete(key);//删除redis缓存
// this.delete(testDto.getId());//这里写操作数据库逻辑 (我这写的是伪代码删除数据库数据)
this.redisQueueTask.pullData(RedisQueueEnums.DELETE_REDIS_CACHE_QUEUE, key);//发消息 延迟删除redis缓存
return Result.success(true);
}
}
- RedisQueueTask. 消费redis队列数据 接口
/*
* Copyright 2022 Wicrenet, Inc. All rights reserved.
*/
package com.yjx.cloud.test1.schedule;
import cn.hutool.json.JSONUtil;
import com.yjx.cloud.common.dto.RedisQueueData;
import com.yjx.cloud.common.enums.CacheConfigEnums;
import com.yjx.cloud.common.enums.RedisQueueEnums;
import com.yjx.cloud.test1.domain.dto.TestDto;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.util.concurrent.TimeUnit;
/**
* 【 redis延时队列消费 】
*
* @author yangjunxiong
* Created on 2022/2/24 13:50
*/
@Slf4j
@Component
public class RedisQueueTask {
@Autowired
@Qualifier("rDelayedQueue")
private RDelayedQueue<RedisQueueData> rDelayedQueue;
@Autowired
@Qualifier("rBlockingQueue")
private RBlockingQueue<RedisQueueData> rBlockingQueue;
private final CacheConfigEnums queueName = CacheConfigEnums.R_BLOCKING_QUEUE;
@Autowired
private RedisTemplate redisTemplate;
/**
* 【 延时队列发消息 】
**/
public void pullData(RedisQueueEnums redisQueueEnums, Object data) {
RedisQueueData dto = new RedisQueueData();
dto.setData(JSONUtil.toJsonStr(data));
dto.setRedisQueueEnums(redisQueueEnums);
this.rDelayedQueue.offerAsync(dto, redisQueueEnums.getTtl().getSeconds(), TimeUnit.SECONDS);
}
/**
* 【 延时队列消费消息 】
**/
@PostConstruct
public void take() {
Thread thread = new Thread(() -> {
while (true) {
try {
RedisQueueData data = rBlockingQueue.take();
switch (data.getRedisQueueEnums()) {
case DEFAULT_QUEUE:
break;
case DELETE_REDIS_CACHE_QUEUE:
this.redisTemplate.delete(data.getData());
break;
case TEST_QUEUE:
TestDto testDto = JSONUtil.toBean(data.getData(), TestDto.class);
log.info("testDto:{}", testDto);
break;
}
} catch (InterruptedException e) {
try {
TimeUnit.MILLISECONDS.sleep(500);
} catch (InterruptedException interruptedException) {
interruptedException.printStackTrace();
}
}
}
});
thread.setName(queueName.getKey() + "-线程");
thread.setDaemon(true);
thread.start();
}
}
更多推荐
所有评论(0)