【redis】spring boot利用redis的Keyspace Notifications实现消息通知

2022-08-02 19:20:31

前言

需求：当redis中的某个key失效的时候，把失效时的value写入数据库。

github： https://github.com/vergilyn/RedisSamples

1、修改redis.conf

安装的redis服务默认是: notify-keyspace-events ""，修改成 notify-keyspace-events Ex;

位置：redis安装目下的redis.windows-service.conf 或 redis.windows.conf。(具体看redis服务加载的哪个配置, 貌似要redis 2.8+才支持)

可以在redis.conf中找到对应的描述

# K    键空间通知，以__keyspace@<db>__为前缀

# E    键事件通知，以__keysevent@<db>__为前缀

# g    del , expipre , rename 等类型无关的通用命令的通知, ...

# $    String命令

# l    List命令

# s    Set命令

# h    Hash命令

# z    有序集合命令

# x    过期事件（每次key过期时生成）

# e    驱逐事件（当key在内存满了被清除时生成）

# A    g$lshzxe的别名，因此”AKE”意味着所有的事件

2、通过JedisPubSub实现

省略spring boot配置，完整代码见github。

/**

 * key过期事件推送到topic中只有key，无value，因为一旦过期，value就不存在了。

 */

@Component

public class JedisExpiredListener extends JedisPubSub {

    /** 参考redis目录下redis.conf中的"EVENT NOTIFICATION", redis默认的db{0, 15}一共16个数据库

     * K    Keyspace events, published with __keyspace@<db>__ prefix.

     * E    Keyevent events, published with __keyevent@<db>__ prefix.

     *

     */

    public final static String LISTENER_PATTERN = "__keyevent@*__:expired";

    /**

     * 虽然能注入,但貌似在listener-class中jedis无法使用(无法建立连接到redis),exception message:     * "only (P)SUBSCRIBE / (P)UNSUBSCRIBE / QUIT allowed in this context"

     */

    @Autowired

    private Jedis jedis;

    /**

     * 初始化按表达式的方式订阅时候的处理

     */

    @Override

    public void onPSubscribe(String pattern, int subscribedChannels) {

        System.out.print("onPSubscribe >> ");

        System.out.println(String.format("pattern: %s, subscribedChannels: %d", pattern, subscribedChannels));

    }

    /**

     * 取得按表达式的方式订阅的消息后的处理

     */

    @Override

    public void onPMessage(String pattern, String channel, String message) {

        System.out.print("onPMessage >> ");

        System.out.println(String.format("key: %s, pattern: %s, channel: %s", message, pattern, channel));

    }

    /**

     * 取得订阅的消息后的处理

     */

    @Override

    public void onMessage(String channel, String message) {

        super.onMessage(channel, message);

    }

    /**

     * 初始化订阅时候的处理

     */

    @Override

    public void onSubscribe(String channel, int subscribedChannels) {

        super.onSubscribe(channel, subscribedChannels);

    }

    /**

     * 取消订阅时候的处理

     */

    @Override

    public void onUnsubscribe(String channel, int subscribedChannels) {

        super.onUnsubscribe(channel, subscribedChannels);

    }

    /**

     * 取消按表达式的方式订阅时候的处理

     */

    @Override

    public void onPUnsubscribe(String pattern, int subscribedChannels) {

        super.onPUnsubscribe(pattern, subscribedChannels);

    }

}

@RunWith(SpringRunner.class)

@SpringBootTest(classes=JedisExpiredApplication.class)

public class JedisExpiredApplicationTest {

    @Autowired

    private Jedis jedis;

    @Autowired

    private JedisExpiredListener expiredListener;

    @Before

    public void before() throws Exception {

        jedis.flushAll();

        jedis.set(JedisConfig.DEFAULE_KEY,"123321");

        System.out.println(JedisConfig.DEFAULE_KEY + " = " + jedis.get(JedisConfig.DEFAULE_KEY));

        System.out.println("set expired 5s");

        jedis.expire(JedisConfig.DEFAULE_KEY,5);

    }

    @Test

    public void testPSubscribe(){

        /* psubscribe是一个阻塞的方法，在取消订阅该频道前，会一直阻塞在这，只有当取消了订阅才会执行下面的other code

         * 可以onMessage/onPMessage里面收到消息后，调用了unsubscribe()/onPUnsubscribe(); 来取消订阅，这样才会执行后面的other code

         */

        jedis.psubscribe(expiredListener,JedisExpiredListener.LISTENER_PATTERN);

        // other code

    }

}

