監(jiān)聽Redis實時數(shù)據(jù)變化實現(xiàn)過程

更新時間：2025年10月22日 08:49:21 作者：有女孩說要娶我

本文介紹了通過定時遍歷Redis并對比緩存Map,實現(xiàn)監(jiān)聽指定前綴鍵的value變化的方法,并簡要提及了可用Keyspace通知等更高效的手段

前言

監(jiān)聽Redis的Value實時變化。

一、實現(xiàn)思路

最簡單粗暴的方式，首先遍歷一遍數(shù)據(jù)存到map，然后不斷遍歷redis判斷value是否跟已有的map值相等即可。

二、實現(xiàn)步驟

1. 定義緩存Map

假設(shè)需要監(jiān)聽數(shù)據(jù)key前綴為aaa，創(chuàng)建定時調(diào)度服務(wù)：

private static final Map<String, String> LAST_VALUES = new ConcurrentHashMap<>();
private static final String KEY_PREFIX = "aaa";
private ScheduledExecutorService scheduler;

2. 入口方法

使用@PostConstruct注解執(zhí)行初始化：

    @PostConstruct
    public void init() {
        // 異步加載所有鍵值，不阻塞啟動
        CompletableFuture.runAsync(this::loadInitialValues);

        // 每5秒掃描一次
        scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.scheduleAtFixedRate(this::scanChanges, 0, 5, TimeUnit.SECONDS);
    }

3. 初始化

使用Jedis創(chuàng)建連接，定義掃描參數(shù)，使用redis的scan命令遍歷所有的匹配"aaa"+ "*"模式的鍵，游標(biāo)從0開始，使用do-while，返回為0表示迭代結(jié)束，將所有value存到map：

private void loadInitialValues() {
        try (Jedis jedis = new Jedis("localhost", 6379)) {
            String cursor = "0";
            ScanParams params = new ScanParams().match(KEY_PREFIX  + "*").count(100);
            do {
                ScanResult<String> scanResult = jedis.scan(cursor, params);
                cursor = scanResult.getCursor();
                for (String key : scanResult.getResult()) {
                    String jsonString = jedis.get(key);
                    if (jsonString != null) {
                        LAST_VALUES.put(key, jsonString);
                    }
                }
            } while (!"0".equals(cursor));
            log.info("Loaded {} initial values from Redis.", LAST_VALUES.size());
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

4. 掃描變更數(shù)據(jù)

判斷新舊數(shù)據(jù)是否一致，不一致則通過自定義方法處理數(shù)據(jù)：

private void scanChanges() {
        try (Jedis jedis = new Jedis("localhost", 6379)) {
            String cursor = "0";
            ScanParams params = new ScanParams().match(KEY_PREFIX  + "*").count(100);

            do {
                ScanResult<String> scanResult = jedis.scan(cursor, params);
                cursor = scanResult.getCursor();

                for (String key : scanResult.getResult()) {
                    String currentValueStr = jedis.get(key);
                    String oldValueStr = LAST_VALUES.getOrDefault(key, null);
                    if (!Objects.equals(currentValueStr, oldValueStr)) {
                        try {
                            //處理數(shù)據(jù)
                            dealWithData(key, oldValueStr, currentValueStr );
                        lastValues.put(key, currentValue);
                            LAST_VALUES.put(key, currentValueStr);
                        } catch (Exception e) {
                            log.error("Failed to process data change for key: {}", key, e);
                        }
                    }
                }
            } while (!"0".equals(cursor));
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }