redis cache抽象实现

概述

redis缓存在项目中经常用到,如果每个需要的地方独立实现,会有各种各样的问题,如实现的功能不健壮,代码冗余

本篇的宗旨是构建一个抽象的,功能完善,代码健壮的缓存模块,使得接入时,通够快速、便捷的使用缓存

Cache

redis中缓存的对象

/**
 * @author liuk
 * 封装此对象的目的
 * 1. 当从redis批量获取对象时,能够清楚获取到的value对应的key
 * 当然redis返回时,是与请求时的key的顺序是一致的,如果未取到会返回null,通过顺序的一致性也是可以确定获取到的value对应的key的
 * 2. 如果该key确实不存在对应的对象,防止一直透过缓存查数据库,故该场景保存的value=null
 */
@Data
public class Cache<K, V> {

    public Cache() {
    }
    public Cache(K key, V value) {
        this.key = key;
        this.value = value;
    }
    /**
     * 缓存key
     */
    private K key;
    /**
     * 缓存value
     */
    private V value;
}

RedisKeyEnum

redis key的枚举,包括key的构造及expire

public enum RedisKeyEnum {

    APPLY_BATCH_EXEC_LOCK_KEY("xx:xx:exec_lock_key:%s", 10)

    RedisKeyEnum(String key, int expire) {
        this.key = key;
        this.expire = expire;
    }

    private String key;
    /**
     * second
     */
    private int expire;

    public String getKey(Object... args) {
        return String.format(key, args);
    }

    public int getExpire() {
        return expire;
    }
}

RedisCache

cache的接口,不止是redis cache 所有的cache基本都是以下方法的实现

public interface RedisCache<K, V> {

    /**
     * 获取缓存
     *
     * @param code
     * @return
     */
    V getByCache(K code);


    /**
     * 批量获取缓存
     *
     * @param codes
     * @return
     */
    Map<K, V> getByCache(Set<K> codes);

    /**
     * 删除缓存
     *
     * @param code
     */
    void delCache(K code);
}

AbstractRedisCache

抽象redis cache实现


/**
 * @author liuk
 */
public abstract class AbstractRedisCache<K, V> implements RedisCache<K, V> {
    private static final ILog LOGGER = LogFactory.getLog(AbstractRedisCache.class);

    /**
     * 需要注入自己的的redis commands 工具
     */
    @Resource
    private JedisCommands jedisCommands;
    /**
     * 需要注入自己的的redis commands 批量获取key value的工具
     */
    @Resource
    private MultiKeyCommands multiKeyCommands;

    /**
     * 获取缓存
     *
     * @param code
     * @return
     */
    //todo key 抽象   解决多参数问题 以后再改造吧
    public V getByCache(K code) {
        if(code == null){
            return null;
        }
        RedisKeyEnum keyEnum = keyEnum();
        String key = keyEnum.getKey(code);
        String cacheStr = this.get(code);
        if (!StringUtils.isEmpty(cacheStr)) {
            Cache<K, V> cache = map2cache(cacheStr);
            return cache.getValue();
        }
        //get
        LOGGER.info("redis_cache||createCache||code={}", code);
        V value = createCache(code);
        this.set(code, value);
        return value;
    }

    private Cache<K, V> map2cache(String cacheStr) {
        Cache<K, V> cache = JSONObject.parseObject(cacheStr, Cache.class);
        V v = JSONObject.parseObject(cacheStr).getObject("value", valueClass());
        cache.setValue(v);
        return cache;
    }