输出结果：

vkey = 123321
   set expired 5s
   onPSubscribe >> pattern: __keyevent@*__:expired, subscribedChannels: 1
   onPMessage >> key: vkey, pattern: __keyevent@*__:expired, channel: __keyevent@0__:expired

3、通过实现添加MessageListener

省略spring boot的redis配置。

@SpringBootApplication

public class RedisExpiredApplication implements CommandLineRunner{

    @Autowired

    private RedisTemplate redisTemplate;

    @Autowired

    private RedisExpiredListener expiredListener;

    /**

     * 解决redisTemplate的key/value乱码问题:

     *  <br/> <a href="http://www.zhimengzhe.com/shujuku/other/192111.html">http://www.zhimengzhe.com/shujuku/other/192111.html</a>

     *  <br/> <a href="http://blog.csdn.net/tianyaleixiaowu/article/details/70595073">http://blog.csdn.net/tianyaleixiaowu/article/details/70595073</a>

     * @return

     */

    @Bean("redis")

    @Primary

    public RedisTemplate redisTemplate(){

        RedisSerializer<String> stringSerializer = new StringRedisSerializer();

        redisTemplate.setKeySerializer(stringSerializer);

        redisTemplate.setValueSerializer(stringSerializer);

        redisTemplate.setHashKeySerializer(stringSerializer);

        redisTemplate.setHashValueSerializer(stringSerializer);

        return redisTemplate;

    }

    @Bean

    public RedisMessageListenerContainer listenerContainer(RedisConnectionFactory redisConnection, Executor executor){

        RedisMessageListenerContainer container = new RedisMessageListenerContainer();

        // 设置Redis的连接工厂

        container.setConnectionFactory(redisConnection);

        // 设置监听使用的线程池

//        container.setTaskExecutor(executor);

        // 设置监听的Topic: PatternTopic/ChannelTopic

        Topic topic = new PatternTopic(RedisExpiredListener.LISTENER_PATTERN);

        // 设置监听器

        container.addMessageListener(new RedisExpiredListener(), topic);

        return container;

    }

    @Bean

    public Executor executor(){

        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();

        executor.setCorePoolSize(10);

        executor.setMaxPoolSize(20);

        executor.setQueueCapacity(100);

        executor.setKeepAliveSeconds(60);

        executor.setThreadNamePrefix("V-Thread");

        // rejection-policy：当pool已经达到max size的时候，如何处理新任务

        // CALLER_RUNS：不在新线程中执行任务，而是由调用者所在的线程来执行

        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

        executor.initialize();

        return executor;

    }

    public static void main(String[] args) {

        SpringApplication.run(RedisExpiredApplication.class,args);

    }

    @Override

    public void run(String... strings) throws Exception {

        redisTemplate.opsForValue().set("vkey", "vergilyn", 5, TimeUnit.SECONDS);

        System.out.println("init : set vkey vergilyn ex 5");

        System.out.println("thread sleep: 10s");

        Thread.sleep(10 * 1000);

        System.out.println("thread recover: get vkey = " + redisTemplate.opsForValue().get("vkey"));

    }

}

@Component

public class RedisExpiredListener implements MessageListener {

    public final static String LISTENER_PATTERN = "__key*__:*";

    /**

     * 客户端监听订阅的topic，当有消息的时候，会触发该方法;

     * 并不能得到value, 只能得到key。

     * 姑且理解为: redis服务在key失效时(或失效后)通知到java服务某个key失效了, 那么在java中不可能得到这个redis-key对应的redis-value。

     *      * 解决方案:

     *  创建copy/shadow key, 例如 set vkey "vergilyn"; 对应copykey: set copykey:vkey "" ex 10;

     *  真正的key是"vkey"(业务中使用), 失效触发key是"copykey:vkey"(其value为空字符为了减少内存空间消耗)。

     *  当"copykey:vkey"触发失效时, 从"vkey"得到失效时的值, 并在逻辑处理完后"del vkey"

     *

     * 缺陷:

     *  1: 存在多余的key; (copykey/shadowkey)

     *  2: 不严谨, 假设copykey在 12:00:00失效, 通知在12:10:00收到, 这间隔的10min内程序修改了key, 得到的并不是 失效时的value.

     *  (第1点影响不大; 第2点貌似redis本身的Pub/Sub就不是严谨的, 失效后还存在value的修改, 应该在设计/逻辑上杜绝)

     *  当"copykey:vkey"触发失效时, 从"vkey"得到失效时的值, 并在逻辑处理完后"del vkey"

     *

     */

