## 查询优化：批量拉黑状态查询

## 目录

• 一、背景
• 二、优化分析
• 三、优化代码
• 四、优化效果

## 一、背景

### 1.1 业务描述

• 1. 查询视频列表，过滤用户拉黑的主播
• 2. 表结构参考from_uid, to_uid, is_blocked

CREATE TABLE `t_user_block` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `from_uid` bigint(20) NOT NULL COMMENT '用户ID',
  `to_uid` bigint(20) NOT NULL COMMENT '拉黑用户ID',
  `is_blocked` tinyint(1) NOT NULL DEFAULT '0' COMMENT '是否拉黑 0-否 1-是',
  `create_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `updated_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  PRIMARY KEY (`id`),
  KEY `idx_from_to` (`from_uid`, `to_uid`)
) ENGINE=InnoDB AUTO_INCREMENT=437 DEFAULT CHARSET=utf8 COMMENT='用户拉黑表';

• 3. 考虑性能采用Redis有序集合存储
• 4. 通过zscore判断是否存储拉黑的用户
• 5. 提供批量查询接口（例如视频一次性拉取100条记录），from_uid，to_uids，批量判断拉黑状态
• 6. 拉黑表数据量大约100多万

### 1.2 问题描述

• 1. 接口响应RT通过Prometheus+Grafana+AlterManager监控，根据接口请求频次，响应时间进行告警
• 2. 收到批量查询拉黑接口响应时间超过100ms+的告警邮件
• 3. 大约1天100多封

## 二、优化分析

### 2.1 代码分析

• 1. 批量查询接口采用循环查询关注关系，100个用户则有100次zscore操作，对应会有100次与redis的网络IO
• 2. 参考Redis的批量操作MGET可以将多个zscore合并成一次执行，但是批量zscore并没有原生支持
• 3. 可以考虑使用Redis中Pipeline合并多次网络IO，自行封装zScore操作的代码

## 三、优化代码

### 3.1 优化之前代码

public Map<Long, Boolean> multiBlockOf(long from, String tos) {
  List<Long> toList = UserUtils.parseUIDString(tos);
  Map<Long, Boolean> result = new HashMap<>(toList.size());
  String blockKey = relationCacheService.getBlockKey(from);

  for (Long uid : toList) {
    result.put(uid, false);
    if (relationCacheService.redisHasKey(blockKey)) {
      ZSetOperations<String, String> zSetOps = relationCacheService.getRedisTemplate().opsForZSet();
      if (zSetOps.score(blockKey, String.valueOf(uid)) != null) {
        result.put(uid, true);
      }
    }
  }
  return result;
}

### 3.1 优化后代码

1. multiBlockOf()代码

public Map<Long, Boolean> multiBlockOf(long from, String tos) {
  List<Long> toList = UserUtils.parseUIDString(tos);

  Map<Long, Boolean> result = new HashMap<>(toList.size());

  // 之需要查询一次缓存Key
  String blockKey = relationCacheService.getBlockKey(from);
  if (!relationCacheService.redisHasKey(blockKey)) {
    return result;
  }

  Map<Long, Object> userExists = relationCacheService.zScoreMultiExists(blockKey, toList);

  for (Long uid : toList) {
    result.put(uid, userExists.containsKey(uid));
  }

  return result;
}

2. zScoreMultiExists代码

public Map<Long, Object> zScoreMultiExists(String key, List<Long> members) {
        if (CollectionUtils.isEmpty(members)) {
            return new HashMap<>();
        }

        List<Object> objects = redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            StringRedisSerializer serializer = new StringRedisSerializer();
            for (Long member : members) {
                connection.zScore(serializer.serialize(key), serializer.serialize(member.toString()));
            }
            return null;
        });

        Map<Long, Object> result = new HashMap<>();
        int size = members.size();
        for (int i = 0; i < size; i++) {
            Long member = members.get(i);
            Object val = objects.get(i);
            if (null != val) {
                result.put(member, val);
            }
        }

        return result;
    }

## 四、优化效果

### 4.1 优化本地测试

### 4.2 线上测试

1. 接口RT平均100ms

2. 接口TR平均1ms

### 4.3 效果

1. 接口性能RT提高100倍

2. 封装到RedisUtils工具类

参考资料

1. [Redis系列十：Pipeline详解](https://blog.csdn.net/w1lgy/article/details/84455579)