第271集使用Redis-Pipeline，接口性能提升10倍架构实战：Redis-Pipeline优化、性能提升方案与企业级Redis-Pipeline应用方案

/**
 * Redis-Pipeline基础服务
 */
@Service
public class RedisPipelineService {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private RedisConnectionFactory connectionFactory;

    /**
     * 批量设置缓存
     */
    public void batchSet(Map<String, Object> keyValueMap) {
        try {
            RedisConnection connection = connectionFactory.getConnection();

            try {
                // 开启Pipeline
                connection.openPipeline();

                // 批量设置
                for (Map.Entry<String, Object> entry : keyValueMap.entrySet()) {
                    String key = entry.getKey();
                    Object value = entry.getValue();

                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(key);
                    byte[] valueBytes = redisTemplate.getValueSerializer().serialize(value);

                    connection.set(keyBytes, valueBytes);
                }

                // 执行Pipeline
                List<Object> results = connection.closePipeline();

                log.info("批量设置缓存完成，数量: {}", keyValueMap.size());

            } finally {
                connection.close();
            }

        } catch (Exception e) {
            log.error("批量设置缓存失败", e);
            throw new RedisException("批量设置缓存失败", e);
        }
    }

    /**
     * 批量获取缓存
     */
    public Map<String, Object> batchGet(List<String> keys) {
        try {
            RedisConnection connection = connectionFactory.getConnection();

            try {
                // 开启Pipeline
                connection.openPipeline();

                // 批量获取
                for (String key : keys) {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(key);
                    connection.get(keyBytes);
                }

                // 执行Pipeline
                List<Object> results = connection.closePipeline();

                // 处理结果
                Map<String, Object> resultMap = new HashMap<>();
                for (int i = 0; i < keys.size(); i++) {
                    String key = keys.get(i);
                    Object value = results.get(i);

                    if (value != null) {
                        resultMap.put(key, value);
                    }
                }

                log.info("批量获取缓存完成，数量: {}", keys.size());
                return resultMap;

            } finally {
                connection.close();
            }

        } catch (Exception e) {
            log.error("批量获取缓存失败", e);
            throw new RedisException("批量获取缓存失败", e);
        }
    }

    /**
     * 批量删除缓存
     */
    public void batchDelete(List<String> keys) {
        try {
            RedisConnection connection = connectionFactory.getConnection();

            try {
                // 开启Pipeline
                connection.openPipeline();

                // 批量删除
                for (String key : keys) {
                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(key);
                    connection.del(keyBytes);
                }

                // 执行Pipeline
                List<Object> results = connection.closePipeline();

                log.info("批量删除缓存完成，数量: {}", keys.size());

            } finally {
                connection.close();
            }

        } catch (Exception e) {
            log.error("批量删除缓存失败", e);
            throw new RedisException("批量删除缓存失败", e);
        }
    }

    /**
     * 批量设置过期时间
     */
    public void batchExpire(Map<String, Long> keyExpireMap) {
        try {
            RedisConnection connection = connectionFactory.getConnection();

            try {
                // 开启Pipeline
                connection.openPipeline();

                // 批量设置过期时间
                for (Map.Entry<String, Long> entry : keyExpireMap.entrySet()) {
                    String key = entry.getKey();
                    Long expireSeconds = entry.getValue();

                    byte[] keyBytes = redisTemplate.getStringSerializer().serialize(key);
                    connection.expire(keyBytes, expireSeconds);
                }

                // 执行Pipeline
                List<Object> results = connection.closePipeline();

                log.info("批量设置过期时间完成，数量: {}", keyExpireMap.size());

            } finally {
                connection.close();
            }

        } catch (Exception e) {
            log.error("批量设置过期时间失败", e);
            throw new RedisException("批量设置过期时间失败", e);
        }
    }

    /**
     * 批量Hash操作
     */
    public void batchHashOperations(String hashKey, Map<String, Object> fieldValueMap) {
        try {
            RedisConnection connection = connectionFactory.getConnection();

            try {
                // 开启Pipeline
                connection.openPipeline();

                byte[] hashKeyBytes = redisTemplate.getStringSerializer().serialize(hashKey);

                // 批量设置Hash字段
                for (Map.Entry<String, Object> entry : fieldValueMap.entrySet()) {
                    String field = entry.getKey();
                    Object value = entry.getValue();

                    byte[] fieldBytes = redisTemplate.getStringSerializer().serialize(field);
                    byte[] valueBytes = redisTemplate.getValueSerializer().serialize(value);

                    connection.hSet(hashKeyBytes, fieldBytes, valueBytes);
                }

                // 执行Pipeline
                List<Object> results = connection.closePipeline();

                log.info("批量Hash操作完成，数量: {}", fieldValueMap.size());

            } finally {
                connection.close();
            }

        } catch (Exception e) {
            log.error("批量Hash操作失败", e);
            throw new RedisException("批量Hash操作失败", e);
        }
    }

    /**
     * 批量List操作
     */
    public void batchListOperations(String listKey, List<Object> values, boolean isLeftPush) {
        try {
            RedisConnection connection = connectionFactory.getConnection();

            try {
                // 开启Pipeline
                connection.openPipeline();

                byte[] listKeyBytes = redisTemplate.getStringSerializer().serialize(listKey);

                // 批量List操作
                for (Object value : values) {
                    byte[] valueBytes = redisTemplate.getValueSerializer().serialize(value);

                    if (isLeftPush) {
                        connection.lPush(listKeyBytes, valueBytes);
                    } else {
                        connection.rPush(listKeyBytes, valueBytes);
                    }
                }

                // 执行Pipeline
                List<Object> results = connection.closePipeline();

                log.info("批量List操作完成，数量: {}", values.size());

            } finally {
                connection.close();
            }

        } catch (Exception e) {
            log.error("批量List操作失败", e);
            throw new RedisException("批量List操作失败", e);
        }
    }

