import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

public class RedisShardingStrategy {
    // 原來(lái)：一個(gè)大 Redis 存所有數(shù)據(jù)
    // JedisPool redisAll = new JedisPool("10.0.0.1", 6379);  // 64GB
    
    // 現(xiàn)在：按數(shù)據(jù)特性拆分
    private JedisPool redisHot;   // 16GB 熱數(shù)據(jù)
    private JedisPool redisWarm;  // 32GB 溫?cái)?shù)據(jù)
    private JedisPool redisCold;  // 16GB 冷數(shù)據(jù)
    
    public void init() {
        JedisPoolConfig config = new JedisPoolConfig();
        config.setMaxTotal(100);
        config.setMaxIdle(20);
        
        redisHot = new JedisPool(config, "10.0.0.2", 6379);
        redisWarm = new JedisPool(config, "10.0.0.3", 6379);
        redisCold = new JedisPool(config, "10.0.0.4", 6379);
        
        // 熱數(shù)據(jù)：頻繁訪問(wèn)，很少持久化
        try (Jedis jedis = redisHot.getResource()) {
            jedis.configSet("save", "");  // 關(guān)閉自動(dòng)保存
        }
        
        // 溫?cái)?shù)據(jù)：定期持久化
        try (Jedis jedis = redisWarm.getResource()) {
            jedis.configSet("save", "1800 1");  // 30分鐘
        }
        
        // 冷數(shù)據(jù)：積極持久化
        try (Jedis jedis = redisCold.getResource()) {
            jedis.configSet("save", "300 10");  // 5分鐘
        }
    }
}

# 查看當(dāng)前透明大頁(yè)設(shè)置
cat /sys/kernel/mm/transparent_hugepage/enabled

# 建議設(shè)為 madvise（讓 Redis 自己決定）
echo madvise > /sys/kernel/mm/transparent_hugepage/enabled

# 預(yù)分配大頁(yè)
echo 20480 > /proc/sys/vm/nr_hugepages  # 20480 * 2MB = 40GB

import redis.clients.jedis.*;
import java.util.*;
import java.util.concurrent.TimeUnit;

public class COWFriendlyRedis {
    private JedisPool jedisPool;
    private boolean inCow = false;
    
    public COWFriendlyRedis(String host, int port) {
        JedisPoolConfig config = new JedisPoolConfig();
        config.setMaxTotal(100);
        this.jedisPool = new JedisPool(config, host, port);
    }
    
    public void batchWrite(Map<String, String> dataMap, boolean cowFriendly) throws InterruptedException {
        if (!cowFriendly) {
            // 傳統(tǒng)方式：直接寫(xiě)
            try (Jedis jedis = jedisPool.getResource()) {
                Pipeline pipe = jedis.pipelined();
                for (Map.Entry<String, String> entry : dataMap.entrySet()) {
                    pipe.set(entry.getKey(), entry.getValue());
                }
                pipe.sync();
                return;
            }
        }
        
        // COW 友好方式：檢測(cè)并避讓
        try (Jedis jedis = jedisPool.getResource()) {
            // 1. 檢查是否正在做持久化
            String persistence = jedis.info("persistence");
            boolean rdbInProgress = persistence.contains("rdb_bgsave_in_progress:1");
            boolean aofInProgress = persistence.contains("aof_rewrite_in_progress:1");
            
            if (rdbInProgress || aofInProgress) {
                System.out.println("檢測(cè)到持久化進(jìn)行中，降速寫(xiě)入...");
                
                // 降速寫(xiě)入，每寫(xiě)入一些就休眠
                int count = 0;
                for (Map.Entry<String, String> entry : dataMap.entrySet()) {
                    jedis.set(entry.getKey(), entry.getValue());
                    if (++count % 100 == 0) {  // 每100個(gè)key休眠一下
                        TimeUnit.MILLISECONDS.sleep(1);
                    }
                }
            } else {
                // 2. 非持久化期間，集中寫(xiě)入
                // 先關(guān)閉自動(dòng)持久化
                String oldSave = jedis.configGet("save").get(1);
                jedis.configSet("save", "");
                
                // 3. 批量寫(xiě)入
                Pipeline pipe = jedis.pipelined();
                for (Map.Entry<String, String> entry : dataMap.entrySet()) {
                    pipe.set(entry.getKey(), entry.getValue());
                }
                pipe.sync();
                
                // 4. 手動(dòng)觸發(fā)一次持久化
                jedis.bgsave();
                
                // 5. 恢復(fù)自動(dòng)持久化配置
                TimeUnit.SECONDS.sleep(1);  // 等 BGSAVE 開(kāi)始
                jedis.configSet("save", oldSave);
            }
        }
    }
}

