第148集Redis批量写操作前先判断总量Java实战

1. Redis批量写操作总量判断概述

Redis批量写操作前先判断总量是保证Redis性能和稳定性的重要手段，通过在执行批量写操作前检查数据总量，可以有效防止内存溢出、性能下降、系统崩溃等问题。在Java应用中，合理使用Redis批量写操作总量判断可以实现内存保护、性能优化、系统稳定性提升等功能。本文将详细介绍Redis批量写操作总量判断的原理、实现方法、性能优化技巧以及在Java实战中的应用。

1.1 Redis批量写操作总量判断核心价值

内存保护: 防止Redis内存溢出和系统崩溃
性能优化: 通过总量判断优化批量操作性能
系统稳定性: 保证Redis系统稳定运行
资源管理: 合理管理Redis内存资源
监控告警: 提供丰富的监控指标

1.2 Redis批量写操作总量判断场景

批量数据导入: 大量数据批量导入Redis
数据迁移: 数据从一个Redis实例迁移到另一个
缓存预热: 批量预热缓存数据
数据同步: 批量同步数据到Redis
系统保护: 防止系统过载

1.3 总量判断策略

内存使用率: 基于Redis内存使用率判断
键数量: 基于Redis键数量判断
数据大小: 基于数据大小判断
操作复杂度: 基于操作复杂度判断

2. Redis批量写操作总量判断基础实现

2.1 Redis批量写操作配置类

/**
 * Redis批量写操作配置类
 * @author 运维实战
 */
@Configuration
@EnableConfigurationProperties(RedisBatchWriteProperties.class)
public class RedisBatchWriteConfig {
    
    @Autowired
    private RedisBatchWriteProperties properties;
    
    /**
     * Redis批量写操作服务
     * @return 批量写操作服务
     */
    @Bean
    public RedisBatchWriteService redisBatchWriteService() {
        return new RedisBatchWriteService();
    }
    
    /**
     * Redis总量检查服务
     * @return 总量检查服务
     */
    @Bean
    public RedisTotalCheckService redisTotalCheckService() {
        return new RedisTotalCheckService();
    }
    
    /**
     * Redis批量写操作监控服务
     * @return 监控服务
     */
    @Bean
    public RedisBatchWriteMonitorService redisBatchWriteMonitorService() {
        return new RedisBatchWriteMonitorService();
    }
    
    private static final Logger logger = LoggerFactory.getLogger(RedisBatchWriteConfig.class);
}

2.2 Redis批量写操作属性配置

/**
 * Redis批量写操作属性配置
 * @author 运维实战
 */
@Data
@ConfigurationProperties(prefix = "redis.batch.write")
public class RedisBatchWriteProperties {
    
    /**
     * 最大批量写入数量
     */
    private int maxBatchSize = 1000;
    
    /**
     * 最大内存使用率阈值
     */
    private double maxMemoryUsageThreshold = 0.8;
    
    /**
     * 最大键数量阈值
     */
    private long maxKeyCountThreshold = 1000000;
    
    /**
     * 是否启用总量检查
     */
    private boolean enableTotalCheck = true;
    
    /**
     * 是否启用内存检查
     */
    private boolean enableMemoryCheck = true;
    
    /**
     * 是否启用键数量检查
     */
    private boolean enableKeyCountCheck = true;
    
    /**
     * 检查间隔（毫秒）
     */
    private long checkInterval = 1000;
    
    /**
     * 是否启用监控
     */
    private boolean enableMonitor = true;
    
    /**
     * 监控间隔（毫秒）
     */
    private long monitorInterval = 30000;
    
    /**
     * 是否启用告警
     */
    private boolean enableAlert = true;
    
    /**
     * 告警阈值
     */
    private double alertThreshold = 0.9;
}

2.3 基础Redis批量写操作服务

/**
 * 基础Redis批量写操作服务
 * @author 运维实战
 */
@Service
public class RedisBatchWriteService {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private RedisTotalCheckService totalCheckService;
    
    @Autowired
    private RedisBatchWriteProperties properties;
    
    @Autowired
    private RedisBatchWriteMonitorService monitorService;
    
    private static final Logger logger = LoggerFactory.getLogger(RedisBatchWriteService.class);
    
    /**
     * 批量写入前先判断总量
     * @param dataMap 数据映射
     * @return 批量写入结果
     */
    public RedisBatchWriteResult batchWriteWithTotalCheck(Map<String, Object> dataMap) {
        logger.info("开始批量写入操作，数据量: {}", dataMap.size());
        
        RedisBatchWriteResult result = new RedisBatchWriteResult();
        result.setTotalCount(dataMap.size());
        result.setStartTime(System.currentTimeMillis());
        
        try {
            // 执行总量检查
            TotalCheckResult checkResult = totalCheckService.checkTotal(dataMap.size());
            
            if (!checkResult.isAllowed()) {
                logger.warn("总量检查失败: {}", checkResult.getMessage());
                result.setSuccess(false);
                result.setError(checkResult.getMessage());
                result.setEndTime(System.currentTimeMillis());
                return result;
            }
            
            // 执行批量写入
            result = executeBatchWrite(dataMap);
            
            // 记录监控指标
            monitorService.recordBatchWrite(result);
            
            logger.info("批量写入完成，成功: {}, 失败: {}, 总耗时: {}ms", 
                result.getSuccessCount(), result.getFailureCount(), result.getDuration());
            
            return result;
            
        } catch (Exception e) {
            logger.error("批量写入异常", e);
            result.setSuccess(false);
            result.setError("批量写入异常: " + e.getMessage());
            result.setEndTime(System.currentTimeMillis());
            return result;
        }
    }
    