    /**
     * 批量Set操作
     */
    public void batchSetOperations(String setKey, Set<Object> values) {
        try {
            RedisConnection connection = connectionFactory.getConnection();

            try {
                // 开启Pipeline
                connection.openPipeline();

                byte[] setKeyBytes = redisTemplate.getStringSerializer().serialize(setKey);

                // 批量Set操作
                for (Object value : values) {
                    byte[] valueBytes = redisTemplate.getValueSerializer().serialize(value);
                    connection.sAdd(setKeyBytes, valueBytes);
                }

                // 执行Pipeline
                List<Object> results = connection.closePipeline();

                log.info("批量Set操作完成，数量: {}", values.size());

            } finally {
                connection.close();
            }

        } catch (Exception e) {
            log.error("批量Set操作失败", e);
            throw new RedisException("批量Set操作失败", e);
        }
    }

    /**
     * 批量ZSet操作
     */
    public void batchZSetOperations(String zsetKey, Map<Object, Double> memberScoreMap) {
        try {
            RedisConnection connection = connectionFactory.getConnection();

            try {
                // 开启Pipeline
                connection.openPipeline();

                byte[] zsetKeyBytes = redisTemplate.getStringSerializer().serialize(zsetKey);

                // 批量ZSet操作
                for (Map.Entry<Object, Double> entry : memberScoreMap.entrySet()) {
                    Object member = entry.getKey();
                    Double score = entry.getValue();

                    byte[] memberBytes = redisTemplate.getValueSerializer().serialize(member);
                    connection.zAdd(zsetKeyBytes, score, memberBytes);
                }

                // 执行Pipeline
                List<Object> results = connection.closePipeline();

                log.info("批量ZSet操作完成，数量: {}", memberScoreMap.size());

            } finally {
                connection.close();
            }

        } catch (Exception e) {
            log.error("批量ZSet操作失败", e);
            throw new RedisException("批量ZSet操作失败", e);
        }
    }
}

/**
 * Redis-Pipeline性能优化服务
 */
@Service
public class RedisPipelineOptimizationService {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private RedisConnectionFactory connectionFactory;

    private final int DEFAULT_BATCH_SIZE = 1000;
    private final int MAX_BATCH_SIZE = 10000;

    /**
     * 优化的批量设置缓存
     */
    public void optimizedBatchSet(Map<String, Object> keyValueMap) {
        try {
            if (keyValueMap.isEmpty()) {
                return;
            }

            // 分批处理
            List<Map<String, Object>> batches = splitIntoBatches(keyValueMap, DEFAULT_BATCH_SIZE);

            // 并行处理批次
            batches.parallelStream().forEach(batch -> {
                try {
                    batchSetWithPipeline(batch);
                } catch (Exception e) {
                    log.error("批次设置缓存失败", e);
                }
            });

            log.info("优化的批量设置缓存完成，总数量: {}", keyValueMap.size());

        } catch (Exception e) {
            log.error("优化的批量设置缓存失败", e);
            throw new RedisException("优化的批量设置缓存失败", e);
        }
    }

    /**
     * 分批处理
     */
    private List<Map<String, Object>> splitIntoBatches(Map<String, Object> keyValueMap, int batchSize) {
        List<Map<String, Object>> batches = new ArrayList<>();
        Map<String, Object> currentBatch = new HashMap<>();

        for (Map.Entry<String, Object> entry : keyValueMap.entrySet()) {
            currentBatch.put(entry.getKey(), entry.getValue());

            if (currentBatch.size() >= batchSize) {
                batches.add(new HashMap<>(currentBatch));
                currentBatch.clear();
            }
        }

        if (!currentBatch.isEmpty()) {
            batches.add(currentBatch);
        }

        return batches;
    }

    /**
     * 使用Pipeline批量设置
     */
    private void batchSetWithPipeline(Map<String, Object> keyValueMap) {
        RedisConnection connection = connectionFactory.getConnection();

        try {
            // 开启Pipeline
            connection.openPipeline();

            // 批量设置
            for (Map.Entry<String, Object> entry : keyValueMap.entrySet()) {
                String key = entry.getKey();
                Object value = entry.getValue();

                byte[] keyBytes = redisTemplate.getStringSerializer().serialize(key);
                byte[] valueBytes = redisTemplate.getValueSerializer().serialize(value);

                connection.set(keyBytes, valueBytes);
            }

            // 执行Pipeline
            List<Object> results = connection.closePipeline();

        } finally {
            connection.close();
        }
    }

    /**
     * 优化的批量获取缓存
     */
    public Map<String, Object> optimizedBatchGet(List<String> keys) {
        try {
            if (keys.isEmpty()) {
                return new HashMap<>();
            }

            // 分批处理
            List<List<String>> batches = splitIntoBatches(keys, DEFAULT_BATCH_SIZE);

            // 并行处理批次
            Map<String, Object> resultMap = new ConcurrentHashMap<>();

            batches.parallelStream().forEach(batch -> {
                try {
                    Map<String, Object> batchResult = batchGetWithPipeline(batch);
                    resultMap.putAll(batchResult);
                } catch (Exception e) {
                    log.error("批次获取缓存失败", e);
                }
            });

            log.info("优化的批量获取缓存完成，总数量: {}", keys.size());
            return resultMap;

        } catch (Exception e) {
            log.error("优化的批量获取缓存失败", e);
            throw new RedisException("优化的批量获取缓存失败", e);
        }
    }

