[大数据]关于缓存细粒度化后的查询优化

?高性能MYSQL中有这部分的描述。此处主要记录下细粒度后结合持久型DB查询的实现问题。

场景为普通的对象查询，定义一个接口方法，目的是查询id列表对应的映射数据，key为id（唯一标识即可），value为对应值对象

Map<Long, DemoDTO> list(Set<Long> idSet);

?定义好一个通用方法

/**
 * 结合MYSQL的管道查询
 *
 * @param targetSet         目标集合
 * @param dbResultMapGetter 未命中缓存时，从MYSQL获取数据映射
 * @param cacheKeyPrefix    缓存前缀
 * @param objectNull        避免缓存穿透时null值默认的空对象
 * @param <K>               map result key
 * @param <V>               map result value
 * @return 最终查询结果映射
 */
<K, V> Map<K, V> pGetCombineDb(Set<K> targetSet, DbResultMapGetter<K, V>    dbResultMapGetter, String cacheKeyPrefix, V objectNull);

?其中targetSet即为查询的id列表，DbResultMapGetter 定义为从MYSQL等持久化库中获取的未命中缓存的对象映射数据接口，cacheKeyPrefix为定义的缓存前缀，objectNull为避免缓存穿透而缓存的空对象。具体实现代码参考如下：

public interface DbResultMapGetter<K, V> {

    Map<K,V> getDbResult(Set<K> dbSearching);
}

    @SuppressWarnings("unchecked")
    @Override
    public <K, V> Map<K, V> pGetCombineDb(Set<K> targetSet, DbResultMapGetter<K, V> dbResultMapGetter, String cacheKeyPrefix, V objectNull) {
        List<String> keys = targetSet.stream().map(e ->
                RedisUtils.mergeKey(cacheKeyPrefix, e))
                .collect(Collectors.toList());
        List<V> cacheResult = (List<V>) pGet(keys);
        Map<K, V> resultMap = new HashMap<>(targetSet.size());
        Map<K, V> dbResultMap = new HashMap<>();
        boolean allHitCache = putCacheResult(new ArrayList<>(targetSet), cacheResult, resultMap, dbResultMap);
        if (allHitCache) {
            return resultMap;
        }
        // 更新真实的Mysql查询结果
        dbResultMap.putAll(dbResultMapGetter.getDbResult(dbResultMap.keySet()));
        updateMap(resultMap, dbResultMap, objectNull);
        // 缓存mysql查询结果
        List<Tuple2<String, Object>> setCommands = new ArrayList<>(dbResultMap.size());
        dbResultMap.forEach((k, v) -> setCommands.add(new Tuple2<>(
                // 缓存参数拼接策略，一般以:隔开
                RedisUtils.mergeKey(cacheKeyPrefix, k), v)));
        pSet(setCommands);
        return resultMap;
    }

    /**
     * @param targetSet   查询目前列表
     * @param cacheResult 管道查询缓存结果有序列表
     * @param resultMap   最终查询结果
     * @param dbResultMap mysql查询结果
     * @param <K>         map result key
     * @param <V>         map result value
     * @return 是否全部命中缓存
     */
    private <K, V> boolean putCacheResult(List<K> targetSet, List<V> cacheResult, Map<K, V> resultMap,
                                          Map<K, V> dbResultMap) {
        for (int i = 0; i < targetSet.size(); i++) {
            if (cacheResult.get(i) == null) {
                // mysql查询的各个结果先视为空
                dbResultMap.put(targetSet.get(i), null);
            } else {
                resultMap.put(targetSet.get(i), cacheResult.get(i));
            }
        }
        return dbResultMap.keySet().size() == 0;
    }

    /**
     * 更新查询结果
     *
     * @param resultMap   最终查询结果
     * @param dbResultMap mysql查询结果
     * @param objectNull  避免缓存穿透时null值默认的空对象
     * @param <K>         map result key
     * @param <V>         map result value
     */
    private <K, V> void updateMap(Map<K, V> resultMap, Map<K, V> dbResultMap, V objectNull) {
        // 避免缓存穿透
        dbResultMap.replaceAll((k, v) -> {
            if (v == null) {
                return objectNull;
            }
            return v;
        });
        // 更新最终查询结果
        resultMap.putAll(dbResultMap);
    }

?这边以redis的pipeline为例，也可用mGet方式

    public List<Object> pGet(List<String> keys) {
        return stringRedisTemplate.executePipelined((RedisCallback<Object>) redisConnection -> {
            for (String key : keys) {
                redisConnection.get(serializeKey(key));
            }
            return null;
        }, RedisSerializer.json());
    }


    public void pSet(List<Tuple2<String, Object>> kvs) {
        stringRedisTemplate.executePipelined((RedisCallback<Object>) redisConnection -> {
            for (Tuple2<String, Object> kv : kvs) {
                byte[] value = RedisSerializer.json().serialize(kv.getSecond());
                if (value == null) {
                    continue;
                }
                redisConnection.set(serializeKey(getRealKey(kv.getFirst())), value,
                        Expiration.milliseconds(24 * 3600 * 1000L +
                                // 避免同一时间过期较多缓存
                                RandomUtils.nextLong(1, 6 * 3600 * 1000L)),
                        RedisStringCommands.SetOption.UPSERT);
            }
            return null;
        }, RedisSerializer.json());
    }

具体到业务查询时就可以这么使用：

    @Override
    public Map<Long, DemoDTO> list(Set<Long> idSet) {
        return redisExtendService.pGetCombineDb(idSet, dbSearching ->         {
            // 用IN查询之类的把未命中缓存的数据查询出来
            List<DemoDTO> dbResult = demoDao.findAll(dbSearching);
            return dbResult.stream().collect(Collectors.toMap(DemoDTO::getId,demo-> demo));
        }, "demo", new DemoDTO());
    }

?其中，具体的mysql业务查询、缓存key前缀、缓存空对象等就可以似具体情况自行编写了。

这边提供一个思路，有更优雅的实现可以分享下哈。