    /**
     * 批量获取缓存
     *
     * @param codes
     * @return
     */
    public Map<K, V> getByCache(Set<K> codes) {
        
        if(CollectionUtils.isEmpty(codes)){
            return new HashMap();
        }
        RedisKeyEnum keyEnum = keyEnum();
        Map<K, V> result = new HashMap<>();
        int max = 20;
        Set<K> noCacheGoodsIds = new HashSet<>(codes);
        Map<String, K> codesMap = map2String(codes);
        for (int i = 0; i < codes.size(); i += max) {
            List<K> tempCodes = new ArrayList<>(codes).subList(i, Math.min(i + max, codes.size()));
            //redis 操作弱依赖
            List<String> values = this.mget(tempCodes);
            if (org.springframework.util.CollectionUtils.isEmpty(values)) {
                continue;
            }
            for (String value : values) {
                //竟然拿到的是null.....
                if (StringUtils.isEmpty(value)) {
                    continue;
                }
                Cache<K, V> cache = map2cache(value);
                if (cache.getValue() != null) {
                    result.put(codesMap.get(String.valueOf(cache.getKey())), cache.getValue());
                    noCacheGoodsIds.remove(codesMap.get(String.valueOf(cache.getKey())));
                }
                noCacheGoodsIds.remove(codesMap.get(String.valueOf(cache.getKey())));

            }
        }
        if (!CollectionUtils.isEmpty(noCacheGoodsIds)) {
            LOGGER.info("redis_cache||createCache||size={}", noCacheGoodsIds.size());
            noCacheGoodsIds = noCacheGoodsIds.stream().filter(Objects::nonNull).collect(Collectors.toSet());
            Map<K, V> values = createCache(noCacheGoodsIds);
            this.set(values);
            result.putAll(values);
        }
        return result;
    }

    /**
     * 异步 and 弱依赖
     *
     * @param cacheMap
     */
    private void set(Map<K, V> cacheMap) {
        ThreadPoolExecutorEnum.BATCH_EXEC_THREAD_POOL.execute(() -> {
            try {
                RedisKeyEnum keyEnum = keyEnum();
                for (K key : cacheMap.keySet()) {
                    Cache<K, V> cache = new Cache<>(key, cacheMap.get(key));
                    jedisCommands.setex(keyEnum.getKey(key), keyEnum.getExpire(), JSONObject.toJSONString(cache));
                }
            } catch (Exception e) {
                LOGGER.error("redis_cache||set||cacheMap.size={}", cacheMap.size());
            }
        });

    }

    /**
     * redis弱依赖,超时或故障不影响业务
     * todo redis sentinal熔断
     *
     * @param keyList
     * @return
     */
    private List<String> mget(List<K> keyList) {
        try {
            RedisKeyEnum keyEnum = keyEnum();
            String[] keys = new String[keyList.size()];
            for (int j = 0; j < keyList.size(); j++) {
                keys[j] = keyEnum.getKey(keyList.get(j));
            }
            List<String> values = multiKeyCommands.mget(keys);
            return values;
        } catch (Exception e) {
            LOGGER.error("redis_cache||mget||keys.size={}", keyList.size());
        }
        return new ArrayList<>();
    }

    /**
     * redis弱依赖,超时或故障不影响业务
     * 当redis超时时,直接调用createCache  是仍然能获取到value的,但是数据库压力会比较大,需考虑是否可承压,可通过开关控制是否可直接调用 createValue
     *
     * @param keyList
     * @return
     */
    private String get(K key) {
        try {
            RedisKeyEnum keyEnum = keyEnum();
            String cacheStr = jedisCommands.get(keyEnum.getKey(key));
            return cacheStr;
        } catch (Exception e) {
            LOGGER.error("redis_cache||get||key={}", key);
        }
        return null;
    }

    /**
     * redis弱依赖,当redis超时或故障时不影响业务
     * 此处如有需要可考虑异步 参考 delCache
     * @param keyList
     * @return
     */
    private void set(K key, V value) {
        try {
            RedisKeyEnum keyEnum = keyEnum();
            Cache<K, V> cache = new Cache<>(key, value);
            jedisCommands.setex(keyEnum.getKey(key), keyEnum.getExpire(), JSONObject.toJSONString(cache));
        } catch (Exception e) {
            LOGGER.error("redis_cache||set||key={}", key);
        }
    }