    /**
     * 分批处理列表
     */
    private List<List<String>> splitIntoBatches(List<String> keys, int batchSize) {
        List<List<String>> batches = new ArrayList<>();

        for (int i = 0; i < keys.size(); i += batchSize) {
            int endIndex = Math.min(i + batchSize, keys.size());
            batches.add(keys.subList(i, endIndex));
        }

        return batches;
    }

    /**
     * 使用Pipeline批量获取
     */
    private Map<String, Object> batchGetWithPipeline(List<String> keys) {
        RedisConnection connection = connectionFactory.getConnection();

        try {
            // 开启Pipeline
            connection.openPipeline();

            // 批量获取
            for (String key : keys) {
                byte[] keyBytes = redisTemplate.getStringSerializer().serialize(key);
                connection.get(keyBytes);
            }

            // 执行Pipeline
            List<Object> results = connection.closePipeline();

            // 处理结果
            Map<String, Object> resultMap = new HashMap<>();
            for (int i = 0; i < keys.size(); i++) {
                String key = keys.get(i);
                Object value = results.get(i);

                if (value != null) {
                    resultMap.put(key, value);
                }
            }

            return resultMap;

        } finally {
            connection.close();
        }
    }

    /**
     * 性能测试
     */
    public PipelinePerformanceTestResult performanceTest(int testCount) {
        try {
            PipelinePerformanceTestResult result = new PipelinePerformanceTestResult();
            result.setTestCount(testCount);

            // 准备测试数据
            Map<String, Object> testData = prepareTestData(testCount);

            // 测试普通操作性能
            long normalTime = testNormalOperations(testData);
            result.setNormalTime(normalTime);

            // 测试Pipeline操作性能
            long pipelineTime = testPipelineOperations(testData);
            result.setPipelineTime(pipelineTime);

            // 计算性能提升
            double improvement = (double) (normalTime - pipelineTime) / normalTime * 100;
            result.setPerformanceImprovement(improvement);

            // 计算吞吐量
            double normalThroughput = (double) testCount / normalTime * 1000;
            double pipelineThroughput = (double) testCount / pipelineTime * 1000;

            result.setNormalThroughput(normalThroughput);
            result.setPipelineThroughput(pipelineThroughput);

            return result;

        } catch (Exception e) {
            log.error("性能测试失败", e);
            throw new RedisException("性能测试失败", e);
        }
    }

    /**
     * 准备测试数据
     */
    private Map<String, Object> prepareTestData(int count) {
        Map<String, Object> testData = new HashMap<>();

        for (int i = 0; i < count; i++) {
            String key = "test_key_" + i;
            String value = "test_value_" + i;
            testData.put(key, value);
        }

        return testData;
    }

    /**
     * 测试普通操作性能
     */
    private long testNormalOperations(Map<String, Object> testData) {
        long startTime = System.currentTimeMillis();

        for (Map.Entry<String, Object> entry : testData.entrySet()) {
            redisTemplate.opsForValue().set(entry.getKey(), entry.getValue());
        }

        long endTime = System.currentTimeMillis();
        return endTime - startTime;
    }

    /**
     * 测试Pipeline操作性能
     */
    private long testPipelineOperations(Map<String, Object> testData) {
        long startTime = System.currentTimeMillis();

        optimizedBatchSet(testData);

        long endTime = System.currentTimeMillis();
        return endTime - startTime;
    }

    /**
     * 动态调整批次大小
     */
    public int calculateOptimalBatchSize(int dataSize, int maxConcurrency) {
        try {
            // 基于数据大小和并发数计算最优批次大小
            int optimalBatchSize = dataSize / maxConcurrency;

            // 限制在合理范围内
            optimalBatchSize = Math.max(optimalBatchSize, 100);
            optimalBatchSize = Math.min(optimalBatchSize, MAX_BATCH_SIZE);

            return optimalBatchSize;

        } catch (Exception e) {
            log.error("计算最优批次大小失败", e);
            return DEFAULT_BATCH_SIZE;
        }
    }

    /**
     * 连接池优化
     */
    public void optimizeConnectionPool() {
        try {
            if (connectionFactory instanceof LettuceConnectionFactory) {
                LettuceConnectionFactory lettuceFactory = (LettuceConnectionFactory) connectionFactory;

                // 优化连接池配置
                GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
                poolConfig.setMaxTotal(100);
                poolConfig.setMaxIdle(50);
                poolConfig.setMinIdle(10);
                poolConfig.setMaxWaitMillis(3000);

                lettuceFactory.setPoolConfig(poolConfig);

                log.info("连接池优化完成");
            }

        } catch (Exception e) {
            log.error("连接池优化失败", e);
        }
    }
}

/**
 * 电商Redis-Pipeline服务
 */
@Service
public class EcommerceRedisPipelineService {

    @Autowired
    private RedisPipelineService pipelineService;

    @Autowired
    private RedisPipelineOptimizationService optimizationService;

    /**
     * 批量缓存用户信息
     */
    public void batchCacheUserInfo(List<UserInfo> userInfos) {
        try {
            Map<String, Object> userInfoMap = new HashMap<>();

            for (UserInfo userInfo : userInfos) {
                String key = "user:info:" + userInfo.getUserId();
                userInfoMap.put(key, userInfo);
            }

            // 使用Pipeline批量设置
            optimizationService.optimizedBatchSet(userInfoMap);

            log.info("批量缓存用户信息完成，数量: {}", userInfos.size());

        } catch (Exception e) {
            log.error("批量缓存用户信息失败", e);
            throw new EcommerceException("批量缓存用户信息失败", e);
        }
    }

