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

前言

监听redis的value实时变化。

一、实现思路

最简单粗暴的方式，首先遍历一遍数据存到map，然后不断遍历redis判断value是否跟已有的map值相等即可。

二、实现步骤

1. 定义缓存map

假设需要监听数据key前缀为aaa，创建定时调度服务：

private static final map<string, string> last_values = new concurrenthashmap<>();
private static final string key_prefix = "aaa";
private scheduledexecutorservice scheduler;

2. 入口方法

使用@postconstruct注解执行初始化：

    @postconstruct
    public void init() {
        // 异步加载所有键值，不阻塞启动
        completablefuture.runasync(this::loadinitialvalues);

        // 每5秒扫描一次
        scheduler = executors.newsinglethreadscheduledexecutor();
        scheduler.scheduleatfixedrate(this::scanchanges, 0, 5, timeunit.seconds);
    }

3. 初始化

使用jedis创建连接，定义扫描参数，使用redis的scan命令遍历所有的匹配"aaa"+ "*"模式的键，游标从0开始，使用do-while，返回为0表示迭代结束，将所有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. 扫描变更数据

判断新旧数据是否一致，不一致则通过自定义方法处理数据：

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