# 1. 限制 AOF 重寫(xiě)頻率
auto-aof-rewrite-percentage 200  # 文件大小翻倍才重寫(xiě)（默認(rèn)100）
auto-aof-rewrite-min-size 1gb    # 最小 1GB 才考慮重寫(xiě)

# 2. 重寫(xiě)期間不做 fsync（危險(xiǎn)但有效）
no-appendfsync-on-rewrite yes

# 3. 使用 RDB+AOF 混合持久化（Redis 4.0+）
aof-use-rdb-preamble yes

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import java.util.concurrent.*;
import java.util.regex.*;

public class AOFRewriteGuard {
    private final JedisPool jedisPool;
    private final int maxBufferMB;
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    
    public AOFRewriteGuard(JedisPool jedisPool, int maxBufferMB) {
        this.jedisPool = jedisPool;
        this.maxBufferMB = maxBufferMB;
    }
    
    public void startMonitoring() {
        scheduler.scheduleAtFixedRate(this::checkAOFRewrite, 0, 5, TimeUnit.SECONDS);
    }
    
    private void checkAOFRewrite() {
        try (Jedis jedis = jedisPool.getResource()) {
            String info = jedis.info("persistence");
            
            // 解析持久化信息
            Pattern inProgressPattern = Pattern.compile("aof_rewrite_in_progress:(\\d)");
            Pattern bufferLengthPattern = Pattern.compile("aof_rewrite_buffer_length:(\\d+)");
            
            Matcher inProgressMatcher = inProgressPattern.matcher(info);
            if (inProgressMatcher.find() && "1".equals(inProgressMatcher.group(1))) {
                // 正在重寫(xiě)，檢查緩沖區(qū)大小
                Matcher bufferMatcher = bufferLengthPattern.matcher(info);
                if (bufferMatcher.find()) {
                    long bufferSize = Long.parseLong(bufferMatcher.group(1));
                    double bufferMB = bufferSize / 1024.0 / 1024.0;
                    
                    if (bufferMB > maxBufferMB) {
                        // 緩沖區(qū)過(guò)大，緊急措施
                        System.err.printf("AOF重寫(xiě)緩沖區(qū)過(guò)大: %.2fMB%n", bufferMB);
                        
                        // 1. 通知應(yīng)用層限流
                        publishAlert("aof_buffer_overflow", bufferMB);
                        
                        // 2. 如果超過(guò) 2 倍閾值，考慮中止重寫(xiě)
                        if (bufferMB > maxBufferMB * 2) {
                            // 這是個(gè)危險(xiǎn)操作，記錄日志
                            System.err.println("警告：AOF緩沖區(qū)超過(guò)2倍閾值，考慮中止重寫(xiě)");
                            // 實(shí)際生產(chǎn)環(huán)境中，這里應(yīng)該發(fā)送告警而不是自動(dòng)中止
                            // jedis.configSet("appendonly", "no");
                            // Thread.sleep(1000);
                            // jedis.configSet("appendonly", "yes");
                        }
                    }
                }
            }
        } catch (Exception e) {
            System.err.println("AOF監(jiān)控異常: " + e.getMessage());
        }
    }
    
    private void publishAlert(String type, double bufferMB) {
        // 實(shí)際環(huán)境中這里應(yīng)該接入告警系統(tǒng)
        System.err.printf("告警: %s, 緩沖區(qū)大小: %.2fMB%n", type, bufferMB);
    }
}

# 占用雙倍內(nèi)存
redis-cli --rdb /backup/dump.rdb