    /**
     * 批量获取用户信息
     */
    public Map<String, UserInfo> batchGetUserInfo(List<String> userIds) {
        try {
            List<String> keys = new ArrayList<>();

            for (String userId : userIds) {
                String key = "user:info:" + userId;
                keys.add(key);
            }

            // 使用Pipeline批量获取
            Map<String, Object> resultMap = optimizationService.optimizedBatchGet(keys);

            // 转换结果
            Map<String, UserInfo> userInfoMap = new HashMap<>();
            for (Map.Entry<String, Object> entry : resultMap.entrySet()) {
                String key = entry.getKey();
                Object value = entry.getValue();

                if (value instanceof UserInfo) {
                    String userId = key.substring("user:info:".length());
                    userInfoMap.put(userId, (UserInfo) value);
                }
            }

            log.info("批量获取用户信息完成，数量: {}", userIds.size());
            return userInfoMap;

        } catch (Exception e) {
            log.error("批量获取用户信息失败", e);
            throw new EcommerceException("批量获取用户信息失败", e);
        }
    }

    /**
     * 批量缓存商品信息
     */
    public void batchCacheProductInfo(List<ProductInfo> productInfos) {
        try {
            Map<String, Object> productInfoMap = new HashMap<>();

            for (ProductInfo productInfo : productInfos) {
                String key = "product:info:" + productInfo.getProductId();
                productInfoMap.put(key, productInfo);
            }

            // 使用Pipeline批量设置
            optimizationService.optimizedBatchSet(productInfoMap);

            log.info("批量缓存商品信息完成，数量: {}", productInfos.size());

        } catch (Exception e) {
            log.error("批量缓存商品信息失败", e);
            throw new EcommerceException("批量缓存商品信息失败", e);
        }
    }

    /**
     * 批量获取商品信息
     */
    public Map<String, ProductInfo> batchGetProductInfo(List<String> productIds) {
        try {
            List<String> keys = new ArrayList<>();

            for (String productId : productIds) {
                String key = "product:info:" + productId;
                keys.add(key);
            }

            // 使用Pipeline批量获取
            Map<String, Object> resultMap = optimizationService.optimizedBatchGet(keys);

            // 转换结果
            Map<String, ProductInfo> productInfoMap = new HashMap<>();
            for (Map.Entry<String, Object> entry : resultMap.entrySet()) {
                String key = entry.getKey();
                Object value = entry.getValue();

                if (value instanceof ProductInfo) {
                    String productId = key.substring("product:info:".length());
                    productInfoMap.put(productId, (ProductInfo) value);
                }
            }

            log.info("批量获取商品信息完成，数量: {}", productIds.size());
            return productInfoMap;

        } catch (Exception e) {
            log.error("批量获取商品信息失败", e);
            throw new EcommerceException("批量获取商品信息失败", e);
        }
    }

    /**
     * 批量缓存订单信息
     */
    public void batchCacheOrderInfo(List<OrderInfo> orderInfos) {
        try {
            Map<String, Object> orderInfoMap = new HashMap<>();

            for (OrderInfo orderInfo : orderInfos) {
                String key = "order:info:" + orderInfo.getOrderId();
                orderInfoMap.put(key, orderInfo);
            }

            // 使用Pipeline批量设置
            optimizationService.optimizedBatchSet(orderInfoMap);

            log.info("批量缓存订单信息完成，数量: {}", orderInfos.size());

        } catch (Exception e) {
            log.error("批量缓存订单信息失败", e);
            throw new EcommerceException("批量缓存订单信息失败", e);
        }
    }

    /**
     * 批量获取订单信息
     */
    public Map<String, OrderInfo> batchGetOrderInfo(List<String> orderIds) {
        try {
            List<String> keys = new ArrayList<>();

            for (String orderId : orderIds) {
                String key = "order:info:" + orderId;
                keys.add(key);
            }

            // 使用Pipeline批量获取
            Map<String, Object> resultMap = optimizationService.optimizedBatchGet(keys);

            // 转换结果
            Map<String, OrderInfo> orderInfoMap = new HashMap<>();
            for (Map.Entry<String, Object> entry : resultMap.entrySet()) {
                String key = entry.getKey();
                Object value = entry.getValue();

                if (value instanceof OrderInfo) {
                    String orderId = key.substring("order:info:".length());
                    orderInfoMap.put(orderId, (OrderInfo) value);
                }
            }

            log.info("批量获取订单信息完成，数量: {}", orderIds.size());
            return orderInfoMap;

        } catch (Exception e) {
            log.error("批量获取订单信息失败", e);
            throw new EcommerceException("批量获取订单信息失败", e);
        }
    }

    /**
     * 批量更新购物车
     */
    public void batchUpdateCart(String userId, Map<String, Integer> cartItems) {
        try {
            String cartKey = "cart:" + userId;

            // 使用Pipeline批量更新购物车
            pipelineService.batchHashOperations(cartKey, cartItems);

            log.info("批量更新购物车完成，用户: {}, 商品数量: {}", userId, cartItems.size());

        } catch (Exception e) {
            log.error("批量更新购物车失败", e);
            throw new EcommerceException("批量更新购物车失败", e);
        }
    }