    /**
     * 分批写入前先判断总量
     * @param dataMap 数据映射
     * @param batchSize 批次大小
     * @return 批量写入结果
     */
    public RedisBatchWriteResult batchWriteWithTotalCheck(Map<String, Object> dataMap, int batchSize) {
        logger.info("开始分批写入操作，总数据量: {}, 批次大小: {}", dataMap.size(), batchSize);
        
        RedisBatchWriteResult result = new RedisBatchWriteResult();
        result.setTotalCount(dataMap.size());
        result.setStartTime(System.currentTimeMillis());
        
        try {
            // 执行总量检查
            TotalCheckResult checkResult = totalCheckService.checkTotal(dataMap.size());
            
            if (!checkResult.isAllowed()) {
                logger.warn("总量检查失败: {}", checkResult.getMessage());
                result.setSuccess(false);
                result.setError(checkResult.getMessage());
                result.setEndTime(System.currentTimeMillis());
                return result;
            }
            
            // 分批执行写入
            List<Map<String, Object>> batches = partitionMap(dataMap, batchSize);
            int successCount = 0;
            int failureCount = 0;
            
            for (int i = 0; i < batches.size(); i++) {
                Map<String, Object> batch = batches.get(i);
                
                try {
                    // 执行批次写入
                    RedisBatchWriteResult batchResult = executeBatchWrite(batch);
                    
                    if (batchResult.isSuccess()) {
                        successCount += batchResult.getSuccessCount();
                        failureCount += batchResult.getFailureCount();
                    } else {
                        failureCount += batch.size();
                    }
                    
                    logger.info("批次 {} 写入完成，成功: {}, 失败: {}", 
                        i + 1, batchResult.getSuccessCount(), batchResult.getFailureCount());
                    
                    // 批次间延迟
                    if (i < batches.size() - 1) {
                        Thread.sleep(properties.getCheckInterval());
                    }
                    
                } catch (Exception e) {
                    logger.error("批次 {} 写入异常", i + 1, e);
                    failureCount += batch.size();
                }
            }
            
            result.setSuccessCount(successCount);
            result.setFailureCount(failureCount);
            result.setSuccess(successCount > 0);
            result.setEndTime(System.currentTimeMillis());
            
            // 记录监控指标
            monitorService.recordBatchWrite(result);
            
            logger.info("分批写入完成，成功: {}, 失败: {}, 总耗时: {}ms", 
                successCount, failureCount, result.getDuration());
            
            return result;
            
        } catch (Exception e) {
            logger.error("分批写入异常", e);
            result.setSuccess(false);
            result.setError("分批写入异常: " + e.getMessage());
            result.setEndTime(System.currentTimeMillis());
            return result;
        }
    }
    
    /**
     * 执行批量写入
     * @param dataMap 数据映射
     * @return 批量写入结果
     */
    private RedisBatchWriteResult executeBatchWrite(Map<String, Object> dataMap) {
        RedisBatchWriteResult result = new RedisBatchWriteResult();
        result.setTotalCount(dataMap.size());
        result.setStartTime(System.currentTimeMillis());
        
        try {
            // 使用Redis Pipeline执行批量写入
            List<Object> results = redisTemplate.executePipelined(new RedisCallback<Object>() {
                @Override
                public Object doInRedis(RedisConnection connection) throws DataAccessException {
                    for (Map.Entry<String, Object> entry : dataMap.entrySet()) {
                        connection.set(entry.getKey().getBytes(), 
                            redisTemplate.getValueSerializer().serialize(entry.getValue()));
                    }
                    return null;
                }
            });
            
            int successCount = 0;
            int failureCount = 0;
            
            for (Object obj : results) {
                if (obj != null && "OK".equals(obj.toString())) {
                    successCount++;
                } else {
                    failureCount++;
                }
            }
            
            result.setSuccessCount(successCount);
            result.setFailureCount(failureCount);
            result.setSuccess(successCount > 0);
            result.setEndTime(System.currentTimeMillis());
            
            return result;
            
        } catch (Exception e) {
            logger.error("执行批量写入异常", e);
            result.setSuccess(false);
            result.setError("执行批量写入异常: " + e.getMessage());
            result.setEndTime(System.currentTimeMillis());
            return result;
        }
    }
    
    /**
     * 分割映射
     * @param dataMap 数据映射
     * @param batchSize 批次大小
     * @return 分割后的映射列表
     */
    private List<Map<String, Object>> partitionMap(Map<String, Object> dataMap, int batchSize) {
        List<Map<String, Object>> batches = new ArrayList<>();
        List<Map.Entry<String, Object>> entries = new ArrayList<>(dataMap.entrySet());
        
        for (int i = 0; i < entries.size(); i += batchSize) {
            int end = Math.min(i + batchSize, entries.size());
            Map<String, Object> batch = new HashMap<>();
            
            for (int j = i; j < end; j++) {
                Map.Entry<String, Object> entry = entries.get(j);
                batch.put(entry.getKey(), entry.getValue());
            }
            
            batches.add(batch);
        }
        
        return batches;
    }
}

2.4 Redis总量检查服务

/**
 * Redis总量检查服务
 * @author 运维实战
 */
@Service
public class RedisTotalCheckService {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private RedisBatchWriteProperties properties;
    
    private static final Logger logger = LoggerFactory.getLogger(RedisTotalCheckService.class);
    