import redis.clients.jedis.Jedis;
import java.io.*;
import java.util.regex.*;
import java.util.concurrent.TimeUnit;

public class COWBackupUtil {
    
    public static void cowBackup(Jedis jedis, String backupPath) throws Exception {
        // 1. 觸發(fā) BGSAVE
        String result = jedis.bgsave();
        System.out.println("BGSAVE 觸發(fā): " + result);
        
        // 2. 等待 BGSAVE 開(kāi)始（不是結(jié)束）
        long childPid = -1;
        Pattern pidPattern = Pattern.compile("rdb_last_bgsave_pid:(\\d+)");
        Pattern progressPattern = Pattern.compile("rdb_bgsave_in_progress:(\\d)");
        
        while (childPid == -1) {
            String info = jedis.info("persistence");
            
            Matcher progressMatcher = progressPattern.matcher(info);
            if (progressMatcher.find() && "1".equals(progressMatcher.group(1))) {
                Matcher pidMatcher = pidPattern.matcher(info);
                if (pidMatcher.find()) {
                    childPid = Long.parseLong(pidMatcher.group(1));
                }
            }
            TimeUnit.MILLISECONDS.sleep(100);
        }
        
        System.out.println("BGSAVE 子進(jìn)程 PID: " + childPid);
        
        // 3. 這時(shí)子進(jìn)程已經(jīng) fork 完成，內(nèi)存是 COW 共享的
        // 可以慢慢導(dǎo)出數(shù)據(jù)，不影響主進(jìn)程
        
        // 4. 通過(guò) /proc 讀取子進(jìn)程的內(nèi)存映射（需要 Linux 環(huán)境）
        File mapsFile = new File("/proc/" + childPid + "/maps");
        if (mapsFile.exists()) {
            try (BufferedReader reader = new BufferedReader(new FileReader(mapsFile))) {
                String line;
                System.out.println("內(nèi)存映射信息:");
                while ((line = reader.readLine()) != null) {
                    // 找到堆區(qū)域（通常包含 Redis 數(shù)據(jù)）
                    if (line.contains("[heap]")) {
                        System.out.println("堆區(qū)域: " + line);
                    }
                }
            }
        }
        
        // 5. 等待 BGSAVE 完成，然后復(fù)制 dump.rdb 文件
        Pattern lastSavePattern = Pattern.compile("rdb_last_bgsave_status:ok");
        while (true) {
            String info = jedis.info("persistence");
            if (lastSavePattern.matcher(info).find()) {
                // BGSAVE 完成，復(fù)制文件
                File sourceFile = new File(jedis.configGet("dir").get(1) + "/" + 
                                          jedis.configGet("dbfilename").get(1));
                File destFile = new File(backupPath);
                
                try (FileInputStream fis = new FileInputStream(sourceFile);
                     FileOutputStream fos = new FileOutputStream(destFile)) {
                    byte[] buffer = new byte[8192];
                    int bytesRead;
                    while ((bytesRead = fis.read(buffer)) != -1) {
                        fos.write(buffer, 0, bytesRead);
                    }
                }
                
                System.out.println("COW 備份完成: " + backupPath);
                break;
            }
            TimeUnit.SECONDS.sleep(1);
        }
    }
}

# ========== 內(nèi)存管理 ==========
maxmemory 32gb
maxmemory-policy volatile-lru
# 預(yù)留 25% 內(nèi)存給 COW
maxmemory-reserved 8gb  # 僅 Redis 7.0+

# ========== RDB 配置 ==========
# 分層保存策略
save 3600 1      # 1小時(shí)內(nèi)至少1個(gè)鍵改變
save 1800 100    # 30分鐘內(nèi)至少100個(gè)鍵改變
save 300 10000   # 5分鐘內(nèi)至少10000個(gè)鍵改變

# 壓縮和校驗(yàn)
rdbcompression yes
rdbchecksum yes

# BGSAVE 失敗不停止寫(xiě)入（危險(xiǎn)但實(shí)用）
stop-writes-on-bgsave-error no

# ========== AOF 配置 ==========
appendonly yes
appendfilename "redis.aof"

# 同步策略（everysec 是最佳平衡）
appendfsync everysec