    @Override

    public void onMessage(Message message, byte[] bytes) {

        byte[] body = message.getBody();// 建议使用: valueSerializer

        byte[] channel = message.getChannel();

        System.out.print("onMessage >> " );

        System.out.println(String.format("channel: %s, body: %s, bytes: %s"

                ,new String(channel), new String(body), new String(bytes)));

    }

}

输出结果：

init : set vkey vergilyn ex 5
   thread sleep: 10s
   onMessage >> channel: __keyevent@0__:expired, body: vkey, bytes: __key*__:*
   thread recover: get vkey = null

4、问题

1) 不管是JedisPubSub，还是MessageListener都不可能得到value。

个人理解：在12:00:00，java推送给redis一条命令”set vkey vergilyn ex 10”。此时redis服务已经完整的知道了这个key的失效时间，在12:00:10时redis服务把”vkey”失效。

然后通知到java(即回调到JedisPubSub/MessageListener)，此时不可能在java中通过”vkey”得到其value。

（最简单的测试，在Listener中打断点，然后通过redis-cli.exe命令查看，“vkey”已经不存在了，但Listener才进入到message()方法）

2) redis的expire不是严格的即时执行

摘自 http://redisdoc.com/topic/notification.html

Redis 使用以下两种方式删除过期的键：

当一个键被访问时，程序会对这个键进行检查，如果键已经过期，那么该键将被删除。

底层系统会在后台渐进地查找并删除那些过期的键，从而处理那些已经过期、但是不会被访问到的键。

当过期键被以上两个程序的任意一个发现、并且将键从数据库中删除时， Redis 会产生一个 expired 通知。

Redis 并不保证生存时间（TTL）变为 0 的键会立即被删除：如果程序没有访问这个过期键，或者带有生存时间的键非常多的话，那么在键的生存时间变为 0 ，直到键真正被删除这中间，可能会有一段比较显著的时间间隔。

因此， Redis 产生 expired 通知的时间为过期键被删除的时候，而不是键的生存时间变为 0 的时候。

如果业务无法容忍从过期到删除中间的时间间隔，那么就只有用其他的方式了。

3) 如何在expire回调中得到expire key的value

参考：https://*.com/questions/26406303/redis-key-expire-notification-with-jedis

set vkey somevalue

set shadowkey:vkey "" ex 10

相当于每个key都有对应的一个shadowkey，”shadowkey”只是用来设置expire时间，”key”才保存value及参与业务逻辑。

所以当”shadowkey”失效通知到listener时，程序中可以通过”key”得到其value，并在逻辑处理完时”del key”。

(“shadowkey”的value为null或空字符串，目的是为了节约内存空间。)

缺陷：

1. 多余了很多无效的 shadowkey；

2. 数据不严谨。假设copykey在 12:00:00失效, 通知在12:10:00收到, 这间隔的10min内程序修改了key, 得到的并不是失效时的value.

相对来说，第1点无关紧要，只是暂时多了一些辅助用的key，但会被程序自己清理掉，不用再去维护，或一直存在于redis缓存中。

第2点，更多的是设计逻辑有缺陷，可以把失效时间定的更长，保证在”那个间隔”内不可能出现失效key的修改。

4) 特别

摘自 http://blog.csdn.net/gqtcgq/article/details/50808729

Redis的发布/订阅目前是即发即弃(fire and forget)模式的，因此无法实现事件的可靠通知。也就是说，如果发布/订阅的客户端断链之后又重连，则在客户端断链期间的所有事件都丢失了。

未来计划支持事件的可靠通知，但是这可能会通过让订阅与发布功能本身变得更可靠来实现，也可能会在Lua脚本中对消息的订阅与发布进行监听，从而实现类似将事件推入到列表这样的操作。

码农公寓

【redis】spring boot利用redis的Keyspace Notifications实现消息通知

前言

1、修改redis.conf

2、通过JedisPubSub实现

3、通过实现添加MessageListener

4、问题

参考：

redis设置键的生存时间或过期时间

Redis Key expire notification with Jedis

JAVA实现redis超时失效key 的监听触发

redis 超时失效key 的监听触发

码农公寓

前言

1、修改redis.conf

2、通过JedisPubSub实现

3、通过实现添加MessageListener

4、问题

参考：

redis设置键的生存时间或过期时间

Redis Key expire notification with Jedis

JAVA实现redis超时失效key 的监听触发

redis 超时失效key 的监听触发

相关文章