    /**
     * 批量获取购物车
     */
    public Map<String, Integer> batchGetCart(String userId) {
        try {
            String cartKey = "cart:" + userId;

            // 使用Pipeline批量获取购物车
            Map<String, Object> cartMap = pipelineService.batchGetHashFields(cartKey);

            // 转换结果
            Map<String, Integer> cartItems = new HashMap<>();
            for (Map.Entry<String, Object> entry : cartMap.entrySet()) {
                String productId = entry.getKey();
                Object quantity = entry.getValue();

                if (quantity instanceof Integer) {
                    cartItems.put(productId, (Integer) quantity);
                }
            }

            log.info("批量获取购物车完成，用户: {}", userId);
            return cartItems;

        } catch (Exception e) {
            log.error("批量获取购物车失败", e);
            throw new EcommerceException("批量获取购物车失败", e);
        }
    }

    /**
     * 批量更新库存
     */
    public void batchUpdateStock(Map<String, Integer> stockUpdates) {
        try {
            String stockKey = "stock:info";

            // 使用Pipeline批量更新库存
            pipelineService.batchHashOperations(stockKey, stockUpdates);

            log.info("批量更新库存完成，商品数量: {}", stockUpdates.size());

        } catch (Exception e) {
            log.error("批量更新库存失败", e);
            throw new EcommerceException("批量更新库存失败", e);
        }
    }

    /**
     * 批量获取库存
     */
    public Map<String, Integer> batchGetStock(List<String> productIds) {
        try {
            String stockKey = "stock:info";

            // 使用Pipeline批量获取库存
            Map<String, Object> stockMap = pipelineService.batchGetHashFields(stockKey);

            // 转换结果
            Map<String, Integer> stockInfo = new HashMap<>();
            for (String productId : productIds) {
                Object stock = stockMap.get(productId);
                if (stock instanceof Integer) {
                    stockInfo.put(productId, (Integer) stock);
                }
            }

            log.info("批量获取库存完成，商品数量: {}", productIds.size());
            return stockInfo;

        } catch (Exception e) {
            log.error("批量获取库存失败", e);
            throw new EcommerceException("批量获取库存失败", e);
        }
    }

    /**
     * 批量更新商品评分
     */
    public void batchUpdateProductRating(Map<String, Double> ratingUpdates) {
        try {
            String ratingKey = "product:rating";

            // 使用Pipeline批量更新商品评分
            pipelineService.batchZSetOperations(ratingKey, ratingUpdates);

            log.info("批量更新商品评分完成，商品数量: {}", ratingUpdates.size());

        } catch (Exception e) {
            log.error("批量更新商品评分失败", e);
            throw new EcommerceException("批量更新商品评分失败", e);
        }
    }

    /**
     * 批量获取商品评分
     */
    public Map<String, Double> batchGetProductRating(List<String> productIds) {
        try {
            String ratingKey = "product:rating";

            // 使用Pipeline批量获取商品评分
            Map<String, Double> ratingInfo = new HashMap<>();

            for (String productId : productIds) {
                // 获取商品评分
                Double rating = pipelineService.getZSetScore(ratingKey, productId);
                if (rating != null) {
                    ratingInfo.put(productId, rating);
                }
            }

            log.info("批量获取商品评分完成，商品数量: {}", productIds.size());
            return ratingInfo;

        } catch (Exception e) {
            log.error("批量获取商品评分失败", e);
            throw new EcommerceException("批量获取商品评分失败", e);
        }
    }
}

/**
 * 社交Redis-Pipeline服务
 */
@Service
public class SocialRedisPipelineService {

    @Autowired
    private RedisPipelineService pipelineService;

    @Autowired
    private RedisPipelineOptimizationService optimizationService;

    /**
     * 批量缓存用户信息
     */
    public void batchCacheUserInfo(List<UserInfo> userInfos) {
        try {
            Map<String, Object> userInfoMap = new HashMap<>();

            for (UserInfo userInfo : userInfos) {
                String key = "user:info:" + userInfo.getUserId();
                userInfoMap.put(key, userInfo);
            }

            // 使用Pipeline批量设置
            optimizationService.optimizedBatchSet(userInfoMap);

            log.info("批量缓存用户信息完成，数量: {}", userInfos.size());

        } catch (Exception e) {
            log.error("批量缓存用户信息失败", e);
            throw new SocialException("批量缓存用户信息失败", e);
        }
    }

    /**
     * 批量获取用户信息
     */
    public Map<String, UserInfo> batchGetUserInfo(List<String> userIds) {
        try {
            List<String> keys = new ArrayList<>();

            for (String userId : userIds) {
                String key = "user:info:" + userId;
                keys.add(key);
            }

            // 使用Pipeline批量获取
            Map<String, Object> resultMap = optimizationService.optimizedBatchGet(keys);

            // 转换结果
            Map<String, UserInfo> userInfoMap = new HashMap<>();
            for (Map.Entry<String, Object> entry : resultMap.entrySet()) {
                String key = entry.getKey();
                Object value = entry.getValue();

                if (value instanceof UserInfo) {
                    String userId = key.substring("user:info:".length());
                    userInfoMap.put(userId, (UserInfo) value);
                }
            }

            log.info("批量获取用户信息完成，数量: {}", userIds.size());
            return userInfoMap;

        } catch (Exception e) {
            log.error("批量获取用户信息失败", e);
            throw new SocialException("批量获取用户信息失败", e);
        }
    }

    /**
     * 批量缓存帖子信息
     */
    public void batchCachePostInfo(List<PostInfo> postInfos) {
        try {
            Map<String, Object> postInfoMap = new HashMap<>();

            for (PostInfo postInfo : postInfos) {
                String key = "post:info:" + postInfo.getPostId();
                postInfoMap.put(key, postInfo);
            }

            // 使用Pipeline批量设置
            optimizationService.optimizedBatchSet(postInfoMap);

            log.info("批量缓存帖子信息完成，数量: {}", postInfos.size());

        } catch (Exception e) {
            log.error("批量缓存帖子信息失败", e);
            throw new SocialException("批量缓存帖子信息失败", e);
        }
    }

