奶奶看了都点头的自定义缓存刷新方案

目录

背景

思路

代码实现

CacheKeyEnum.java 枚举类

RefreshCacheAnn.java 注解

RefreshCacheAspect.java 切面 

CacheQueryHelper.java 查询器

刷新事件&监听器

缓存加载器 & 基于redis的加载

用法

1、在相应动作点加注解：

2、查询时用法

用到的工具类

背景

        业务表的单表缓存，大幅提升项目中接口的查询速度。

思路

1、实现一个枚举，记录单表缓存的配置项。

2、实现一个缓存注解

3、实现切面，在方法成功返回后切入

4、增加事件监听，确保事务提交后再刷新。

5、增加刷新机制：启动、到达切点时刷新。

代码实现

CacheKeyEnum.java 枚举类

public enum CacheKeyEnum {

	T_API("t_api", "id",""),
	T_DYNAMIC_MASKING_STRATEGY("t_dynamic_masking_strategy", "id",""),
	T_DYNAMIC_MASKING_FIELD("t_dynamic_masking_field", "id",""),
	T_DYNAMIC_MASKING_TABLE("t_dynamic_masking_table", "id",""),
	T_DYNAMIC_MASKING_USER("t_dynamic_masking_user", "id",""),
	T_RES_TYPE("t_res_type", "id",""),
	T_DESENST_ALGORITHM("t_desenst_algorithm", "id",""),
	T_TRANSFER_APPLY("t_transfer_apply", "id",""),
	T_TRANSFER("t_transfer", "id",""),
	T_LEVEL_THEME("t_level_theme", "id",""),
	T_TRANSFER_AUTHORIZE("t_transfer_authorize", "id",""),
	T_DATA_SOURCE("t_data_source", "id",""),
	T_ALARM_RULE("t_alarm_rule", "id","is_deleted"),
	T_ALARM_OPS_CONTACT("t_alarm_ops_contact", "id","is_deleted"),
	;


	public static final String DSJ_CACHE = "dsj:cache:";
	/**
	 * 表名
	 */
	public String table;

	/**
	 * 数据表中
	 */
	public String tablePrimaryKey;

	/**
	 * 删除字段
	 */
	public String deletedField;
	/**
	 * redis key
	 */
	public String key;


	CacheKeyEnum(String table, String tablePrimaryKey, String deletedField) {
		this.table = table;
		this.tablePrimaryKey = tablePrimaryKey;
		this.deletedField = deletedField;
		this.key = DSJ_CACHE + table;
	}

	public static CacheKeyEnum of(String table) {
		for (CacheKeyEnum cacheKeyEnum : CacheKeyEnum.values()) {
			if (cacheKeyEnum.table.equals(table)) {
				return cacheKeyEnum;
			}

		}
		return null;
	}
}

RefreshCacheAnn.java 注解

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface RefreshCacheAnn {
/**
* 配置缓存key枚举，每个表对应一个枚举
* @return
*/
CacheKeyEnum[] value();
}

RefreshCacheAspect.java 切面 

@Slf4j
@Aspect
@Component
public class RefreshCacheAspect {

	@Resource
	private ApplicationEventPublisher applicationEventPublisher;


	@AfterReturning("@annotation(refreshCacheAnn)")
	public void triggerCacheRefresh(JoinPoint point, RefreshCacheAnn refreshCacheAnn) {
		CacheKeyEnum[] cacheKeyEnum = refreshCacheAnn.value();
		applicationEventPublisher.publishEvent(new CacheRefreshEvent(point.getThis(), cacheKeyEnum));
	}

}

CacheQueryHelper.java 查询器

@Component
public class CacheQueryHelper {

    @Autowired
    private RedisUtils redisUtils;

    /**
     * 从缓存中获取数据
     *
     * @param cacheKeyEnum cacheKey
     * @param id           记录ID
     * @param clazz        返回封装的类类型
     * @param callback     缓存不存在的回调函数
     * @param <T>          缓存对象类型
     * @return T 缓存对象
     */
    public  <T> T get(CacheKeyEnum cacheKeyEnum, String id, Class<T> clazz, Callback<T> callback) {
        T t = redisUtils.hgetOneByClass(cacheKeyEnum.key, id, clazz);
        if (null == t) {
            t = callback.getObjForDatabase();
        }
        return t;
    }