# AOF 重寫(xiě)
auto-aof-rewrite-percentage 100
auto-aof-rewrite-min-size 2gb
no-appendfsync-on-rewrite yes

# 混合持久化（強(qiáng)烈推薦）
aof-use-rdb-preamble yes

# ========== 系統(tǒng)層面 ==========
# /etc/sysctl.conf
vm.overcommit_memory=1        # 允許 overcommit
vm.overcommit_ratio=100       # 可以用到 100% 物理內(nèi)存
vm.swappiness=1               # 盡量不用 swap
vm.dirty_background_ratio=5   # 后臺(tái)開(kāi)始刷盤(pán)的臟頁(yè)比例
vm.dirty_ratio=10             # 前臺(tái)強(qiáng)制刷盤(pán)的臟頁(yè)比例

# 透明大頁(yè)（THP）
echo madvise > /sys/kernel/mm/transparent_hugepage/enabled
echo never > /sys/kernel/mm/transparent_hugepage/defrag

import redis.clients.jedis.Jedis;
import java.io.*;
import java.util.*;
import java.util.regex.*;

public class RedisCoWMonitor {
    private final Jedis redis;
    private final Map<String, Double> metrics = new HashMap<>();
    
    public RedisCoWMonitor(Jedis redis) {
        this.redis = redis;
    }
    
    public Map<String, Double> collectMetrics() {
        String info = redis.info();
        Map<String, String> infoMap = parseInfo(info);
        
        // 核心指標(biāo)
        metrics.put("cow_overhead", calculateCowOverhead(infoMap));
        metrics.put("fork_time_ms", 
            Double.parseDouble(infoMap.getOrDefault("latest_fork_usec", "0")) / 1000);
        metrics.put("fragmentation", 
            Double.parseDouble(infoMap.getOrDefault("mem_fragmentation_ratio", "1.0")));
        
        // 持久化相關(guān)
        String persistInfo = redis.info("persistence");
        Map<String, String> persistMap = parseInfo(persistInfo);
        metrics.put("rdb_in_progress", 
            Double.parseDouble(persistMap.getOrDefault("rdb_bgsave_in_progress", "0")));
        metrics.put("aof_rewrite_in_progress", 
            Double.parseDouble(persistMap.getOrDefault("aof_rewrite_in_progress", "0")));
        metrics.put("aof_buffer_length", 
            Double.parseDouble(persistMap.getOrDefault("aof_rewrite_buffer_length", "0")));
        
        // 內(nèi)存使用
        double usedMemory = Double.parseDouble(infoMap.getOrDefault("used_memory", "0"));
        double usedMemoryRss = Double.parseDouble(infoMap.getOrDefault("used_memory_rss", "0"));
        
        metrics.put("used_memory_gb", usedMemory / Math.pow(1024, 3));
        metrics.put("used_memory_rss_gb", usedMemoryRss / Math.pow(1024, 3));
        metrics.put("memory_available", (double) getSystemAvailableMemory());
        
        return metrics;
    }
    
    private Map<String, String> parseInfo(String info) {
        Map<String, String> result = new HashMap<>();
        String[] lines = info.split("\n");
        for (String line : lines) {
            if (line.contains(":")) {
                String[] parts = line.split(":", 2);
                result.put(parts[0].trim(), parts[1].trim());
            }
        }
        return result;
    }
    
    private double calculateCowOverhead(Map<String, String> infoMap) {
        double used = Double.parseDouble(infoMap.getOrDefault("used_memory", "0"));
        double rss = Double.parseDouble(infoMap.getOrDefault("used_memory_rss", "0"));
        double fragmentation = Double.parseDouble(
            infoMap.getOrDefault("mem_fragmentation_ratio", "1.0"));
        
        // RSS 包含：實(shí)際使用 + COW復(fù)制 + 內(nèi)存碎片
        // 減去碎片部分，剩下的主要是 COW
        double cowOverhead = rss - used * fragmentation;
        
        return Math.max(0, cowOverhead);  // 可能是負(fù)數(shù)，取 0
    }
    