    /**
     * 映射code,进行K与string的映射,否则当key是Long时,可能会映射为Integer而非Long
     * 故通过此方法,记录key的原值
     *
     * @param codes
     * @return
     */
    private Map<String, K> map2String(Set<K> codes) {
        Map<String, K> result = new HashMap<>();
        for (K code : codes) {
            result.put(String.valueOf(code), code);
        }
        return result;
    }

    /**
     * 删除缓存
     * 弱依赖,删除失败不影响现有逻辑,等缓存超时,但是也需要关注删除失败的场景,如果出现,则后续缓存失效前使用的都是老数据,根据数据及时性,做不同的逻辑
     *
     * @param code
     */
    public void delCache(K code) {
        try {

            RedisKeyEnum keyEnum = keyEnum();
            String key = keyEnum.getKey(code);

            //有些是在事务中的操作,为了尽快释放事务,异步删除
            ThreadPoolExecutorEnum.REDIS_CACHE_THREAD_POOL.execute(() -> {
                jedisCommands.del(key);
            });

            //延时双删,防止主备延迟,导致上面删除后,仍然读到旧的数据,此处延时双删下,如不需要可删除
            ScheduledExecutor.run(() -> {
                jedisCommands.del(key);
            }, 1);
        } catch (Exception e) {
            LOGGER.error("redis_cache||delCache||key={}", code);
        }

    }

    /**
     * 缓存不存在时单个生成缓存对象
     *
     * @param key
     * @return
     */
    private V createCache(K key) {
        Map<K, V> map = createCache(Sets.newHashSet(key));
        if (map != null && map.containsKey(key)) {
            return map.get(key);
        }
        return null;
    }

    /**
     * 缓存不存在时,用于批量生成缓存对象
     *
     * @param key
     * @return
     */
    protected abstract Map<K, V> createCache(Set<K> key);

    /**
     * 缓存key枚举  存在key及expire
     *
     * @return
     */
    protected abstract RedisKeyEnum keyEnum();

    /**
     * 缓存class类型,用于将字符串map为V 对象,因为泛型在编译后会被擦除,所以通过泛型无法正常映射
     *
     * @return
     */
    protected abstract Class<V> valueClass();
}


ThreadPoolExecutorEnum

异常处理线程池

public enum ThreadPoolExecutorEnum {

    BATCH_EXEC_THREAD_POOL(2, 5, 60, 100, "batch_exec_thread_pool", new ThreadPoolExecutor.AbortPolicy()),
    REDIS_CACHE_THREAD_POOL(2, 5, 60, 100, "cache_batch_set_thread_pool", new ThreadPoolExecutor.DiscardPolicy()),
    ;

    private TaxiThreadPoolExecutor executor;

    ThreadPoolExecutorEnum(int corePoolSize, int maxPoolSize, int keepAliveTime, int queueSize, String name, RejectedExecutionHandler handler) {
        //XXXThreadPoolExecutor 保证trace等数据完整
        this.executor = new XXXThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(queueSize), new ThreadFactoryBuilder(name), handler);
    }


    public void execute(Runnable runnable) {
        executor.execute(runnable);
    }
}

延时处理器

用于进行延时处理

public class ScheduledExecutor {

    /**
     *  延时处理器
     */
    private static ScheduledExecutorService executorService = new TaxiScheduledThreadPoolExecutorWrapper(2,
            new ThreadFactoryBuilder("scheduled_executor"));

    /**
     * 延时执行
     *
     * @param runnable
     * @param second   秒
     */
    public static void run(Runnable runnable, int second) {
        //ScheduledExecutorService 用于延时,为了防止阻塞其它的延时处理逻辑,延时后通过新的线程池执行
        Runnable threadPoolRunnable = () -> {
            ThreadPoolExecutorEnum.REDIS_CACHE_THREAD_POOL.execute(runnable);
        };
        executorService.schedule(threadPoolRunnable, second, TimeUnit.SECONDS);
    }

}