    /**
     * 检查总量
     * @param dataCount 数据数量
     * @return 检查结果
     */
    public TotalCheckResult checkTotal(int dataCount) {
        logger.info("开始总量检查，数据数量: {}", dataCount);
        
        TotalCheckResult result = new TotalCheckResult();
        result.setDataCount(dataCount);
        result.setCheckTime(System.currentTimeMillis());
        
        try {
            // 检查批量大小
            if (dataCount > properties.getMaxBatchSize()) {
                result.setAllowed(false);
                result.setMessage("批量大小超过限制: " + dataCount + " > " + properties.getMaxBatchSize());
                return result;
            }
            
            // 检查内存使用率
            if (properties.isEnableMemoryCheck()) {
                double memoryUsage = getMemoryUsage();
                if (memoryUsage > properties.getMaxMemoryUsageThreshold()) {
                    result.setAllowed(false);
                    result.setMessage("内存使用率过高: " + String.format("%.2f", memoryUsage * 100) + "%");
                    return result;
                }
                result.setMemoryUsage(memoryUsage);
            }
            
            // 检查键数量
            if (properties.isEnableKeyCountCheck()) {
                long keyCount = getKeyCount();
                if (keyCount > properties.getMaxKeyCountThreshold()) {
                    result.setAllowed(false);
                    result.setMessage("键数量超过限制: " + keyCount + " > " + properties.getMaxKeyCountThreshold());
                    return result;
                }
                result.setKeyCount(keyCount);
            }
            
            // 检查通过
            result.setAllowed(true);
            result.setMessage("总量检查通过");
            
            logger.info("总量检查通过，数据数量: {}, 内存使用率: {}%, 键数量: {}", 
                dataCount, String.format("%.2f", result.getMemoryUsage() * 100), result.getKeyCount());
            
            return result;
            
        } catch (Exception e) {
            logger.error("总量检查异常", e);
            result.setAllowed(false);
            result.setMessage("总量检查异常: " + e.getMessage());
            return result;
        }
    }
    
    /**
     * 获取内存使用率
     * @return 内存使用率
     */
    private double getMemoryUsage() {
        try {
            Properties info = redisTemplate.getConnectionFactory().getConnection().info("memory");
            String usedMemory = info.getProperty("used_memory");
            String maxMemory = info.getProperty("maxmemory");
            
            if (usedMemory != null && maxMemory != null) {
                long used = Long.parseLong(usedMemory);
                long max = Long.parseLong(maxMemory);
                return (double) used / max;
            }
            
            return 0.0;
            
        } catch (Exception e) {
            logger.error("获取内存使用率失败", e);
            return 0.0;
        }
    }
    
    /**
     * 获取键数量
     * @return 键数量
     */
    private long getKeyCount() {
        try {
            return redisTemplate.getConnectionFactory().getConnection().dbSize();
        } catch (Exception e) {
            logger.error("获取键数量失败", e);
            return 0;
        }
    }
    
    /**
     * 获取Redis信息
     * @return Redis信息
     */
    public RedisInfo getRedisInfo() {
        try {
            Properties info = redisTemplate.getConnectionFactory().getConnection().info();
            
            RedisInfo redisInfo = new RedisInfo();
            redisInfo.setUsedMemory(Long.parseLong(info.getProperty("used_memory", "0")));
            redisInfo.setMaxMemory(Long.parseLong(info.getProperty("maxmemory", "0")));
            redisInfo.setKeyCount(getKeyCount());
            redisInfo.setConnectedClients(Integer.parseInt(info.getProperty("connected_clients", "0")));
            redisInfo.setUptimeInSeconds(Long.parseLong(info.getProperty("uptime_in_seconds", "0")));
            redisInfo.setCurrentTime(System.currentTimeMillis());
            
            return redisInfo;
            
        } catch (Exception e) {
            logger.error("获取Redis信息失败", e);
            return null;
        }
    }
}

2.5 Redis批量写入结果类

/**
 * Redis批量写入结果类
 * @author 运维实战
 */
@Data
public class RedisBatchWriteResult {
    
    private boolean success;
    private int totalCount;
    private int successCount;
    private int failureCount;
    private String error;
    private long startTime;
    private long endTime;
    
    public RedisBatchWriteResult() {
        this.success = false;
        this.successCount = 0;
        this.failureCount = 0;
    }
    
    /**
     * 获取处理耗时
     * @return 处理耗时（毫秒）
     */
    public long getDuration() {
        return endTime - startTime;
    }
    
    /**
     * 获取成功率
     * @return 成功率
     */
    public double getSuccessRate() {
        if (totalCount == 0) return 0.0;
        return (double) successCount / totalCount * 100;
    }
    
    /**
     * 获取失败率
     * @return 失败率
     */
    public double getFailureRate() {
        if (totalCount == 0) return 0.0;
        return (double) failureCount / totalCount * 100;
    }
    
    /**
     * 是否全部成功
     * @return 是否全部成功
     */
    public boolean isAllSuccess() {
        return failureCount == 0;
    }
}

2.6 总量检查结果类

/**
 * 总量检查结果类
 * @author 运维实战
 */
@Data
public class TotalCheckResult {
    
    private boolean allowed;
    private int dataCount;
    private String message;
    private double memoryUsage;
    private long keyCount;
    private long checkTime;
    
    public TotalCheckResult() {
        this.allowed = false;
        this.memoryUsage = 0.0;
        this.keyCount = 0;
        this.checkTime = System.currentTimeMillis();
    }
    
    /**
     * 获取内存使用率百分比
     * @return 内存使用率百分比
     */
    public double getMemoryUsagePercentage() {
        return memoryUsage * 100;
    }
    
    /**
     * 是否通过检查
     * @return 是否通过检查
     */
    public boolean isPassed() {
        return allowed;
    }
}

2.7 Redis信息类

/**
 * Redis信息类
 * @author 运维实战
 */
@Data
public class RedisInfo {
    
    private long usedMemory;
    private long maxMemory;
    private long keyCount;
    private int connectedClients;
    private long uptimeInSeconds;
    private long currentTime;
    
    public RedisInfo() {
        this.currentTime = System.currentTimeMillis();
    }
    
    /**
     * 获取内存使用率
     * @return 内存使用率
     */
    public double getMemoryUsage() {
        if (maxMemory == 0) return 0.0;
        return (double) usedMemory / maxMemory;
    }
    
    /**
     * 获取内存使用率百分比
     * @return 内存使用率百分比
     */
    public double getMemoryUsagePercentage() {
        return getMemoryUsage() * 100;
    }
    