    /**
     * 从缓存中获取所有的数据
     * @param cacheKeyEnum
     * @param clazz
     * @return
     * @param <T>
     */
    public  <T> List<T> getAll(CacheKeyEnum cacheKeyEnum, Class<T> clazz) {
       return redisUtils.hgetListByClass(cacheKeyEnum.key, clazz);
    }

    /**
     * 通过条件查找
     * @param cacheKeyEnum 枚举Key
     * @param clazz 返回类类型
     * @param condition 条件
     * @return 返回符合条件的集合
     * @param <T> 表映射类型
     */
    public <T> List<T> getListByCondition(CacheKeyEnum cacheKeyEnum, Class<T> clazz, Condition<T> condition){
        return redisUtils.hgetListByClass(cacheKeyEnum.key, clazz).stream().filter(condition::condition).collect(Collectors.toList());
    }
    public <T> T getObjectByCondition(CacheKeyEnum cacheKeyEnum,Class<T> clazz,Condition<T> condition){
        List<T> collect = redisUtils.hgetListByClass(cacheKeyEnum.key, clazz).stream().filter(condition::condition).collect(Collectors.toList());
        return CollUtil.isEmpty(collect)?null:collect.get(0);
    }

	/**
	 * 根据条件查询不存在时则回调查库
	 * @since 1:48 PM 2024/2/28
	 * @param cacheKeyEnum cacheKeyEnum
	 * @param clazz clazz
	 * @param condition condition
	 * @param callback callback
	 * @return {@link T}
	 **/
	public <T> T getObjectByCondition(CacheKeyEnum cacheKeyEnum,Class<T> clazz,Condition<T> condition, Callback<T> callback){
		List<T> collect = redisUtils.hgetListByClass(cacheKeyEnum.key, clazz).stream().filter(condition::condition).collect(Collectors.toList());

		if(CollUtil.isEmpty(collect)){
			return callback.getObjForDatabase();
		}

		return CollUtil.isEmpty(collect)?null:collect.get(0);
	}

    public interface Condition<T>{
       boolean condition(T t);
    }

    public interface Callback<T> {
        T getObjForDatabase();
    }
}

刷新事件&监听器

public class CacheRefreshEvent extends ApplicationEvent {

	private CacheKeyEnum[] keys;

	public CacheRefreshEvent(Object source, CacheKeyEnum... keys) {
		super(source);
		this.keys = keys;
	}

	public CacheKeyEnum[] getKeys() {
		return keys;
	}
}


@Component
public class CacheRefreshListener implements ApplicationListener<CacheRefreshEvent> {

	@Resource
	private CacheInitializerImpl cacheInitializer;

	@Override
	public void onApplicationEvent(CacheRefreshEvent event) {
		CacheKeyEnum[] cacheKeyEnum = event.getKeys();
		cacheInitializer.refresh(cacheKeyEnum);
	}
}

缓存加载器 & 基于redis的加载

public interface CacheInitializer {

	/**
	 * 加载缓存
	 */
	void loadOnStartup();

	/**
	 * 重新加载缓存
	 */
	void reloadDaily();

	/**
	 * 加载缓存
	 */
	void refresh(CacheKeyEnum... cacheKeyEnums);

	/**
	 * 清空缓存
	 */
	void clearCache(String key);


}


@Component
public class CacheRefreshListener implements ApplicationListener<CacheRefreshEvent> {

	@Resource
	private CacheInitializerImpl cacheInitializer;

	@Override
	public void onApplicationEvent(CacheRefreshEvent event) {
		CacheKeyEnum[] cacheKeyEnum = event.getKeys();
		cacheInitializer.refresh(cacheKeyEnum);
	}
}





@Component
@Slf4j
public class CacheInitializerImpl implements CacheInitializer {

	@Resource
	private RedisTemplate<String, Object> redisTemplate;

	/**
	 * 线程池处理所有的刷新缓存任务
	 */
	private final ExecutorService executorService = Executors.newFixedThreadPool(2);

	@Autowired
	private JdbcTemplate jdbcTemplate;

	public static final int MAX_BATCH_SIZE = 5000;

	public static final String[] TABLES = {
		""
	};

	@Override
	@PostConstruct
	public void loadOnStartup() {
		log.info("[cache]xxx平台缓存初始化开始...");
		refresh(CacheKeyEnum.values());
		log.info("[cache]xxx平台缓存初始化结束.");
	}

	@Scheduled(cron = "0 0 23 * * ?")
	@Override
	public void reloadDaily() {
		log.info("[cache]每日23点重置xxx平台缓存开始...");
		refresh(CacheKeyEnum.values());
		log.info("[cache]每日23点重置xxx平台缓存完成");
	}

