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

前言

需求：当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://stackoverflow.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 的监听触发

最新文章

热门文章