    /**
     * 批量获取帖子信息
     */
    public Map<String, PostInfo> batchGetPostInfo(List<String> postIds) {
        try {
            List<String> keys = new ArrayList<>();

            for (String postId : postIds) {
                String key = "post:info:" + postId;
                keys.add(key);
            }

            // 使用Pipeline批量获取
            Map<String, Object> resultMap = optimizationService.optimizedBatchGet(keys);

            // 转换结果
            Map<String, PostInfo> postInfoMap = new HashMap<>();
            for (Map.Entry<String, Object> entry : resultMap.entrySet()) {
                String key = entry.getKey();
                Object value = entry.getValue();

                if (value instanceof PostInfo) {
                    String postId = key.substring("post:info:".length());
                    postInfoMap.put(postId, (PostInfo) value);
                }
            }

            log.info("批量获取帖子信息完成，数量: {}", postIds.size());
            return postInfoMap;

        } catch (Exception e) {
            log.error("批量获取帖子信息失败", e);
            throw new SocialException("批量获取帖子信息失败", e);
        }
    }

    /**
     * 批量缓存评论信息
     */
    public void batchCacheCommentInfo(List<CommentInfo> commentInfos) {
        try {
            Map<String, Object> commentInfoMap = new HashMap<>();

            for (CommentInfo commentInfo : commentInfos) {
                String key = "comment:info:" + commentInfo.getCommentId();
                commentInfoMap.put(key, commentInfo);
            }

            // 使用Pipeline批量设置
            optimizationService.optimizedBatchSet(commentInfoMap);

            log.info("批量缓存评论信息完成，数量: {}", commentInfos.size());

        } catch (Exception e) {
            log.error("批量缓存评论信息失败", e);
            throw new SocialException("批量缓存评论信息失败", e);
        }
    }

    /**
     * 批量获取评论信息
     */
    public Map<String, CommentInfo> batchGetCommentInfo(List<String> commentIds) {
        try {
            List<String> keys = new ArrayList<>();

            for (String commentId : commentIds) {
                String key = "comment:info:" + commentId;
                keys.add(key);
            }

            // 使用Pipeline批量获取
            Map<String, Object> resultMap = optimizationService.optimizedBatchGet(keys);

            // 转换结果
            Map<String, CommentInfo> commentInfoMap = new HashMap<>();
            for (Map.Entry<String, Object> entry : resultMap.entrySet()) {
                String key = entry.getKey();
                Object value = entry.getValue();

                if (value instanceof CommentInfo) {
                    String commentId = key.substring("comment:info:".length());
                    commentInfoMap.put(commentId, (CommentInfo) value);
                }
            }

            log.info("批量获取评论信息完成，数量: {}", commentIds.size());
            return commentInfoMap;

        } catch (Exception e) {
            log.error("批量获取评论信息失败", e);
            throw new SocialException("批量获取评论信息失败", e);
        }
    }

    /**
     * 批量更新用户关注
     */
    public void batchUpdateUserFollow(String userId, Set<String> followUserIds) {
        try {
            String followKey = "user:follow:" + userId;

            // 使用Pipeline批量更新关注
            pipelineService.batchSetOperations(followKey, followUserIds);

            log.info("批量更新用户关注完成，用户: {}, 关注数量: {}", userId, followUserIds.size());

        } catch (Exception e) {
            log.error("批量更新用户关注失败", e);
            throw new SocialException("批量更新用户关注失败", e);
        }
    }

    /**
     * 批量获取用户关注
     */
    public Set<String> batchGetUserFollow(String userId) {
        try {
            String followKey = "user:follow:" + userId;

            // 使用Pipeline批量获取关注
            Set<Object> followSet = pipelineService.getSetMembers(followKey);

            // 转换结果
            Set<String> followUserIds = new HashSet<>();
            for (Object follow : followSet) {
                if (follow instanceof String) {
                    followUserIds.add((String) follow);
                }
            }

            log.info("批量获取用户关注完成，用户: {}", userId);
            return followUserIds;

        } catch (Exception e) {
            log.error("批量获取用户关注失败", e);
            throw new SocialException("批量获取用户关注失败", e);
        }
    }

    /**
     * 批量更新用户粉丝
     */
    public void batchUpdateUserFans(String userId, Set<String> fanUserIds) {
        try {
            String fanKey = "user:fan:" + userId;

            // 使用Pipeline批量更新粉丝
            pipelineService.batchSetOperations(fanKey, fanUserIds);

            log.info("批量更新用户粉丝完成，用户: {}, 粉丝数量: {}", userId, fanUserIds.size());

        } catch (Exception e) {
            log.error("批量更新用户粉丝失败", e);
            throw new SocialException("批量更新用户粉丝失败", e);
        }
    }

    /**
     * 批量获取用户粉丝
     */
    public Set<String> batchGetUserFans(String userId) {
        try {
            String fanKey = "user:fan:" + userId;

            // 使用Pipeline批量获取粉丝
            Set<Object> fanSet = pipelineService.getSetMembers(fanKey);

            // 转换结果
            Set<String> fanUserIds = new HashSet<>();
            for (Object fan : fanSet) {
                if (fan instanceof String) {
                    fanUserIds.add((String) fan);
                }
            }

            log.info("批量获取用户粉丝完成，用户: {}", userId);
            return fanUserIds;

        } catch (Exception e) {
            log.error("批量获取用户粉丝失败", e);
            throw new SocialException("批量获取用户粉丝失败", e);
        }
    }