	@Override
	public void refresh(CacheKeyEnum... cacheKeyEnums) {
		Arrays.stream(cacheKeyEnums).forEach(this::doRefresh);
	}

	/**
	 * 分页查询数据表中的数据，依次加入至缓存中
	 *
	 * @param cacheKeyEnum 缓存key枚举
	 */
	private void doRefresh(CacheKeyEnum cacheKeyEnum) {
		executorService.execute(() -> {
			if (null == cacheKeyEnum) {
				return;
			}
			clearCache(cacheKeyEnum.key);
			Page page = new Page();
			page.setPageSize(MAX_BATCH_SIZE);
			try {
				//分页查询表中所有数据
				Entity entity = Entity.create(cacheKeyEnum.table);
				Optional.ofNullable(cacheKeyEnum.deletedField).filter(StringUtils::isNotEmpty).ifPresent(deletedField -> entity.set(deletedField, 0));

				PageResult<Entity> result = Db.use(jdbcTemplate.getDataSource())
					.page(entity, page);
				int currentSize = result.size();
				
				pipelineToRedis(result, cacheKeyEnum, result.getTotal(), currentSize);
				while (!result.isLast()) {
					page.setPageNumber(page.getPageNumber() + 1);
					result = Db.use(jdbcTemplate.getDataSource()).page(Entity.create(cacheKeyEnum.table), page);
					currentSize += result.size();
					
					pipelineToRedis(result, cacheKeyEnum, result.getTotal(), currentSize);
				}
			} catch (SQLException e) {
				log.error("[cache]xxx平台刷新缓存失败：" + e.getMessage());
			}
		});

	}

	private void pipelineToRedis(PageResult<Entity> result, CacheKeyEnum cacheKeyEnum, int total, int currentSize) {
		redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
			for (Entity entity : result) {
				String key = String.valueOf(entity.get(cacheKeyEnum.tablePrimaryKey));
				Object value = JSONObject.toJSON(entity);
				connection.hSet(cacheKeyEnum.key.getBytes(StandardCharsets.UTF_8), key.getBytes(StandardCharsets.UTF_8), value.toString().getBytes(StandardCharsets.UTF_8));
			}
			log.info("[cache]完成加载xxx平台缓存-{}，{}/{}", cacheKeyEnum.table, total, currentSize);
			return null;
		});
	}

	@Override
	public void clearCache(String key) {
		redisTemplate.opsForHash().getOperations().delete(key);
		log.info("[cache]清除xxx平台缓存，key={}", key);
	}
}

用法

1、在相应动作点加注解：

在服务层动作增加注解，即可完成动作后立即刷新某个/某些缓存项。

@RefreshCacheAnn({CacheKeyEnum.T_TRANSFER_AUTHORIZE, CacheKeyEnum.T_TRANSFER_APPLY})
	@Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
	public void doSmt(CensorRequest censorRequest) {


}

2、查询时用法

// get
ApiModelDO apiInCache = Optional.ofNullable(cache.get(CacheKeyEnum.T_API, path, ApiModelDO.class, () -> bdcpClient.queryOne("select * from  t_api where api_url_postfix = ?", new Object[]{path}, ApiModelDO.class))).orElseThrow(() -> new ApiException("接口不存在或未发布", SYSTEM_EXCEPTION));


// getAll
List<DynamicMaskingUser> dynamicMaskingUsers = cache.getAll(CacheKeyEnum.T_DYNAMIC_MASKING_USER, DynamicMaskingUser.class);


// getListByCondition
List<DesensitizationAlgorithm> algorithmListInCache = cache.getListByCondition(CacheKeyEnum.T_DESENST_ALGORITHM, DesensitizationAlgorithm.class, algorithm -> maskingAlgorithmIdList.contains(algorithm.getId()));


//  getObjectByCondition
Optional<TransferAuthorize> transferAuthorizeOpt = Optional.ofNullable(cache.getObjectByCondition(CacheKeyEnum.T_TRANSFER_AUTHORIZE, TransferAuthorize.class, authorize -> appKey.equals(authorize.getAppKey()), () -> bdcpClient.queryOne("select * from t_transfer_authorize where app_key = ?", new Object[]{appKey}, TransferAuthorize.class)));

用到的工具类

@Component
public class RedisUtils {

	@Resource
	private RedisTemplate<String, Object> redisTemplate;

	/**
	 * 开启Redis事务
	 */
	public void multi() {
		redisTemplate.multi();
	}

