redis 使用Pipelined 管道命令批量操作 减少网络操作次数

用Pipelined 管道命令批量查询 / 更新 redis里1000个key, 每个key value设置独立的过期时间都是30分钟

// 批量获取Redis键值 List<String> appMetricKeys = metricDataList.stream().map(metricData -> metricData.getSourceMetricName() + StrPool.COLON + metricData.getSourceMetricId()).collect(Collectors.toList()); // 使用管道批量查询 List<Object> redisValues = redisTemplate.executePipelined((RedisCallback<Object>)connection -> { for (String key : appMetricKeys) { connection.get(redisTemplate.getKeySerializer().serialize(key)); } return null; }); // 处理需要发送的数据 List<MetricData> needSendData = new ArrayList<>(); for (int i = 0; i < metricDataList.size(); i++) { MetricData metricData = metricDataList.get(i); String redisValue = redisValues.get(i) != null ? redisValues.get(i).toString() : null; if (redisValue == null || !ObjectUtil.equals(metricData.getSourceMetricData(), redisValue)) { needSendData.add(metricData); } } // 批量更新Redis if (CollUtil.isNotEmpty(needSendData)) { redisTemplate.executePipelined((RedisCallback<Object>)connection -> { for (MetricData metricData : needSendData) { String key = TINGYUN_ACTION_REDIS_KEY + metricData.getSourceMetricName() + StrPool.COLON + metricData.getSourceMetricId(); Object value = metricData.getSourceMetricData(); connection.setEx(redisTemplate.getKeySerializer().serialize(key), 30 * 60, redisTemplate.getValueSerializer().serialize(value)); // 30分钟 } return null; }); // 批量发送Kafka needSendData.forEach(metricData -> { kafkaTemplate.send(Constant.METRIC_DATA_TOPIC, JSON.toJSONString(metricData)); }); }