    /**
     * 批量更新帖子点赞
     */
    public void batchUpdatePostLike(String postId, Set<String> likeUserIds) {
        try {
            String likeKey = "post:like:" + postId;

            // 使用Pipeline批量更新点赞
            pipelineService.batchSetOperations(likeKey, likeUserIds);

            log.info("批量更新帖子点赞完成，帖子: {}, 点赞数量: {}", postId, likeUserIds.size());

        } catch (Exception e) {
            log.error("批量更新帖子点赞失败", e);
            throw new SocialException("批量更新帖子点赞失败", e);
        }
    }

    /**
     * 批量获取帖子点赞
     */
    public Set<String> batchGetPostLike(String postId) {
        try {
            String likeKey = "post:like:" + postId;

            // 使用Pipeline批量获取点赞
            Set<Object> likeSet = pipelineService.getSetMembers(likeKey);

            // 转换结果
            Set<String> likeUserIds = new HashSet<>();
            for (Object like : likeSet) {
                if (like instanceof String) {
                    likeUserIds.add((String) like);
                }
            }

            log.info("批量获取帖子点赞完成，帖子: {}", postId);
            return likeUserIds;

        } catch (Exception e) {
            log.error("批量获取帖子点赞失败", e);
            throw new SocialException("批量获取帖子点赞失败", e);
        }
    }

    /**
     * 批量更新帖子评论
     */
    public void batchUpdatePostComment(String postId, List<String> commentIds) {
        try {
            String commentKey = "post:comment:" + postId;

            // 使用Pipeline批量更新评论
            pipelineService.batchListOperations(commentKey, commentIds, false);

            log.info("批量更新帖子评论完成，帖子: {}, 评论数量: {}", postId, commentIds.size());

        } catch (Exception e) {
            log.error("批量更新帖子评论失败", e);
            throw new SocialException("批量更新帖子评论失败", e);
        }
    }

    /**
     * 批量获取帖子评论
     */
    public List<String> batchGetPostComment(String postId, int start, int end) {
        try {
            String commentKey = "post:comment:" + postId;

            // 使用Pipeline批量获取评论
            List<Object> commentList = pipelineService.getListRange(commentKey, start, end);

            // 转换结果
            List<String> commentIds = new ArrayList<>();
            for (Object comment : commentList) {
                if (comment instanceof String) {
                    commentIds.add((String) comment);
                }
            }

            log.info("批量获取帖子评论完成，帖子: {}", postId);
            return commentIds;

        } catch (Exception e) {
            log.error("批量获取帖子评论失败", e);
            throw new SocialException("批量获取帖子评论失败", e);
        }
    }
}

/**
 * Redis-Pipeline性能优化服务
 */
@Service
public class RedisPipelinePerformanceOptimizationService {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private RedisConnectionFactory connectionFactory;

    private final String PIPELINE_CACHE_PREFIX = "pipeline_cache:";

    /**
     * 缓存Pipeline结果
     */
    public void cachePipelineResult(String operation, Object result) {
        String cacheKey = PIPELINE_CACHE_PREFIX + operation;

        try {
            // 写入本地缓存
            localCache.put(cacheKey, result);

            // 写入Redis缓存
            String redisCacheKey = "redis_cache:" + cacheKey;
            redisTemplate.opsForValue().set(redisCacheKey, result, Duration.ofHours(1));

        } catch (Exception e) {
            log.error("缓存Pipeline结果失败", e);
        }
    }

    /**
     * 获取缓存的Pipeline结果
     */
    public Object getCachedPipelineResult(String operation) {
        String cacheKey = PIPELINE_CACHE_PREFIX + operation;

        try {
            // 从本地缓存获取
            Object cachedResult = localCache.getIfPresent(cacheKey);
            if (cachedResult != null) {
                return cachedResult;
            }

            // 从Redis获取
            String redisCacheKey = "redis_cache:" + cacheKey;
            Object redisResult = redisTemplate.opsForValue().get(redisCacheKey);
            if (redisResult != null) {
                localCache.put(cacheKey, redisResult);
                return redisResult;
            }

        } catch (Exception e) {
            log.error("获取缓存的Pipeline结果失败", e);
        }

        return null;
    }

    /**
     * 批量Pipeline优化
     */
    public void batchPipelineOptimized(List<PipelineOperation> operations) {
        try {
            // 按操作类型分组
            Map<String, List<PipelineOperation>> groupedOperations = operations.stream()
                    .collect(Collectors.groupingBy(PipelineOperation::getOperationType));

            // 并行处理各类型操作
            groupedOperations.entrySet().parallelStream().forEach(entry -> {
                String operationType = entry.getKey();
                List<PipelineOperation> operationList = entry.getValue();

                try {
                    processBatchOperations(operationType, operationList);
                } catch (Exception e) {
                    log.error("批量Pipeline操作失败: {}", operationType, e);
                }
            });

        } catch (Exception e) {
            log.error("批量Pipeline优化失败", e);
            throw new RedisException("批量Pipeline优化失败", e);
        }
    }

    /**
     * 处理批量操作
     */
    private void processBatchOperations(String operationType, List<PipelineOperation> operations) {
        try {
            RedisConnection connection = connectionFactory.getConnection();

            try {
                // 开启Pipeline
                connection.openPipeline();

                // 执行操作
                for (PipelineOperation operation : operations) {
                    executeOperation(connection, operation);
                }

                // 执行Pipeline
                List<Object> results = connection.closePipeline();

                // 处理结果
                processResults(operations, results);

            } finally {
                connection.close();
            }

        } catch (Exception e) {
            log.error("处理批量操作失败", e);
        }
    }