    /**
     * 是否健康
     * @return 是否健康
     */
    public boolean isHealthy() {
        return getMemoryUsage() < 0.8 && connectedClients < 1000;
    }
}

3. 高级功能实现

3.1 Redis批量写操作监控服务

/**
 * Redis批量写操作监控服务
 * @author 运维实战
 */
@Service
public class RedisBatchWriteMonitorService {
    
    private final AtomicLong totalBatchWrites = new AtomicLong(0);
    private final AtomicLong totalSuccessCount = new AtomicLong(0);
    private final AtomicLong totalFailureCount = new AtomicLong(0);
    private final AtomicLong totalDataCount = new AtomicLong(0);
    
    private long lastResetTime = System.currentTimeMillis();
    private final long resetInterval = 300000; // 5分钟重置一次
    
    private static final Logger logger = LoggerFactory.getLogger(RedisBatchWriteMonitorService.class);
    
    /**
     * 记录批量写入
     * @param result 批量写入结果
     */
    public void recordBatchWrite(RedisBatchWriteResult result) {
        totalBatchWrites.incrementAndGet();
        totalSuccessCount.addAndGet(result.getSuccessCount());
        totalFailureCount.addAndGet(result.getFailureCount());
        totalDataCount.addAndGet(result.getTotalCount());
        
        logger.debug("记录批量写入: 总数量={}, 成功={}, 失败={}, 耗时={}ms", 
            result.getTotalCount(), result.getSuccessCount(), result.getFailureCount(), result.getDuration());
    }
    
    /**
     * 获取监控指标
     * @return 监控指标
     */
    public RedisBatchWriteMetrics getMetrics() {
        // 检查是否需要重置
        if (System.currentTimeMillis() - lastResetTime > resetInterval) {
            resetMetrics();
        }
        
        RedisBatchWriteMetrics metrics = new RedisBatchWriteMetrics();
        metrics.setTotalBatchWrites(totalBatchWrites.get());
        metrics.setTotalSuccessCount(totalSuccessCount.get());
        metrics.setTotalFailureCount(totalFailureCount.get());
        metrics.setTotalDataCount(totalDataCount.get());
        metrics.setTimestamp(System.currentTimeMillis());
        
        return metrics;
    }
    
    /**
     * 重置指标
     */
    private void resetMetrics() {
        totalBatchWrites.set(0);
        totalSuccessCount.set(0);
        totalFailureCount.set(0);
        totalDataCount.set(0);
        lastResetTime = System.currentTimeMillis();
        
        logger.info("Redis批量写操作监控指标重置");
    }
    
    /**
     * 定期监控Redis批量写操作状态
     */
    @Scheduled(fixedRate = 30000) // 每30秒监控一次
    public void monitorRedisBatchWriteStatus() {
        try {
            RedisBatchWriteMetrics metrics = getMetrics();
            
            logger.info("Redis批量写操作监控: 总批次={}, 总成功数={}, 总失败数={}, 总数据量={}, 成功率={}%, 平均每批次数据量={}", 
                metrics.getTotalBatchWrites(), metrics.getTotalSuccessCount(), metrics.getTotalFailureCount(), 
                metrics.getTotalDataCount(), String.format("%.2f", metrics.getSuccessRate()), 
                String.format("%.2f", metrics.getAverageDataPerBatch()));
            
            // 检查异常情况
            if (metrics.getFailureRate() > 10) {
                logger.warn("Redis批量写操作失败率过高: {}%", String.format("%.2f", metrics.getFailureRate()));
            }
            
            if (metrics.getAverageDataPerBatch() > 1000) {
                logger.warn("Redis批量写操作平均批次大小过大: {}", String.format("%.2f", metrics.getAverageDataPerBatch()));
            }
            
        } catch (Exception e) {
            logger.error("Redis批量写操作状态监控失败", e);
        }
    }
}

3.2 Redis批量写操作指标类

/**
 * Redis批量写操作指标类
 * @author 运维实战
 */
@Data
public class RedisBatchWriteMetrics {
    
    private long totalBatchWrites;
    private long totalSuccessCount;
    private long totalFailureCount;
    private long totalDataCount;
    private long timestamp;
    
    public RedisBatchWriteMetrics() {
        this.timestamp = System.currentTimeMillis();
    }
    
    /**
     * 获取成功率
     * @return 成功率
     */
    public double getSuccessRate() {
        if (totalDataCount == 0) return 0.0;
        return (double) totalSuccessCount / totalDataCount * 100;
    }
    
    /**
     * 获取失败率
     * @return 失败率
     */
    public double getFailureRate() {
        if (totalDataCount == 0) return 0.0;
        return (double) totalFailureCount / totalDataCount * 100;
    }
    
    /**
     * 获取平均每批次数据量
     * @return 平均每批次数据量
     */
    public double getAverageDataPerBatch() {
        if (totalBatchWrites == 0) return 0.0;
        return (double) totalDataCount / totalBatchWrites;
    }
    
    /**
     * 是否健康
     * @return 是否健康
     */
    public boolean isHealthy() {
        return getFailureRate() < 10 && getAverageDataPerBatch() < 1000;
    }
}

3.3 Redis批量写操作优化服务

/**
 * Redis批量写操作优化服务
 * @author 运维实战
 */
@Service
public class RedisBatchWriteOptimizeService {
    
    @Autowired
    private RedisTotalCheckService totalCheckService;
    
    @Autowired
    private RedisBatchWriteProperties properties;
    
    private static final Logger logger = LoggerFactory.getLogger(RedisBatchWriteOptimizeService.class);
    