    public List<String> alertIfNeeded() {
        List<String> alerts = new ArrayList<>();
        
        // COW 開(kāi)銷超過(guò) 50%
        double usedMemoryBytes = metrics.get("used_memory_gb") * Math.pow(1024, 3);
        if (metrics.get("cow_overhead") > usedMemoryBytes * 0.5) {
            alerts.add("COW 內(nèi)存開(kāi)銷過(guò)大");
        }
        
        // Fork 時(shí)間超過(guò) 200ms
        if (metrics.get("fork_time_ms") > 200) {
            alerts.add(String.format("Fork 耗時(shí)過(guò)長(zhǎng): %.2fms", metrics.get("fork_time_ms")));
        }
        
        // 內(nèi)存碎片嚴(yán)重
        if (metrics.get("fragmentation") > 1.5) {
            alerts.add(String.format("內(nèi)存碎片率過(guò)高: %.2f", metrics.get("fragmentation")));
        }
        
        // AOF 重寫(xiě)緩沖區(qū)過(guò)大
        if (metrics.get("aof_buffer_length") > Math.pow(1024, 3)) {  // 1GB
            double bufferGb = metrics.get("aof_buffer_length") / Math.pow(1024, 3);
            alerts.add(String.format("AOF 重寫(xiě)緩沖區(qū)過(guò)大: %.2fGB", bufferGb));
        }
        
        // 可用內(nèi)存不足
        if (metrics.get("memory_available") < metrics.get("used_memory_gb") * 0.3 * Math.pow(1024, 3)) {
            alerts.add("系統(tǒng)可用內(nèi)存不足，COW 可能失敗");
        }
        
        return alerts;
    }
    
    private long getSystemAvailableMemory() {
        // Linux 系統(tǒng)獲取可用內(nèi)存
        try (BufferedReader reader = new BufferedReader(new FileReader("/proc/meminfo"))) {
            String line;
            while ((line = reader.readLine()) != null) {
                if (line.startsWith("MemAvailable:")) {
                    String[] parts = line.split("\\s+");
                    return Long.parseLong(parts[1]) * 1024;  // KB to Bytes
                }
            }
        } catch (IOException e) {
            System.err.println("無(wú)法讀取系統(tǒng)內(nèi)存信息: " + e.getMessage());
        }
        return 0;
    }
}

# 關(guān)閉持久化，完全避免 COW
save ""
appendonly no

# 或者只做最低限度的持久化
save 7200 1  # 2小時(shí)
appendonly no

# RDB 足夠，AOF 太重
save 900 1 300 10 60 10000
appendonly no

# 用主從復(fù)制代替 AOF
# 從節(jié)點(diǎn)可以更激進(jìn)地做持久化

# AOF 為主，RDB 為輔
appendonly yes
appendfsync everysec
save 1800 1

# 考慮使用 Redis Streams + Consumer Groups
# 自帶持久化和 ack 機(jī)制

import redis.clients.jedis.*;
import java.util.Set;
import java.util.concurrent.*;

// 時(shí)間窗口策略
public class TimeWindowRedis {
    private final JedisPool currentPool;  // 當(dāng)前窗口（不持久化）
    private final JedisPool historyPool;  // 歷史窗口（定期持久化）
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    
    public TimeWindowRedis() {
        JedisPoolConfig config = new JedisPoolConfig();
        config.setMaxTotal(50);
        
        // 雙緩沖：當(dāng)前窗口 + 歷史窗口
        this.currentPool = new JedisPool(config, "localhost", 6379);
        this.historyPool = new JedisPool(config, "localhost", 6380);
        
        // 配置不同的持久化策略
        try (Jedis current = currentPool.getResource()) {
            current.configSet("save", "");  // 不持久化
        }
        
        try (Jedis history = historyPool.getResource()) {
            history.configSet("save", "1800 1");  // 30分鐘持久化
        }
        
        // 每小時(shí)輪轉(zhuǎn)一次窗口
        scheduler.scheduleAtFixedRate(this::rotateWindow, 1, 1, TimeUnit.HOURS);
    }
    