    /**
     * 执行操作
     */
    private void executeOperation(RedisConnection connection, PipelineOperation operation) {
        try {
            switch (operation.getOperationType()) {
                case "SET":
                    connection.set(operation.getKeyBytes(), operation.getValueBytes());
                    break;
                case "GET":
                    connection.get(operation.getKeyBytes());
                    break;
                case "DEL":
                    connection.del(operation.getKeyBytes());
                    break;
                case "HSET":
                    connection.hSet(operation.getKeyBytes(), operation.getFieldBytes(), operation.getValueBytes());
                    break;
                case "HGET":
                    connection.hGet(operation.getKeyBytes(), operation.getFieldBytes());
                    break;
                default:
                    log.warn("不支持的操作类型: {}", operation.getOperationType());
                    break;
            }
        } catch (Exception e) {
            log.error("执行操作失败", e);
        }
    }

    /**
     * 处理结果
     */
    private void processResults(List<PipelineOperation> operations, List<Object> results) {
        try {
            for (int i = 0; i < operations.size(); i++) {
                PipelineOperation operation = operations.get(i);
                Object result = results.get(i);

                // 处理结果
                operation.setResult(result);
            }
        } catch (Exception e) {
            log.error("处理结果失败", e);
        }
    }

    /**
     * 预热Pipeline缓存
     */
    @PostConstruct
    public void warmupPipelineCache() {
        try {
            // 预热常用操作
            List<String> commonOperations = Arrays.asList("user:info", "product:info", "order:info");

            for (String operation : commonOperations) {
                try {
                    // 预热操作结果
                    cachePipelineResult(operation, new Object());
                } catch (Exception e) {
                    log.error("预热Pipeline缓存失败: {}", operation, e);
                }
            }

        } catch (Exception e) {
            log.error("预热Pipeline缓存失败", e);
        }
    }

    /**
     * 清理过期缓存
     */
    @Scheduled(fixedRate = 300000) // 5分钟
    public void cleanupExpiredCache() {
        try {
            // 清理本地缓存
            localCache.cleanUp();

            // 清理Redis过期缓存
            cleanupRedisExpiredCache();

        } catch (Exception e) {
            log.error("清理过期缓存失败", e);
        }
    }

    /**
     * 清理Redis过期缓存
     */
    private void cleanupRedisExpiredCache() {
        try {
            Set<String> cacheKeys = redisTemplate.keys("redis_cache:" + PIPELINE_CACHE_PREFIX + "*");

            for (String key : cacheKeys) {
                Long ttl = redisTemplate.getExpire(key);
                if (ttl != null && ttl <= 0) {
                    redisTemplate.delete(key);
                }
            }

        } catch (Exception e) {
            log.error("清理Redis过期缓存失败", e);
        }
    }
}

/**
 * Redis-Pipeline监控指标
 */
@Component
public class RedisPipelineMetrics {

    private final MeterRegistry meterRegistry;

    public RedisPipelineMetrics(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
    }

    /**
     * 记录Pipeline操作次数
     */
    public void recordPipelineOperationCount(String operationType) {
        Counter.builder("redis.pipeline.operation.count")
                .description("Redis-Pipeline操作次数")
                .tag("operation_type", operationType)
                .register(meterRegistry)
                .increment();
    }

    /**
     * 记录Pipeline操作时间
     */
    public void recordPipelineOperationTime(String operationType, long duration) {
        Timer.builder("redis.pipeline.operation.time")
                .description("Redis-Pipeline操作时间")
                .tag("operation_type", operationType)
                .register(meterRegistry)
                .record(duration, TimeUnit.MILLISECONDS);
    }

    /**
     * 记录Pipeline操作成功率
     */
    public void recordPipelineOperationSuccess(String operationType) {
        Counter.builder("redis.pipeline.operation.success")
                .description("Redis-Pipeline操作成功次数")
                .tag("operation_type", operationType)
                .register(meterRegistry)
                .increment();
    }

    /**
     * 记录Pipeline操作失败次数
     */
    public void recordPipelineOperationFailure(String operationType) {
        Counter.builder("redis.pipeline.operation.failure")
                .description("Redis-Pipeline操作失败次数")
                .tag("operation_type", operationType)
                .register(meterRegistry)
                .increment();
    }

    /**
     * 记录Pipeline批量大小
     */
    public void recordPipelineBatchSize(String operationType, int batchSize) {
        Gauge.builder("redis.pipeline.batch.size")
                .description("Redis-Pipeline批量大小")
                .tag("operation_type", operationType)
                .register(meterRegistry, batchSize);
    }

    /**
     * 记录Pipeline吞吐量
     */
    public void recordPipelineThroughput(String operationType, double throughput) {
        Gauge.builder("redis.pipeline.throughput")
                .description("Redis-Pipeline吞吐量")
                .tag("operation_type", operationType)
                .register(meterRegistry, throughput);
    }

    /**
     * 记录Pipeline网络往返次数
     */
    public void recordPipelineNetworkRoundtrips(String operationType, int roundtrips) {
        Gauge.builder("redis.pipeline.network.roundtrips")
                .description("Redis-Pipeline网络往返次数")
                .tag("operation_type", operationType)
                .register(meterRegistry, roundtrips);
    }

    /**
     * 记录Pipeline性能提升
     */
    public void recordPipelinePerformanceImprovement(String operationType, double improvement) {
        Gauge.builder("redis.pipeline.performance.improvement")
                .description("Redis-Pipeline性能提升")
                .tag("operation_type", operationType)
                .register(meterRegistry, improvement);
    }
}