Yilena
境界の備忘録
这里是一处被时间遗忘的避难所。
浮空之岛上，堆叠着理性构筑的文字、应对命运审判的答卷，还有那些在午夜寂静时、从指缝间溜走的虚幻灵感。
终究是要展翅远航的吧。在你的羽翼尚未丰满前，愿这些微弱的火光，能稍微温暖你疲惫的旅程。
文章标签
1321 字
3 分钟
116秒→6秒：Redis管道+批处理优化用户好友关系校验的方案
目录
一、业务需求
在社交平台中，当用户互相关注时需建立好友关系并写入好友表。
但在极端情况下（如插入失败），可能导致好友关系数据不一致。
需一个定时任务，检查并修复所有用户的好友关系数据。
已知用户量约 5 万。
二、分析
核心逻辑：对每个用户，计算其关注对象与粉丝对象的交集（即应存在的好友关系），与好友表现有数据进行对比并更新。
关键在于优化任务耗时，应对 5 万量级甚至未来持续增长的数据。
（一）初版
初版方案如上所示，核心逻辑是逐个遍历用户，然后遍历的过程中取交集，检查与好友表数据是否一致，不一致则删除好友表对应的数据再插入，同时删除redis缓存并更新。
初版方案最终耗时为116s，可见是非常慢的。
为什么会这么慢？初步推断或许是用户数据量太大，单线程串行处理效率太低。
因此进行修改。
（二）修改
可以看到在此方案中，我们引入了线程池进行并行处理，并且为了防止，在用户id处还进行了游标分页。
修改方案最终耗时为70s。
虽有提升，但仍不理想。扩大尺寸后耗时未明显降低，排除分页查询次数为主要瓶颈。
那只可能在核心逻辑上出了问题，让我们分析整体逻辑，我们每遍历一个用户，就要先查用户关注的对象id，再查关注用户的对象id，然后还要查用户好友列表，如果前面两者的交集与好友列表不一致的话则需要先删除好友列表的数据再进行插入，最后再删除缓存并进行更新。
一套操作下来，每个用户处理最多会发起5次DB请求和2次Redis请求，一共有5w用户，累计下来最多会发起
5 万用户 * (5 DB + 2 Redis) =** 35 万次**，
并且随着用户的增加还会越来越多……
显然，这么设计是极其不合理的，因此，再次进行修改。
（三）终版
此方案中，取消了游标分页，因为整体只有5w数据量，占不了多少内存，使用分页反而会增加耗时。但不过在数据量抵达百万甚至更多时最好还是采用游标进行分页，不然可能会引发oom。
除此之外，本方案最大的改动就是遍历用户前查询关注表中所有的数据，并建立关系映射进行分组，好友表也是以每批用户的频率进行查询，然后在遍历的过程中，将要插入的数据放进一个集合当中，这样的话在遍历完成后DB和Redis就可以拿着这两个集合里的数据进行更新，
将原本将近35w的请求次数降到了1 (加载关注) + 1 (加载粉丝) + 1 (批量加载好友) + 1 (批量删除) + 1 (批量插入) + 1（Redis管道请求） =** 6 次批操作**
请求量从 35 万次降至常数级！
什么是Redis的管道？
Redis 的 管道 是一种优化网络通信的技术，它的核心思想非常简单：
打包发送： 客户端可以将多个需要执行的 Redis 命令一次性收集起来，打包成一个批次。一次传输： 将这个命令批次一次性发送给 Redis 服务器，而不是每个命令都单独发起一次网络请求。批量执行： Redis 服务器接收到这个批次后，会按顺序依次执行其中的所有命令。打包返回： 服务器将所有命令的执行结果一次性收集起来，打包成一个批次。一次接收： 服务器将这个结果批次一次性发送回给客户端。
该方案最终耗时为6s，足足优化了110s，因此最终采取该方案。
三、流程图总览
四、代码实现
（一）修改方案
@Slf4j@RequiredArgsConstructorpublic class CheckAndUpdateFriendTask { private final RedisTemplate<String,Object> redisTemplate; private final UserMapper userMapper; private final FriendMapper friendMapper; private final FollowMapper followMapper; private final ThreadPoolExecutor threadPoolExecutor;
@Scheduled(cron = "0 0 2 * * ?") public void processCheckAndUpdateFriend() { LocalDateTime now = LocalDateTime.now(); log.info("开始检查用户好友关系并更新，执行时间：{}", now);
// 游标分批获取用户id列表 // 先查最早注册的用户数据 LambdaQueryWrapper<User> buildCursor = Wrappers.lambdaQuery(User.class) .select(User::getId) .orderByAsc(User::getCreateTime) .last("limit 1");
User earliestUser = userMapper.selectOne(buildCursor);
// 构造游标 int batchSize = 1000; Long cursor = earliestUser.getId();
List<CompletableFuture<Void>> futures = new ArrayList<>(); // 游标查询 while (true) { // 构建查询条件 LambdaQueryWrapper<User> queryWrapper = Wrappers.lambdaQuery(User.class) .select(User::getId) .gt(User::getId, cursor) .orderByAsc(User::getId) .last("limit " + batchSize);
// 查询 List<Long> batchUserIds = userMapper.selectList(queryWrapper) .stream() .map(User::getId) .toList();
// 如果没有数据则结束查询 if (batchUserIds.isEmpty()) break;
// 先更新游标，即使后面出错，也不影响后面的查询 cursor = batchUserIds.getLast();
// 添加到结果列表中 CompletableFuture<Void> future = CompletableFuture.runAsync(() -> { try { dealUserIds(batchUserIds); } catch (Exception e) { log.error("处理好友关系时出错, batchSize={}", batchUserIds.size(), e); } }, threadPoolExecutor); futures.add(future); }
// 统一等待所有批次完成 ThreadPoolUtil.allFuturesWait(futures);
log.info("用户好友关系检查完成，结束时间：{} , 执行时长：{}", LocalDateTime.now() , Duration.between(now, LocalDateTime.now()).getSeconds()); }
private void dealUserIds(List<Long> userIds) { // 遍历用户id列表，检查用户互关关系，从而得到现在用户应该有的好友关系 for (Long userId : userIds) {
log.info("开始检查id为：{} 的好友关系", userId);
try { // 获取用户当前的关注对象id集合 LambdaQueryWrapper<Follow> isFollowQueryWrapper = Wrappers.lambdaQuery(Follow.class) .eq(Follow::getMyUserId, userId); List<Long> userFollowingIds = followMapper.selectList(isFollowQueryWrapper).stream() .map(Follow::getFollowUserId) .toList();
// 创建用户现在应该有的好友关系的集合 List<Long> expectedFriendIds = new ArrayList<>();
// 遍历用户关注对象id列表，检查用户是否互相关注 if (!userFollowingIds.isEmpty()) { // 查询所有互相关注的用户 LambdaQueryWrapper<Follow> mutualFollowWrapper = Wrappers.lambdaQuery(Follow.class) .eq(Follow::getFollowUserId, userId) .in(Follow::getMyUserId, userFollowingIds); List<Follow> mutualFollows = followMapper.selectList(mutualFollowWrapper);
// 提取互相关注的用户ID expectedFriendIds = mutualFollows.stream() .map(Follow::getMyUserId) .toList(); }
// 比较用户现在应该有的好友关系和用户当前的关注关系，如果不一致，则更新用户好友关系和redis数据 if(!expectedFriendIds.equals(userFollowingIds)) { // 删除用户好友表并进行更新 LambdaQueryWrapper<Friend> deleteByMyUserId = Wrappers.lambdaQuery(Friend.class) .eq(Friend::getMyUserId, userId); friendMapper.delete(deleteByMyUserId);
// 插入新的好友表数据 for (Long expectedFriendId : expectedFriendIds) { Friend friend = Friend.builder() .myUserId(userId) .friendUserId(expectedFriendId) .build(); friendMapper.insert(friend); }
// 删除现在的redis并进行更新 redisTemplate.delete(RedisConstant.FRIEND_USER_FOLLOW + userId);
// 插入新的redis数据 if (!expectedFriendIds.isEmpty()) { redisTemplate.opsForSet().add(RedisConstant.FRIEND_USER_FOLLOW + userId, expectedFriendIds.toArray()); } } } catch (Exception e) { log.error("检查用户id为{}的好友关系并更新出错：{}",userId, e.getMessage()); } } }}（二）最终方案
@Slf4j@RequiredArgsConstructor@Componentpublic class CheckAndUpdateFriendTask { private final RedisTemplate<String, Object> redisTemplate; private final UserMapper userMapper; private final FriendMapper friendMapper; private final FollowMapper followMapper; private final ThreadPoolExecutor threadPoolExecutor;
// 批量插入大小 private static final int BATCH_INSERT_SIZE = 1000;
@Transactional @Scheduled(cron = "0 0 2 * * ?") public void processCheckAndUpdateFriend() { LocalDateTime now = LocalDateTime.now(); log.info("开始检查用户好友关系并更新，执行时间：{}", now);
// 全量加载用户ID(因为数据量不大，所以一次性加载没有问题) List<Long> allUserIds = userMapper.selectList(Wrappers.lambdaQuery(User.class) .select(User::getId)).stream() .map(User::getId) .toList();
// 加载所有关注关系 List<Follow> allFollows = followMapper.selectList(Wrappers.emptyWrapper());
// 构建关系映射 Map<Long, Set<Long>> followingsMap = new HashMap<>(allUserIds.size()); Map<Long, Set<Long>> followersMap = new HashMap<>(allUserIds.size());
allFollows.forEach(follow -> { followingsMap.computeIfAbsent(follow.getMyUserId(), k -> new HashSet<>()) .add(follow.getFollowUserId()); followersMap.computeIfAbsent(follow.getFollowUserId(), k -> new HashSet<>()) .add(follow.getMyUserId()); });
// 分片处理（每 5000 用户一个分片） int sliceSize = 5000; List<List<Long>> userSlices = Lists.partition(allUserIds, sliceSize);
// 并行处理分片 List<CompletableFuture<Void>> futures = userSlices.stream() .map(slice -> CompletableFuture.runAsync(() -> processUserSlice(slice, followingsMap, followersMap), threadPoolExecutor)) .collect(Collectors.toList());
// 等待所有分片完成 ThreadPoolUtil.allFuturesWait(futures);
log.info("用户好友关系检查完成，结束时间：{} , 执行时长：{}s", LocalDateTime.now(), Duration.between(now, LocalDateTime.now()).getSeconds()); }
private void processUserSlice(List<Long> userIds, Map<Long, Set<Long>> followingsMap, Map<Long, Set<Long>> followersMap) {
// 获取现有好友关系 Map<Long, Set<Long>> currentFriendsMap = getCurrentFriends(userIds);
// 记录需要更新的用户 Set<Long> changedUserIds = new HashSet<>(); // 需要插入的新数据 List<Friend> toInsert = new ArrayList<>(); // 需要更新的Redis数据 Map<Long, Set<Long>> redisUpdates = new HashMap<>();
for (Long userId : userIds) { // 计算互关关系（交集） Set<Long> followings = followingsMap.getOrDefault(userId, Collections.emptySet()); Set<Long> followers = followersMap.getOrDefault(userId, Collections.emptySet()); Set<Long> mutualFollows = Sets.intersection(followings, followers);
// 检测变更 Set<Long> currentFriends = currentFriendsMap.getOrDefault(userId, Collections.emptySet()); if (!mutualFollows.equals(currentFriends)) { // 添加变更用户 changedUserIds.add(userId); // 收集数据库更新 mutualFollows.forEach(friendId -> toInsert.add(Friend.builder() .myUserId(userId) .friendUserId(friendId) .build()));
// 收集Redis更新 redisUpdates.put(userId, mutualFollows); } }
// 批量更新数据库 if (!changedUserIds.isEmpty()) { // 批量删除旧数据 friendMapper.delete(Wrappers.lambdaQuery(Friend.class) .in(Friend::getMyUserId, changedUserIds));
// 批量插入新数据 if (!toInsert.isEmpty()) { List<List<Friend>> batches = Lists.partition(toInsert, BATCH_INSERT_SIZE); batches.forEach(batch -> { if (!batch.isEmpty()) { batch.forEach(friendMapper::insert); } }); } }
// 批量更新Redis if (!redisUpdates.isEmpty()) { batchUpdateRedis(redisUpdates); } }
// 批量获取现有好友关系 private Map<Long, Set<Long>> getCurrentFriends(List<Long> userIds) { if (CollectionUtils.isEmpty(userIds)) { return Collections.emptyMap(); }
Map<Long, Set<Long>> result = new HashMap<>(); friendMapper.selectList(Wrappers.lambdaQuery(Friend.class) .in(Friend::getMyUserId, userIds)) .forEach(friend -> result.computeIfAbsent(friend.getMyUserId(), k -> new HashSet<>()) .add(friend.getFriendUserId()));
return result; }
// 批量更新Redis void batchUpdateRedis(Map<Long, Set<Long>> updates) { try (RedisConnection connection = Objects.requireNonNull(redisTemplate.getConnectionFactory()).getConnection()) { connection.openPipeline();
RedisKeyCommands keyCommands = connection.keyCommands(); RedisSetCommands setCommands = connection.setCommands();
updates.forEach((userId, friendIds) -> { String key = RedisConstant.FRIEND_USER_FOLLOW + userId; // 删除旧数据 keyCommands.del(key.getBytes());
// 插入新数据 if (!friendIds.isEmpty()) { byte[][] members = friendIds.stream() .map(String::valueOf) .map(String::getBytes) .toArray(byte[][]::new); setCommands.sAdd(key.getBytes(), members); } });
connection.closePipeline(); } }}如果你有更好的方案，请在评论区告诉我！
码文不易，留个赞再走吧
原文链接: 116秒→6秒：Redis管道+批处理优化用户好友关系校验的方案 作者: Yilena
分享
如果这篇文章对你有帮助，欢迎分享给更多人！
116秒→6秒：Redis管道+批处理优化用户好友关系校验的方案
https://github.com/emn178/markdown 部分信息可能已经过时
相关文章 智能推荐
1
360s→15s：近25倍的性能优化重构十万Excel券码批量导入方案
业务拆解 本文针对十万级Excel券码批量导入场景，将原逐行解析导致的20万次网络IO，通过Redis管道与MQ批处理优化至182次。方案兼顾了宕机恢复与库存扣减，最终将耗时从360秒大幅缩减至15秒左右。
2
布隆过滤器因内存上限无法处理超量数据的优化方案
业务拆解 本文针对布隆过滤器因数据量增加导致内存上限不足的问题，深入分析并提出了三种优化方案。首先是分片布隆过滤器，通过哈希取模将数据分散，简单高效但受限于单机内存且易引发GC问题；其次是分布式布隆过滤器，借助Redis摆脱单机限制，并结合一致性哈希实现分片以优化缓存空间；最后是可扩展布隆过滤器，通过动态增加层数和容量来应对数据增长，但查询效率会随层数增加而降低。综合对比，推荐单机项目使用分片方案，分布式架构采用Redis分布式分片方案，以有效解决超量数据处理难题。
3
如何应对海量Key带来的redis内存占用问题？
业务拆解 本文针对海量Key导致的Redis内存占用过高问题，深入分析了内存碎片化与元数据开销的根源，并提出了五种切实可行的解决方案。从基础的合并小Key（利用Hash结构）与临时添加TTL救急，到架构层面的Redis集群模式与Redis on Flash（内存+SSD）降本方案，再到结合MySQL的冷热数据分离策略（全量/冷数据存DB，热数据存Redis）。文章通过详实的流程图与优劣对比，帮助开发者在不同预算与业务场景下，科学应对Redis内存瓶颈。
4
优化日志分析店铺推荐方案：用户范围的精确度以及ES与MySQL的查询效率差异
业务拆解 本文针对点餐场景下的店铺推荐方案进行了深度优化。首先，通过Redis记录用户月度登录天数，精准筛选出热点用户，解决了旧版方案中推荐用户范围不精确的问题。其次，对比了Elasticsearch与MySQL在十万级数据量下的查询效率，将热数据迁移至MySQL并建立索引以提升查询性能，同时保留ES作为冷数据备份。文章详细展示了优化后的业务流程图及Java代码实现，包括基于Lua脚本的Redis原子操作、多线程并行处理用户日志分析以及加权评分推荐算法，为高并发场景下的日志分析与个性化推荐提供了高效的工程实践。
5
优化：将针对单一日志表的冷热数据分离类改造成通用类
业务拆解 本文记录了将针对单一日志表的冷热数据分离逻辑重构为通用类的优化过程。为解决旧版方案中代码冗余及复用性差的问题，新方案通过参数化固定逻辑、引入泛型机制适配不同数据实体，并为不同日志表提供专属的Elasticsearch插入重载函数，实现了高度定制化与通用性的统一。文章详细展示了重构后的具体流程图及Java代码实现，包括利用CompletableFuture进行多线程并行处理、基于游标的批量数据查询与迁移、以及完善的重试与异常处理机制，有效提升了海量日志数据冷热分离任务的执行效率与代码可维护性。