	/**
	 * 执行Redis事务并提交
	 */
	public void exec() {
		redisTemplate.exec();
	}


	/**
	 * 实现命令：KEYS pattern，查找所有符合给定模式 pattern的 key
	 *
	 * @param pattern the pattern
	 * @return the set
	 */
	public Set<String> keys(String pattern) {
		return redisTemplate.keys(pattern);
	}

	/**
	 * 实现命令：DEL key，删除一个key
	 *
	 * @param key the key
	 */
	public void del(String key) {
		redisTemplate.delete(key);
	}

	/**
	 * 实现命令：SET key value，设置一个key-value（将字符串值 value关联到 key）
	 *
	 * @param key   the key
	 * @param value the value
	 */
	public void set(String key, Object value) {
		redisTemplate.opsForValue().set(key, value);
	}

	/**
	 * 实现命令：SET key value EX seconds，设置key-value和超时时间（秒）
	 *
	 * @param key     the key
	 * @param value   the value
	 * @param timeout （以秒为单位）
	 */
	public void set(String key, Object value, long timeout) {
		redisTemplate.opsForValue().set(key, value, timeout, TimeUnit.SECONDS);
	}

	/**
	 * 实现命令：HSET key field value，将哈希表 key中的字段 field的值设为 value
	 *
	 * @param key   the key
	 * @param field the field
	 * @param value the value
	 */
	public void hSet(String key, String field, Object value) {
		redisTemplate.opsForHash().put(key, field, value);
	}

	/**
	 * 实现命令：GET key，返回 key所关联的字符串值。
	 *
	 * @param key the key
	 * @return value string
	 */
	public Object get(String key) {
		return redisTemplate.opsForValue().get(key);
	}


	/**
	 * Hget string.
	 *
	 * @param key   the key
	 * @param field the field
	 * @return the string
	 */
	public String hgetPipleline(String key, String field) {
		byte[] redisKey = key.getBytes(StandardCharsets.UTF_8);
		byte[] redisField = field.getBytes(StandardCharsets.UTF_8);
		Object value = redisTemplate.execute(connection -> {
			byte[] result = connection.hGet(redisKey, redisField);
			return (result != null) ? new String(result, StandardCharsets.UTF_8) : null;
		}, true);
		return (value != null) ? value.toString() : null;
	}

	/**
	 * Retrieve all fields and their corresponding values from a Redis hash given a key.
	 *
	 * @param key the key
	 * @return the map of fields and values
	 */
	public Map<String, String> hgetBatch(String key) {
		byte[] redisKey = key.getBytes(StandardCharsets.UTF_8);

		return redisTemplate.execute((RedisCallback<Map<String, String>>) connection -> {
			Map<byte[], byte[]> results = connection.hGetAll(redisKey);
			if (results != null) {
				Map<String, String> resultMap = new HashMap<>();
				for (Map.Entry<byte[], byte[]> entry : results.entrySet()) {
					String field = new String(entry.getKey(), StandardCharsets.UTF_8);
					String value = new String(entry.getValue(), StandardCharsets.UTF_8);
					resultMap.put(field, value);
				}
				return resultMap;
			}
			return null;
		});
	}

	/**
	 * Hget one string.
	 *
	 * @param key   the key
	 * @param field the field
	 * @return the string
	 */
	public String hgetOne(String key, String field) {
		byte[] redisKey = key.getBytes(StandardCharsets.UTF_8);
		byte[] redisField = field.getBytes(StandardCharsets.UTF_8);

		return redisTemplate.execute((RedisCallback<String>) connection -> {
			byte[] result = connection.hGet(redisKey, redisField);
			return (result != null) ? new String(result, StandardCharsets.UTF_8) : null;
		});
	}


	/**
	 * 获取hash列表
	 *
	 * @param key   键
	 * @param clazz 类型
	 * @return {@link List<T>}
	 * @author chentl
	 * @version v1.0.0
	 * @since 11:50 AM 2023/6/30
	 **/
	public <T> List<T> hgetListByClass(String key, Class<T> clazz) {
		Map<String, String> cacheMap = hgetBatch(key);
		return cacheMap.values()
			.stream()
			.map(tag -> JSONObject.parseObject(tag, clazz))
			.collect(Collectors.toList());
	}

	/**
	 * 获取哈希表中指定字段对应的单个对象
	 *
	 * @param key   键
	 * @param field 字段名
	 * @param clazz 类型
	 * @return 指定字段对应的对象
	 */
	public <T> T hgetOneByClass(String key, String field, Class<T> clazz) {
		Map<String, String> cacheMap = hgetBatch(key);
		//返回指定字段对应的对象
		return JSONObject.parseObject(cacheMap.get(field), clazz);
	}