    public void write(String key, String value) {
        // 寫(xiě)入當(dāng)前窗口
        try (Jedis jedis = currentPool.getResource()) {
            jedis.set(key, value);
        }
    }
    
    public void rotateWindow() {
        System.out.println("開(kāi)始窗口輪轉(zhuǎn)...");
        
        try (Jedis current = currentPool.getResource();
             Jedis history = historyPool.getResource()) {
            
            // 1. 當(dāng)前窗口數(shù)據(jù)導(dǎo)入歷史
            ScanParams scanParams = new ScanParams().count(100);
            String cursor = "0";
            
            do {
                ScanResult<String> scanResult = current.scan(cursor, scanParams);
                for (String key : scanResult.getResult()) {
                    String value = current.get(key);
                    if (value != null) {
                        history.set(key, value);
                    }
                }
                cursor = scanResult.getCursor();
            } while (!"0".equals(cursor));
            
            // 2. 清空當(dāng)前窗口
            current.flushDB();
            
            // 3. 歷史窗口做持久化
            history.bgsave();
            
            System.out.println("窗口輪轉(zhuǎn)完成");
        } catch (Exception e) {
            System.err.println("窗口輪轉(zhuǎn)失敗: " + e.getMessage());
        }
    }
    
    public void shutdown() {
        scheduler.shutdown();
        currentPool.close();
        historyPool.close();
    }
}

import org.springframework.stereotype.Service;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.beans.factory.annotation.Autowired;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.Jedis;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Service
public class RedisCowService {
    private static final Logger logger = LoggerFactory.getLogger(RedisCowService.class);
    
    @Autowired
    private JedisPool jedisPool;
    
    // 每分鐘檢查一次 COW 狀態(tài)
    @Scheduled(fixedDelay = 60000)
    public void monitorCowStatus() {
        try (Jedis jedis = jedisPool.getResource()) {
            String info = jedis.info();
            
            // 解析關(guān)鍵指標(biāo)
            long usedMemory = parseInfoLong(info, "used_memory:");
            long usedMemoryRss = parseInfoLong(info, "used_memory_rss:");
            double fragRatio = parseInfoDouble(info, "mem_fragmentation_ratio:");
            
            // 計(jì)算 COW 壓力指數(shù)
            double cowPressure = (usedMemoryRss - usedMemory) / (double) usedMemory;
            
            if (cowPressure > 0.3) {
                logger.warn("COW內(nèi)存壓力過(guò)高: {}%, 考慮優(yōu)化寫(xiě)入模式", cowPressure * 100);
                
                // 動(dòng)態(tài)調(diào)整策略
                adjustStrategy(jedis, cowPressure);
            }
            
            // 記錄指標(biāo)到監(jiān)控系統(tǒng)（Prometheus/Grafana）
            recordMetrics(usedMemory, usedMemoryRss, fragRatio, cowPressure);
        }
    }
    
    private void adjustStrategy(Jedis jedis, double cowPressure) {
        if (cowPressure > 0.5) {
            // 壓力極高，臨時(shí)關(guān)閉自動(dòng)持久化
            jedis.configSet("save", "");
            logger.warn("COW壓力極高，已臨時(shí)關(guān)閉自動(dòng)持久化");
            
            // 通知運(yùn)維人員
            sendAlert("Redis COW壓力超過(guò)50%，請(qǐng)立即檢查");
        } else if (cowPressure > 0.3) {
            // 壓力較高，延長(zhǎng)持久化間隔
            jedis.configSet("save", "3600 1");  // 改為1小時(shí)
            logger.info("調(diào)整持久化間隔為1小時(shí)");
        }
    }
    
    private long parseInfoLong(String info, String key) {
        int start = info.indexOf(key);
        if (start == -1) return 0;
        start += key.length();
        int end = info.indexOf('\r', start);
        if (end == -1) end = info.indexOf('\n', start);
        return Long.parseLong(info.substring(start, end));
    }
    
    private double parseInfoDouble(String info, String key) {
        int start = info.indexOf(key);
        if (start == -1) return 0;
        start += key.length();
        int end = info.indexOf('\r', start);
        if (end == -1) end = info.indexOf('\n', start);
        return Double.parseDouble(info.substring(start, end));
    }
    
