使用Google Guava Cache开发本地缓存

需求

有些频繁访问，更新不太频繁的io密集型业务数据可以存在缓存中，减少业务数据库io压力，由于项目开发中没有redis可用，需要自己开发一个本地cache。

实现代码

@Slf4j
public class LocalCache<K, V> {

    private final Cache<K, TimeValue<V>> lruCache;

    private final ExecutorService threadPool;

    /**
     * 当缓存失效，多个线程并发更新缓存时，用于存储更新当前K并发更新任务，防止缓存击穿问题
     */
    private final Map<K, Future<TimeValue<V>>> runningTasks = Maps.newConcurrentMap();

    public LocalCache(int initialCapacity, long maximumSize, Duration expireTime, ExecutorService threadPool) {
        this.threadPool = threadPool;
        lruCache =
                CacheBuilder.newBuilder()
                        .expireAfterWrite(expireTime)
                        .initialCapacity(initialCapacity)
                        .maximumSize(maximumSize)
                        .build();
    }

    /**
     * 获取缓存，过期时间为时间点LocalDateTime
     * @param key  缓存key
     * @param expireTime  过期时间点
     * @param maxWait  最大等待时间
     * @param function  获取缓存的函数
     * @return V
     */
    public V get(K key, LocalDateTime expireTime, Duration maxWait,
                 Function<? super K, ? extends V> function) {

        Objects.requireNonNull(expireTime);
        Objects.requireNonNull(maxWait);
        Objects.requireNonNull(function);

        return Optional.ofNullable(key)
                .map(ThrowingFunction.handleException(
                    k -> {
                        TimeValue<V> cacheValue = lruCache.get(k, () -> refresh(k, function).get());

                        Duration existDuration =
                                Duration.between(cacheValue.getCreateTime(), LocalDateTime.now());
                        Duration expireDuration =
                                Duration.between(cacheValue.getCreateTime(), expireTime);

                        return getV(maxWait, function, k, cacheValue, existDuration, expireDuration);
                    },
                    ex -> {
                        log.info("Get key error {}", key, ex);
                        return null;
                    }))
                .orElse(null);
    }

    /**
     * 获取缓存，过期时间为时间段Duration
     * @param key  缓存key
     * @param expireDuration  过期时间段
     * @param maxWait  最大等待时间
     * @param function  获取缓存的函数
     * @return V
     */
    public V get(K key, Duration expireDuration, Duration maxWait,
                 Function<? super K, ? extends V> function) {

        Objects.requireNonNull(expireDuration);
        Objects.requireNonNull(maxWait);
        Objects.requireNonNull(function);

        return Optional.ofNullable(key)
                .map(
                        ThrowingFunction.handleException(
                            k -> {
                                TimeValue<V> cacheValue = lruCache.get(k, () -> refresh(k, function).get());

                                Duration existDuration =
                                        Duration.between(cacheValue.getCreateTime(), LocalDateTime.now());

                                return getV(maxWait, function, k, cacheValue, existDuration, expireDuration);
                            },
                            ex -> {
                                log.info("Get key error {}", key, ex);
                                return null;
                            }))
                .orElse(null);
    }

    private V getV(Duration maxWait, Function<? super K, ? extends V> function,
                   K k, TimeValue<V> cacheValue, Duration existDuration, Duration expireDuration) {
        if (existDuration.toMillis() < expireDuration.toMillis()) {
            log.debug("Key get result {}", k);
            return cacheValue.getValue();
        }

        try {
            TimeValue<V> newValue =
                    loadWithTimeout(k, maxWait.toMillis(), function);
            log.info(
                    "Key expired, get from db {} {} {}",
                    k,
                    cacheValue.getCreateTime(),
                    existDuration);
            return newValue.getValue();
        } catch (TimeoutException e) {
            log.info("Query timeout ,return old value {}", k);
            return cacheValue.getValue();
        }
    }

    /**
     * 获取缓存创建时间
     * @param key
     * @return LocalDateTime or Null
     */
    public LocalDateTime getCreateTimeIfPresent(K key) {
        return Optional
                .ofNullable(lruCache.getIfPresent(key))
                .map(TimeValue::getCreateTime)
                .orElse(null);
    }

    /**
     * 创建一个刷新key的任务，最多等待maxWait时间，超时之后抛出TimeoutException,
     * 但是刷新任务后台仍继续跑
     */
    protected TimeValue<V> loadWithTimeout(K key, long maxWait, Function<? super K, ? extends V> function)
            throws TimeoutException {
        Future<TimeValue<V>> future = refresh(key, function);
        try {
            return future.get(maxWait, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            log.info("Key get from db InterruptedException {} {}", key, maxWait, e);
            throw new TimeoutException();
        } catch (ExecutionException e) {
            log.info("Key get from db ExecutionException {} {}", key, maxWait, e);
            throw new TimeoutException();
        } catch (TimeoutException e) {
            log.info("Key get from db timeout {} {}", key, maxWait, e);
            throw new TimeoutException();
        }
    }

    /**
     * 创建刷新key的任务，立即返回Future
     * 在请求后台结束之后，会进行两个操作， 1，更新LRU Cache， 2. 清除RunningTasks
     * 避免大量并发请求导致的缓存击穿问题
     */
    protected Future<TimeValue<V>> refresh(K key, Function<? super K, ? extends V> function) {
        return runningTasks.computeIfAbsent(key, k ->
                threadPool.submit(
                    () -> {
                        try {
                            V result = function.apply(k);
                            TimeValue<V> value = new TimeValue<>(result);
                            lruCache.put(k, value);
                            return value;
                        } finally {
                            runningTasks.remove(k);
                        }
                    }));
    }
}

public class TimeValue<V> {

    private LocalDateTime createTime;
    private V value;

    public TimeValue(V value) {
        this.createTime = LocalDateTime.now();
        this.value = value;
    }

    public LocalDateTime getCreateTime() {
        return createTime;
    }

    public V getValue() {
        return value;
    }
}

@FunctionalInterface
public interface ThrowingFunction<T, R, E extends Exception> {

    /**
     *
     * @param t
     * @return
     * @throws E
     */
    R apply(T t) throws E;

    /**
     * 包装Function函数，在lambda表达式中处理异常
     * @param function
     * @param handler
     * @param <T>
     * @param <R>
     * @param <E>
     * @return Function
     */
    static <T, R, E extends Exception> Function<T, R> handleException(ThrowingFunction<T, R, E> function,
                                                                             Function<Exception, R> handler) {
        Objects.requireNonNull(function);

        return t -> {
            try {
                return function.apply(t);
            } catch (Exception e) {
                return handler.apply(e);
            }
        };
    }
}