	/**
	 * 获取有序集合中指定排名范围内的元素以及对应的分数（分数降序）
	 *
	 * @param key   the key of the sorted set
	 * @param start the start index of the range
	 * @param end   the end index of the range
	 * @return the set of elements with scores
	 */
	public Set<ZSetOperations.TypedTuple<Object>> zrevrangeWithScores(String key, long start, long end) {
		return redisTemplate.opsForZSet().reverseRangeWithScores(key, start, end);
	}

	/**
	 * 将一个或多个成员元素及其分数值加入到有序集合中
	 *
	 * @param key    the key of the sorted set
	 * @param score  the score of the member element
	 * @param member the member element to add
	 * @return the number of elements added to the sorted set, not including all the elements already present in the set
	 */
	public Double zadd(String key, double score, String member) {
		Double currentScore = redisTemplate.opsForZSet().score(key, member);
		if (currentScore != null) {
			// 成员已存在，累加分数
			return redisTemplate.opsForZSet().incrementScore(key, member, score);
		} else {
			// 成员不存在，直接添加
			Boolean added = redisTemplate.opsForZSet().add(key, member, score);
			return added ? score : null;
		}
	}

	/**
	 * 将一个或多个成员元素及其分数值加入到有序集合中，并设置过期时间
	 *
	 * @param key        the key of the sorted set
	 * @param score      the score of the member element
	 * @param member     the member element to add
	 * @param expireTime the expiration time in seconds
	 * @return the number of elements added to the sorted set, not including all the elements already present in the set
	 */
	public Double zaddWithExpire(String key, double score, String member, long expireTime) {
		Double currentScore = redisTemplate.opsForZSet().score(key, member);
		if (currentScore != null) {
			// 成员已存在，累加分数
			Double newScore = redisTemplate.opsForZSet().incrementScore(key, member, score);
			if (expireTime > 0) {
				redisTemplate.expire(key, expireTime, TimeUnit.SECONDS);
			}
			return newScore;
		} else {
			// 成员不存在，直接添加
			Boolean added = redisTemplate.opsForZSet().add(key, member, score);
			if (added && expireTime > 0) {
				redisTemplate.expire(key, expireTime, TimeUnit.SECONDS);
			}
			return added ? score : null;
		}
	}


	/**
	 * 根据排名范围删除有序集合中的元素
	 *
	 * @param key   the key of the sorted set
	 * @param start the start index of the range
	 * @param end   the end index of the range
	 * @return the number of elements removed
	 */
	public long zremrangeByRank(String key, long start, long end) {
		return redisTemplate.opsForZSet().removeRange(key, start, end);
	}

	/**
	 * 获取有序集合中指定成员的分数
	 *
	 * @param key    键
	 * @param member 成员
	 * @return 分数
	 */
	public Double zscore(String key, String member) {
		return redisTemplate.opsForZSet().score(key, member);
	}

	/**
	 * 对有序集合成员的分数进行原子性增减操作
	 *
	 * @param key    键
	 * @param member 成员
	 * @param delta  增减值
	 * @return 变更后的分数
	 */
	public Double zincrby(String key, String member, double delta) {
		return redisTemplate.opsForZSet().incrementScore(key, member, delta);
	}


	/**
	 * 执行Lua脚本
	 *
	 * @param luaScript lua 脚本
	 * @param keys      键集合
	 * @param args      参数集合
	 * @return
	 */
	public String eval(String luaScript, List<String> keys, List<String> args) {
		return redisTemplate.execute((RedisCallback<String>) connection -> {
			try {
				byte[] luaScriptBytes = luaScript.getBytes(StandardCharsets.UTF_8);
				int keyCount = keys.size();
				List<byte[]> argList = new ArrayList<>(keyCount + args.size());
				for (String key : keys) {
					argList.add(key.getBytes(StandardCharsets.UTF_8));
				}
				for (String arg : args) {
					argList.add(arg.getBytes(StandardCharsets.UTF_8));
				}
				byte[][] argArray = argList.toArray(new byte[argList.size()][]);
				String result = connection.eval(luaScriptBytes, ReturnType.VALUE, keyCount, argArray);
				return result;
			} catch (Exception ex) {
				throw new RuntimeException("Failed to execute Lua script", ex);
			}
		});
	}