    private void recordMetrics(long usedMemory, long usedMemoryRss, 
                               double fragRatio, double cowPressure) {
        // 這里接入你的監(jiān)控系統(tǒng)
        // 比如 Micrometer、Prometheus 等
    }
    
    private void sendAlert(String message) {
        // 接入告警系統(tǒng)：釘釘、企業(yè)微信、PagerDuty 等
        logger.error("告警: " + message);
    }
}

public class LuaOptimizedRedis {
    private final JedisPool jedisPool;
    
    // 批量更新的 Lua 腳本
    private static final String BATCH_UPDATE_SCRIPT = 
        "local count = 0\n" +
        "for i = 1, #KEYS do\n" +
        "    redis.call('SET', KEYS[i], ARGV[i])\n" +
        "    count = count + 1\n" +
        "    -- 每100個(gè)key做一次內(nèi)存整理\n" +
        "    if count % 100 == 0 then\n" +
        "        redis.call('MEMORY', 'PURGE')\n" +
        "    end\n" +
        "end\n" +
        "return count";
    
    // 條件性持久化的 Lua 腳本    
    private static final String CONDITIONAL_SAVE_SCRIPT = 
        "local info = redis.call('INFO', 'memory')\n" +
        "local used = tonumber(string.match(info, 'used_memory:(%d+)'))\n" +
        "local total = tonumber(string.match(info, 'total_system_memory:(%d+)'))\n" +
        "if used / total < 0.7 then\n" +
        "    redis.call('BGSAVE')\n" +
        "    return 'BGSAVE triggered'\n" +
        "else\n" +
        "    return 'Memory usage too high, skip BGSAVE'\n" +
        "end";
    
    public LuaOptimizedRedis(JedisPool jedisPool) {
        this.jedisPool = jedisPool;
    }
    
    public long batchUpdate(Map<String, String> data) {
        try (Jedis jedis = jedisPool.getResource()) {
            List<String> keys = new ArrayList<>(data.keySet());
            List<String> values = new ArrayList<>();
            for (String key : keys) {
                values.add(data.get(key));
            }
            
            // 執(zhí)行 Lua 腳本，原子操作減少內(nèi)存碎片
            Object result = jedis.eval(BATCH_UPDATE_SCRIPT, keys, values);
            return (Long) result;
        }
    }
    
    public String conditionalSave() {
        try (Jedis jedis = jedisPool.getResource()) {
            // 只在內(nèi)存使用率低于70%時(shí)觸發(fā)持久化
            return (String) jedis.eval(CONDITIONAL_SAVE_SCRIPT, 0);
        }
    }
}

import java.util.concurrent.*;
import java.util.*;
import redis.clients.jedis.*;

public class MassiveDataCowOptimizer {
    private final List<JedisPool> shardPools = new ArrayList<>();
    private final int SHARD_COUNT = 16;  // 16個(gè)分片
    private final ExecutorService executor = Executors.newFixedThreadPool(32);
    
    public MassiveDataCowOptimizer() {
        // 初始化16個(gè)Redis分片，每個(gè)2GB
        for (int i = 0; i < SHARD_COUNT; i++) {
            JedisPoolConfig config = new JedisPoolConfig();
            config.setMaxTotal(10);
            config.setMaxIdle(5);
            
            JedisPool pool = new JedisPool(config, "redis-shard-" + i, 6379);
            shardPools.add(pool);
            
            // 每個(gè)分片不同的持久化策略
            try (Jedis jedis = pool.getResource()) {
                if (i < 4) {
                    // 前4個(gè)分片存熱數(shù)據(jù)，不持久化
                    jedis.configSet("save", "");
                } else if (i < 12) {
                    // 中間8個(gè)分片，適度持久化
                    jedis.configSet("save", "1800 100");
                } else {
                    // 最后4個(gè)分片存冷數(shù)據(jù)，頻繁持久化
                    jedis.configSet("save", "300 10");
                }
            }
        }
    }
    
    // 一致性哈希分片
    private int getShard(String key) {
        return Math.abs(key.hashCode()) % SHARD_COUNT;
    }
    
