监听Redis实时数据变化实现过程

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

    @PostConstruct
    public void init() {
        // 异步加载所有键值，不阻塞启动
        CompletableFuture.runAsync(this::loadInitialValues);

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

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);
        }
    }

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 {
                            //处理数据
                            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);
        }
    }