    /**
     * 优化批量写入大小
     * @param dataCount 数据数量
     * @return 优化后的批量大小
     */
    public int optimizeBatchSize(int dataCount) {
        logger.info("开始优化批量写入大小，数据数量: {}", dataCount);
        
        try {
            // 获取Redis信息
            RedisInfo redisInfo = totalCheckService.getRedisInfo();
            if (redisInfo == null) {
                return Math.min(dataCount, properties.getMaxBatchSize());
            }
            
            // 根据内存使用率调整批量大小
            double memoryUsage = redisInfo.getMemoryUsage();
            int optimizedBatchSize = properties.getMaxBatchSize();
            
            if (memoryUsage > 0.9) {
                // 内存使用率过高，减少批量大小
                optimizedBatchSize = Math.max(100, properties.getMaxBatchSize() / 4);
            } else if (memoryUsage > 0.8) {
                // 内存使用率较高，减少批量大小
                optimizedBatchSize = Math.max(200, properties.getMaxBatchSize() / 2);
            } else if (memoryUsage < 0.5) {
                // 内存使用率较低，可以增加批量大小
                optimizedBatchSize = Math.min(2000, properties.getMaxBatchSize() * 2);
            }
            
            // 根据键数量调整批量大小
            long keyCount = redisInfo.getKeyCount();
            if (keyCount > properties.getMaxKeyCountThreshold() * 0.9) {
                // 键数量接近阈值，减少批量大小
                optimizedBatchSize = Math.max(100, optimizedBatchSize / 2);
            }
            
            // 确保不超过数据数量
            optimizedBatchSize = Math.min(optimizedBatchSize, dataCount);
            
            logger.info("批量写入大小优化完成: 原始数量={}, 优化后批量大小={}, 内存使用率={}%, 键数量={}", 
                dataCount, optimizedBatchSize, String.format("%.2f", memoryUsage * 100), keyCount);
            
            return optimizedBatchSize;
            
        } catch (Exception e) {
            logger.error("优化批量写入大小失败", e);
            return Math.min(dataCount, properties.getMaxBatchSize());
        }
    }
    
    /**
     * 计算最优批次数量
     * @param dataCount 数据数量
     * @return 最优批次数量
     */
    public int calculateOptimalBatchCount(int dataCount) {
        int optimizedBatchSize = optimizeBatchSize(dataCount);
        return (int) Math.ceil((double) dataCount / optimizedBatchSize);
    }
    
    /**
     * 获取写入策略建议
     * @param dataCount 数据数量
     * @return 写入策略建议
     */
    public WriteStrategy getWriteStrategy(int dataCount) {
        WriteStrategy strategy = new WriteStrategy();
        strategy.setDataCount(dataCount);
        strategy.setRecommendedBatchSize(optimizeBatchSize(dataCount));
        strategy.setRecommendedBatchCount(calculateOptimalBatchCount(dataCount));
        strategy.setTimestamp(System.currentTimeMillis());
        
        // 根据数据量选择策略
        if (dataCount <= 100) {
            strategy.setStrategy("SINGLE_BATCH");
            strategy.setDescription("数据量较小，建议单批次写入");
        } else if (dataCount <= 1000) {
            strategy.setStrategy("SMALL_BATCHES");
            strategy.setDescription("数据量中等，建议小批次写入");
        } else if (dataCount <= 10000) {
            strategy.setStrategy("MEDIUM_BATCHES");
            strategy.setDescription("数据量较大，建议中批次写入");
        } else {
            strategy.setStrategy("LARGE_BATCHES");
            strategy.setDescription("数据量很大，建议大批次写入");
        }
        
        return strategy;
    }
}

3.4 写入策略类

/**
 * 写入策略类
 * @author 运维实战
 */
@Data
public class WriteStrategy {
    
    private int dataCount;
    private int recommendedBatchSize;
    private int recommendedBatchCount;
    private String strategy;
    private String description;
    private long timestamp;
    
    public WriteStrategy() {
        this.timestamp = System.currentTimeMillis();
    }
    
    /**
     * 获取预计耗时（毫秒）
     * @return 预计耗时
     */
    public long getEstimatedDuration() {
        // 基于经验值估算：每批次约100ms，加上批次间延迟
        return recommendedBatchCount * 100 + (recommendedBatchCount - 1) * 50;
    }
    
    /**
     * 获取预计内存使用量（字节）
     * @return 预计内存使用量
     */
    public long getEstimatedMemoryUsage() {
        // 基于经验值估算：每个键值对约1KB
        return dataCount * 1024;
    }
}

4. Redis批量写操作控制器

4.1 Redis批量写操作REST控制器

/**
 * Redis批量写操作REST控制器
 * @author 运维实战
 */
@RestController
@RequestMapping("/api/redis/batch/write")
public class RedisBatchWriteController {
    
    @Autowired
    private RedisBatchWriteService redisBatchWriteService;
    
    @Autowired
    private RedisTotalCheckService redisTotalCheckService;
    
    @Autowired
    private RedisBatchWriteOptimizeService optimizeService;
    
    @Autowired
    private RedisBatchWriteMonitorService monitorService;
    
    private static final Logger logger = LoggerFactory.getLogger(RedisBatchWriteController.class);
    
    /**
     * 批量写入前先判断总量
     * @param request 批量写入请求
     * @return 批量写入结果
     */
    @PostMapping("/with-check")
    public ResponseEntity<RedisBatchWriteResult> batchWriteWithCheck(@RequestBody RedisBatchWriteRequest request) {
        try {
            logger.info("接收到批量写入请求，数据量: {}", request.getDataMap().size());
            
            RedisBatchWriteResult result = redisBatchWriteService.batchWriteWithTotalCheck(request.getDataMap());
            
            return ResponseEntity.ok(result);
            
        } catch (Exception e) {
            logger.error("批量写入失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 分批写入前先判断总量
     * @param request 分批写入请求
     * @return 批量写入结果
     */
    @PostMapping("/with-check/batch")
    public ResponseEntity<RedisBatchWriteResult> batchWriteWithCheckAndBatch(@RequestBody RedisBatchWriteRequest request) {
        try {
            logger.info("接收到分批写入请求，数据量: {}, 批次大小: {}", 
                request.getDataMap().size(), request.getBatchSize());
            
            RedisBatchWriteResult result = redisBatchWriteService.batchWriteWithTotalCheck(
                request.getDataMap(), request.getBatchSize());
            
            return ResponseEntity.ok(result);
            
        } catch (Exception e) {
            logger.error("分批写入失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 智能批量写入
     * @param request 智能批量写入请求
     * @return 批量写入结果
     */
    @PostMapping("/smart")
    public ResponseEntity<RedisBatchWriteResult> smartBatchWrite(@RequestBody RedisBatchWriteRequest request) {
        try {
            logger.info("接收到智能批量写入请求，数据量: {}", request.getDataMap().size());
            
            // 获取写入策略
            WriteStrategy strategy = optimizeService.getWriteStrategy(request.getDataMap().size());
            
            // 使用优化后的批次大小执行写入
            RedisBatchWriteResult result = redisBatchWriteService.batchWriteWithTotalCheck(
                request.getDataMap(), strategy.getRecommendedBatchSize());
            
            return ResponseEntity.ok(result);
            
        } catch (Exception e) {
            logger.error("智能批量写入失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 获取总量检查结果
     * @param dataCount 数据数量
     * @return 总量检查结果
     */
    @GetMapping("/check/{dataCount}")
    public ResponseEntity<TotalCheckResult> checkTotal(@PathVariable int dataCount) {
        try {
            TotalCheckResult result = redisTotalCheckService.checkTotal(dataCount);
            return ResponseEntity.ok(result);
        } catch (Exception e) {
            logger.error("总量检查失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 获取Redis信息
     * @return Redis信息
     */
    @GetMapping("/info")
    public ResponseEntity<RedisInfo> getRedisInfo() {
        try {
            RedisInfo info = redisTotalCheckService.getRedisInfo();
            if (info != null) {
                return ResponseEntity.ok(info);
            } else {
                return ResponseEntity.notFound().build();
            }
        } catch (Exception e) {
            logger.error("获取Redis信息失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 获取写入策略
     * @param dataCount 数据数量
     * @return 写入策略
     */
    @GetMapping("/strategy/{dataCount}")
    public ResponseEntity<WriteStrategy> getWriteStrategy(@PathVariable int dataCount) {
        try {
            WriteStrategy strategy = optimizeService.getWriteStrategy(dataCount);
            return ResponseEntity.ok(strategy);
        } catch (Exception e) {
            logger.error("获取写入策略失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 获取监控指标
     * @return 监控指标
     */
    @GetMapping("/metrics")
    public ResponseEntity<RedisBatchWriteMetrics> getMetrics() {
        try {
            RedisBatchWriteMetrics metrics = monitorService.getMetrics();
            return ResponseEntity.ok(metrics);
        } catch (Exception e) {
            logger.error("获取监控指标失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
}

4.2 请求类定义

/**
 * Redis批量写入请求类
 * @author 运维实战
 */
@Data
public class RedisBatchWriteRequest {
    
    private Map<String, Object> dataMap;
    private int batchSize = 1000;
    
    public RedisBatchWriteRequest() {}
    
    public RedisBatchWriteRequest(Map<String, Object> dataMap) {
        this.dataMap = dataMap;
    }
    
    public RedisBatchWriteRequest(Map<String, Object> dataMap, int batchSize) {
        this.dataMap = dataMap;
        this.batchSize = batchSize;
    }
}

5. Redis批量写操作注解和AOP

5.1 Redis批量写操作注解

/**
 * Redis批量写操作注解
 * @author 运维实战
 */
@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RedisBatchWrite {
    
    /**
     * 最大批量大小
     */
    int maxBatchSize() default 1000;
    
    /**
     * 是否启用总量检查
     */
    boolean enableTotalCheck() default true;
    
    /**
     * 是否启用内存检查
     */
    boolean enableMemoryCheck() default true;
    
    /**
     * 是否启用键数量检查
     */
    boolean enableKeyCountCheck() default true;
    
    /**
     * 最大内存使用率阈值
     */
    double maxMemoryUsageThreshold() default 0.8;
    
    /**
     * 最大键数量阈值
     */
    long maxKeyCountThreshold() default 1000000;
    
    /**
     * 检查失败时的消息
     */
    String message() default "Redis批量写操作总量检查失败，请稍后重试";
    
    /**
     * 检查失败时的HTTP状态码
     */
    int statusCode() default 429;
}

5.2 Redis批量写操作AOP切面

/**
 * Redis批量写操作AOP切面
 * @author 运维实战
 */
@Aspect
@Component
public class RedisBatchWriteAspect {
    
    @Autowired
    private RedisTotalCheckService redisTotalCheckService;
    
    @Autowired
    private RedisBatchWriteOptimizeService optimizeService;
    
    private static final Logger logger = LoggerFactory.getLogger(RedisBatchWriteAspect.class);
    
    /**
     * Redis批量写操作切点
     */
    @Pointcut("@annotation(redisBatchWrite)")
    public void redisBatchWritePointcut(RedisBatchWrite redisBatchWrite) {}
    
    /**
     * Redis批量写操作环绕通知
     * @param joinPoint 连接点
     * @param redisBatchWrite 批量写操作注解
     * @return 执行结果
     * @throws Throwable 异常
     */
    @Around("redisBatchWritePointcut(redisBatchWrite)")
    public Object around(ProceedingJoinPoint joinPoint, RedisBatchWrite redisBatchWrite) throws Throwable {
        String methodName = joinPoint.getSignature().getName();
        
        try {
            // 获取方法参数
            Object[] args = joinPoint.getArgs();
            
            // 查找数据参数
            Map<String, Object> dataMap = findDataMap(args);
            
            if (dataMap != null) {
                // 执行总量检查
                TotalCheckResult checkResult = redisTotalCheckService.checkTotal(dataMap.size());
                
                if (!checkResult.isAllowed()) {
                    logger.warn("Redis批量写操作总量检查失败: method={}, message={}", methodName, checkResult.getMessage());
                    throw new RedisBatchWriteException(redisBatchWrite.message(), redisBatchWrite.statusCode());
                }
                
                // 获取写入策略
                WriteStrategy strategy = optimizeService.getWriteStrategy(dataMap.size());
                
                logger.info("Redis批量写操作总量检查通过: method={}, dataCount={}, strategy={}", 
                    methodName, dataMap.size(), strategy.getStrategy());
            }
            
            // 执行原方法
            return joinPoint.proceed();
            
        } catch (RedisBatchWriteException e) {
            throw e;
        } catch (Exception e) {
            logger.error("Redis批量写操作处理异常: method={}", methodName, e);
            throw new RedisBatchWriteException("Redis批量写操作处理异常", 500);
        }
    }
    
    /**
     * 查找数据映射参数
     * @param args 方法参数
     * @return 数据映射
     */
    private Map<String, Object> findDataMap(Object[] args) {
        for (Object arg : args) {
            if (arg instanceof Map) {
                @SuppressWarnings("unchecked")
                Map<String, Object> map = (Map<String, Object>) arg;
                return map;
            }
        }
        return null;
    }
}

5.3 Redis批量写操作异常类

/**
 * Redis批量写操作异常类
 * @author 运维实战
 */
public class RedisBatchWriteException extends RuntimeException {
    
    private final int statusCode;
    
    public RedisBatchWriteException(String message) {
        super(message);
        this.statusCode = 429;
    }
    
    public RedisBatchWriteException(String message, int statusCode) {
        super(message);
        this.statusCode = statusCode;
    }
    
    public RedisBatchWriteException(String message, Throwable cause) {
        super(message, cause);
        this.statusCode = 429;
    }
    
    public RedisBatchWriteException(String message, Throwable cause, int statusCode) {
        super(message, cause);
        this.statusCode = statusCode;
    }
    
    public int getStatusCode() {
        return statusCode;
    }
}

5.4 Redis批量写操作异常处理器

/**
 * Redis批量写操作异常处理器
 * @author 运维实战
 */
@ControllerAdvice
public class RedisBatchWriteExceptionHandler {
    
    private static final Logger logger = LoggerFactory.getLogger(RedisBatchWriteExceptionHandler.class);
    
    /**
     * 处理Redis批量写操作异常
     * @param e 异常
     * @return 错误响应
     */
    @ExceptionHandler(RedisBatchWriteException.class)
    public ResponseEntity<Map<String, Object>> handleRedisBatchWriteException(RedisBatchWriteException e) {
        logger.warn("Redis批量写操作异常: {}", e.getMessage());
        
        Map<String, Object> response = new HashMap<>();
        response.put("error", "REDIS_BATCH_WRITE_CHECK_FAILED");
        response.put("message", e.getMessage());
        response.put("timestamp", System.currentTimeMillis());
        
        return ResponseEntity.status(e.getStatusCode()).body(response);
    }
}

6. 实际应用示例

6.1 使用Redis批量写操作注解的服务

/**
 * 使用Redis批量写操作注解的服务
 * @author 运维实战
 */
@Service
public class RedisBatchWriteExampleService {
    
    private static final Logger logger = LoggerFactory.getLogger(RedisBatchWriteExampleService.class);
    
    /**
     * 基础批量写入示例
     * @param dataMap 数据映射
     * @return 处理结果
     */
    @RedisBatchWrite(maxBatchSize = 500, maxMemoryUsageThreshold = 0.7, 
        message = "基础批量写入：总量检查失败")
    public String basicBatchWrite(Map<String, Object> dataMap) {
        logger.info("执行基础批量写入示例，数据量: {}", dataMap.size());
        
        // 模拟Redis批量写入
        try {
            Thread.sleep(dataMap.size() * 10);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        return "基础批量写入完成，数据量: " + dataMap.size();
    }
    
    /**
     * 大批量写入示例
     * @param dataMap 数据映射
     * @return 处理结果
     */
    @RedisBatchWrite(maxBatchSize = 2000, maxMemoryUsageThreshold = 0.9, 
        message = "大批量写入：总量检查失败")
    public String largeBatchWrite(Map<String, Object> dataMap) {
        logger.info("执行大批量写入示例，数据量: {}", dataMap.size());
        
        // 模拟Redis大批量写入
        try {
            Thread.sleep(dataMap.size() * 5);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        return "大批量写入完成，数据量: " + dataMap.size();
    }
    
    /**
     * 严格限制批量写入示例
     * @param dataMap 数据映射
     * @return 处理结果
     */
    @RedisBatchWrite(maxBatchSize = 100, maxMemoryUsageThreshold = 0.5, maxKeyCountThreshold = 500000,
        message = "严格限制批量写入：总量检查失败")
    public String strictBatchWrite(Map<String, Object> dataMap) {
        logger.info("执行严格限制批量写入示例，数据量: {}", dataMap.size());
        
        // 模拟Redis严格限制批量写入
        try {
            Thread.sleep(dataMap.size() * 20);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        return "严格限制批量写入完成，数据量: " + dataMap.size();
    }
}

6.2 Redis批量写操作测试控制器

/**
 * Redis批量写操作测试控制器
 * @author 运维实战
 */
@RestController
@RequestMapping("/api/redis/batch/write/test")
public class RedisBatchWriteTestController {
    
    @Autowired
    private RedisBatchWriteExampleService exampleService;
    
    @Autowired
    private RedisTotalCheckService redisTotalCheckService;
    
    @Autowired
    private RedisBatchWriteOptimizeService optimizeService;
    
    @Autowired
    private RedisBatchWriteMonitorService monitorService;
    
    private static final Logger logger = LoggerFactory.getLogger(RedisBatchWriteTestController.class);
    
    /**
     * 基础批量写入测试
     * @param dataCount 数据数量
     * @return 测试结果
     */
    @GetMapping("/basic")
    public ResponseEntity<Map<String, String>> testBasicBatchWrite(@RequestParam int dataCount) {
        try {
            // 生成测试数据
            Map<String, Object> dataMap = generateTestData(dataCount);
            
            String result = exampleService.basicBatchWrite(dataMap);
            
            Map<String, String> response = new HashMap<>();
            response.put("status", "SUCCESS");
            response.put("result", result);
            response.put("timestamp", String.valueOf(System.currentTimeMillis()));
            
            return ResponseEntity.ok(response);
            
        } catch (RedisBatchWriteException e) {
            logger.warn("基础批量写入测试被限制: {}", e.getMessage());
            return ResponseEntity.status(e.getStatusCode()).build();
        } catch (Exception e) {
            logger.error("基础批量写入测试失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 大批量写入测试
     * @param dataCount 数据数量
     * @return 测试结果
     */
    @GetMapping("/large")
    public ResponseEntity<Map<String, String>> testLargeBatchWrite(@RequestParam int dataCount) {
        try {
            // 生成测试数据
            Map<String, Object> dataMap = generateTestData(dataCount);
            
            String result = exampleService.largeBatchWrite(dataMap);
            
            Map<String, String> response = new HashMap<>();
            response.put("status", "SUCCESS");
            response.put("result", result);
            response.put("timestamp", String.valueOf(System.currentTimeMillis()));
            
            return ResponseEntity.ok(response);
            
        } catch (RedisBatchWriteException e) {
            logger.warn("大批量写入测试被限制: {}", e.getMessage());
            return ResponseEntity.status(e.getStatusCode()).build();
        } catch (Exception e) {
            logger.error("大批量写入测试失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 严格限制批量写入测试
     * @param dataCount 数据数量
     * @return 测试结果
     */
    @GetMapping("/strict")
    public ResponseEntity<Map<String, String>> testStrictBatchWrite(@RequestParam int dataCount) {
        try {
            // 生成测试数据
            Map<String, Object> dataMap = generateTestData(dataCount);
            
            String result = exampleService.strictBatchWrite(dataMap);
            
            Map<String, String> response = new HashMap<>();
            response.put("status", "SUCCESS");
            response.put("result", result);
            response.put("timestamp", String.valueOf(System.currentTimeMillis()));
            
            return ResponseEntity.ok(response);
            
        } catch (RedisBatchWriteException e) {
            logger.warn("严格限制批量写入测试被限制: {}", e.getMessage());
            return ResponseEntity.status(e.getStatusCode()).build();
        } catch (Exception e) {
            logger.error("严格限制批量写入测试失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 获取总量检查结果
     * @param dataCount 数据数量
     * @return 总量检查结果
     */
    @GetMapping("/check/{dataCount}")
    public ResponseEntity<TotalCheckResult> getTotalCheckResult(@PathVariable int dataCount) {
        try {
            TotalCheckResult result = redisTotalCheckService.checkTotal(dataCount);
            return ResponseEntity.ok(result);
        } catch (Exception e) {
            logger.error("获取总量检查结果失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 获取写入策略
     * @param dataCount 数据数量
     * @return 写入策略
     */
    @GetMapping("/strategy/{dataCount}")
    public ResponseEntity<WriteStrategy> getWriteStrategy(@PathVariable int dataCount) {
        try {
            WriteStrategy strategy = optimizeService.getWriteStrategy(dataCount);
            return ResponseEntity.ok(strategy);
        } catch (Exception e) {
            logger.error("获取写入策略失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 获取监控指标
     * @return 监控指标
     */
    @GetMapping("/metrics")
    public ResponseEntity<RedisBatchWriteMetrics> getMetrics() {
        try {
            RedisBatchWriteMetrics metrics = monitorService.getMetrics();
            return ResponseEntity.ok(metrics);
        } catch (Exception e) {
            logger.error("获取监控指标失败", e);
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
        }
    }
    
    /**
     * 生成测试数据
     * @param count 数据数量
     * @return 测试数据
     */
    private Map<String, Object> generateTestData(int count) {
        Map<String, Object> dataMap = new HashMap<>();
        
        for (int i = 0; i < count; i++) {
            String key = "test_key_" + i;
            Map<String, Object> value = new HashMap<>();
            value.put("id", i);
            value.put("name", "test_name_" + i);
            value.put("timestamp", System.currentTimeMillis());
            dataMap.put(key, value);
        }
        
        return dataMap;
    }
}

7. 总结

7.1 Redis批量写操作总量判断最佳实践

合理设置总量阈值: 根据Redis性能和业务需求设置总量阈值
选择合适的检查策略: 根据场景选择内存使用率、键数量等检查策略
监控总量状态: 实时监控Redis总量状态和性能指标
动态调整参数: 根据负载情况动态调整总量检查参数
异常处理: 实现完善的异常处理和用户友好提示

7.2 性能优化建议

智能批量大小: 根据Redis状态智能调整批量大小
分批处理: 使用分批处理避免单次操作过大
监控告警: 建立完善的监控和告警机制
缓存优化: 合理使用缓存减少检查开销
异步处理: 使用异步处理提升系统响应性能

7.3 运维管理要点

实时监控: 监控Redis总量状态和性能指标
动态调整: 根据负载情况动态调整总量检查参数
异常处理: 建立异常处理和告警机制
日志管理: 完善日志记录和分析
性能调优: 根据监控数据优化总量检查参数

通过本文的Redis批量写操作前先判断总量Java实战指南，您可以掌握Redis批量写操作总量判断的原理、实现方法、性能优化技巧以及在企业级应用中的最佳实践，构建高效、稳定的Redis批量写操作系统！