    // 并行批量寫(xiě)入，最小化COW影響
    public void parallelBatchWrite(Map<String, String> bigData) throws InterruptedException {
        // 按分片分組數(shù)據(jù)
        Map<Integer, Map<String, String>> shardedData = new HashMap<>();
        for (Map.Entry<String, String> entry : bigData.entrySet()) {
            int shard = getShard(entry.getKey());
            shardedData.computeIfAbsent(shard, k -> new HashMap<>())
                      .put(entry.getKey(), entry.getValue());
        }
        
        // 并行寫(xiě)入各分片
        CountDownLatch latch = new CountDownLatch(shardedData.size());
        
        for (Map.Entry<Integer, Map<String, String>> entry : shardedData.entrySet()) {
            final int shardId = entry.getKey();
            final Map<String, String> shardData = entry.getValue();
            
            executor.submit(() -> {
                try {
                    writeToShard(shardId, shardData);
                } finally {
                    latch.countDown();
                }
            });
        }
        
        latch.await(30, TimeUnit.SECONDS);
    }
    
    private void writeToShard(int shardId, Map<String, String> data) {
        JedisPool pool = shardPools.get(shardId);
        
        try (Jedis jedis = pool.getResource()) {
            // 檢查當(dāng)前分片的內(nèi)存狀態(tài)
            String info = jedis.info("memory");
            long usedMemory = parseMemory(info, "used_memory:");
            
            if (usedMemory > 1_500_000_000) {  // 超過(guò)1.5GB
                // 內(nèi)存過(guò)高，分批寫(xiě)入
                Pipeline pipe = jedis.pipelined();
                int count = 0;
                
                for (Map.Entry<String, String> entry : data.entrySet()) {
                    pipe.set(entry.getKey(), entry.getValue());
                    
                    if (++count % 100 == 0) {
                        pipe.sync();
                        // 讓COW有機(jī)會(huì)處理
                        Thread.sleep(1);
                        pipe = jedis.pipelined();
                    }
                }
                
                if (count % 100 != 0) {
                    pipe.sync();
                }
            } else {
                // 內(nèi)存充足，一次性寫(xiě)入
                Pipeline pipe = jedis.pipelined();
                for (Map.Entry<String, String> entry : data.entrySet()) {
                    pipe.set(entry.getKey(), entry.getValue());
                }
                pipe.sync();
            }
        } catch (Exception e) {
            System.err.println("分片" + shardId + "寫(xiě)入失敗: " + e.getMessage());
        }
    }
    
    // 智能持久化調(diào)度
    public void smartPersistence() {
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
        
        scheduler.scheduleAtFixedRate(() -> {
            for (int i = 0; i < SHARD_COUNT; i++) {
                final int shardId = i;
                executor.submit(() -> checkAndPersist(shardId));
            }
        }, 0, 5, TimeUnit.MINUTES);
    }
    
    private void checkAndPersist(int shardId) {
        JedisPool pool = shardPools.get(shardId);
        
        try (Jedis jedis = pool.getResource()) {
            String info = jedis.info();
            
            // 解析關(guān)鍵指標(biāo)
            boolean rdbInProgress = info.contains("rdb_bgsave_in_progress:1");
            long changedKeys = parseMemory(info, "rdb_changes_since_last_save:");
            
            // 根據(jù)變化量決定是否持久化
            if (!rdbInProgress && changedKeys > 10000) {
                // 檢查系統(tǒng)負(fù)載
                double loadAvg = getSystemLoadAverage();
                
                if (loadAvg < 2.0) {  // 負(fù)載低時(shí)才持久化
                    jedis.bgsave();
                    System.out.println("分片" + shardId + "觸發(fā)持久化，變更鍵數(shù): " + changedKeys);
                }
            }
        }
    }
    
    private long parseMemory(String info, String key) {
        int idx = info.indexOf(key);
        if (idx == -1) return 0;
        idx += key.length();
        int end = info.indexOf('\n', idx);
        if (end == -1) end = info.length();
        return Long.parseLong(info.substring(idx, end).trim());
    }
    
    private double getSystemLoadAverage() {
        return ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage();
    }
}