redis生产使用场景（一）：并行流+二级缓存

本文主要介绍 redis 缓存在线上的使用场景

由于业务的特殊性，在生产库用户表中，大概有 50 多万的测试用户，在真实业务计算中，要把测试用户给筛选掉，所以在计算前，需要把测试用户加载到 redis 缓存中，并构建本地缓存。

查询数据库

为了提高查询效率，先得到最大主键ID，以每次查询5000 条提前构建好批次范围，并采用并行查询的方式。

     long maxUserId = odsUserDAO.getMaxUserId(); // 查询最大 id// 构建批次范围列表List<LinkedList<Long>> pagingIntervalList = new ArrayList<>();//初始化分页列表for(long leftBoundary = 0; leftBoundary < maxUserId; leftBoundary += 5000){LinkedList<Long> pagingInterval = new LinkedList<>();long rightBoundary = leftBoundary + 5000;if(rightBoundary >= maxUserId){pagingInterval.add(leftBoundary);pagingInterval.add(maxUserId + 1);pagingIntervalList.add(pagingInterval);break;}pagingInterval.add(leftBoundary);pagingInterval.add(rightBoundary);pagingIntervalList.add(pagingInterval);}//并行加载测试学生Set<Long> testUserIdSet = pagingIntervalList.parallelStream().map(pagingInterval -> odsUserDAO.incrementGetTestUser(pagingInterval.peekFirst(), pagingInterval.peekLast())).flatMap(List::stream).map(o -> o.getId()).collect(Collectors.toSet());

分页查询 sql

<select id="incrementGetTestUser" resultMap="BaseResultMap">select  <include refid="Base_Column_List"/>from userwhere (account_type ='TEST'or name like '%test%'or name like '%TEST%'or name like '%测试%'or status <![CDATA[ <> ]]> 'NORMAL')and id between #{pagingStartIdx} and #{pagingEndIdx}and status = 1</select>

写回缓存

private static volatile Set<Long> testUserLocalCache = null; //全局变量testUserLocalCache = testUserIdSet; //本地缓存
String[] testUserIdArray = testUserIdSet.parallelStream().map(String::valueOf).toArray(String[]::new);
redisCluster.sadd("test_user_data", testUserIdArray); //redis缓存

查询缓存

在处理业务时，如果该用户是测试数据，则跳过。

private static volatile Set<Long> testUserLocalCache = null; //全局变量public boolean isTestUser(Long userId){if (testUserLocalCache == null || testUserLocalCache.isEmpty()){boolean isRemoteTestUserCacheExists = redisCluster.exists("test_user_data");if(!isRemoteTestUserCacheExists){throw new RuntimeException("failed to hit test user remote cache, cause test user never preload");}// 如果本地缓存为空，则构建synchronized(this){if (testUserLocalCache == null || testUserLocalCache.isEmpty()){Set<String> testUserSet = redisCluster.sscanAll("test_user_data"); //采用 sscan 游标方式查询testUserLocalCache = testUserSet.parallelStream().map(Long::valueOf).collect(Collectors.toSet());}}}//从本地缓存查询数据return testUserLocalCache.contains(userId);}

总结

由于业务性质及生产机器配置够高，缓存不设置过期时间，只是在下次计算前，先清除缓存即可，再构建本次新的缓存。