Spring Boot集成Redis实现keyspace监听 | Spring Cloud 34

一、前言

在前面我们通过以下章节对Redis的keyevent（键事件通知）使用有了基础的了解：

Spring Boot集成Redis实现keyevent监听 | Spring Cloud 33

现在开始我们正式学习Redis的keyspace（键空间通知），在本章节主要进行对以下部分讲解说明：

• keyspace（键空间通知）介绍
• Redis事件通知开启方式及配置说明
• keyspace各种事件通知的产生
• keyspace监听的redis-cli客户端验证
• Spring Boot集成Redis实现keyspace监听
• RedisMessageListenerContainer 使用自定义线程池

二、Redis事件通知

2.1 keyspace介绍

keyspace（键空间通知）针对指定key发生的一切改动，推送给订阅的客户端，侧重于针对指定key的操作

键空间通知监听格式：

__keyspace@<db>__:<key>

2.2 Redis事件通知配置

因为Redis事件通知功能需要消耗一定的资源，默认情况下该功能的关闭的。

• 使用config set命令来临时开启或关闭键空间通知功能（此方式Redis重启将会被还原）

• 通过修改redis.conf配置，永久开启此功能

notify-keyspace-events可以是以下任意组合：

字符	发送的通知
K	键空间通知，所有通知以 __keyspace@<db>__ 为前缀
E	键事件通知，所有通知以 __keyevent@<db>__ 为前缀
g	del、expire、rename 等类型无关的通用命令的通知
$	string命令的通知
l	list命令的通知
s	set命令的通知
h	hash命令的通知
z	zset命令的通知
x	过期事件：每当有过期键被删除时发送
e	驱逐(evict)事件：每当有键因为 maxmemory 政策而被删除时发送
A	参数 g$lshzxe 的别名

注意：参数至少要有一个K或者E，否则不会有任何通知，将参数设为KEA表示所有类型的通知

2.3 keyspace各种事件通知的产生

• DEL key [key …] 命令为每个被删除的键产生一个 del 通知。

• RENAME key newkey 产生两个通知：为来源键（source key）产生一个 rename_from 通知，并为目标键（destination key）产生一个 rename_to 通知。

• EXPIRE key seconds 和 EXPIREAT key timestamp 在键被正确设置过期时间时产生一个 expire 通知。当 EXPIREAT key timestamp 设置的时间已经过期，或者 EXPIRE key seconds 传入的时间为负数值时，键被删除，并产生一个 del 通知。

• SORT key [BY pattern] [LIMIT offset count] [GET pattern [GET pattern …]] [ASC | DESC] [ALPHA] [STORE destination] 在命令带有 STORE 参数时产生一个 sortstore 事件。如果 STORE 指示的用于保存排序结果的键已经存在，那么程序还会发送一个 del 事件。

• SET key value [EX seconds] [PX milliseconds] [NX|XX] 以及它的所有变种（SETEX key seconds value 、 SETNX key value 和 GETSET key value）都产生 set 通知。其中 SETEX key seconds value 还会产生 expire 通知。

• MSET key value [key value …] 为每个键产生一个 set 通知。

• SETRANGE key offset value 产生一个 setrange 通知。

• INCR key 、 DECR key 、 INCRBY key increment 和 DECRBY key decrement 都产生 incrby 通知。

• INCRBYFLOAT key increment 产生 incrbyfloat 通知。

• APPEND key value 产生 append 通知。

• LPUSH key value [value …] 和 LPUSHX key value 都产生单个 lpush 通知，即使有多个输入元素时，也是如此。

• RPUSH key value [value …] 和 RPUSHX key value 都产生单个 rpush 通知，即使有多个输入元素时，也是如此。

• RPOP key 产生 rpop 通知。如果被弹出的元素是列表的最后一个元素，那么还会产生一个 del 通知。

• LPOP key 产生 lpop 通知。如果被弹出的元素是列表的最后一个元素，那么还会产生一个 del 通知。

• LINSERT key BEFORE|AFTER pivot value 产生一个 linsert 通知。

• LSET key index value 产生一个 lset 通知。

• LTRIM key start stop 产生一个 ltrim 通知。如果 LTRIM key start stop 执行之后，列表键被清空，那么还会产生一个 del 通知。

• RPOPLPUSH source destination 和 BRPOPLPUSH source destination timeout 产生一个 rpop 通知，以及一个 lpush 通知。两个命令都会保证 rpop 的通知在 lpush 的通知之前分发。如果从键弹出元素之后，被弹出的列表键被清空，那么还会产生一个 del 通知。

• HSET hash field value 、 HSETNX hash field value 和 HMSET 都只产生一个 hset 通知。

• HINCRBY 产生一个 hincrby 通知。

• HINCRBYFLOAT 产生一个 hincrbyfloat 通知。

• HDEL 产生一个 hdel 通知。如果执行 HDEL 之后，哈希键被清空，那么还会产生一个 del 通知。

• SADD key member [member …] 产生一个 sadd 通知，即使有多个输入元素时，也是如此。

• SREM key member [member …] 产生一个 srem 通知，如果执行 SREM key member [member …] 之后，集合键被清空，那么还会产生一个 del 通知。

• SMOVE source destination member 为来源键（source key）产生一个 srem 通知，并为目标键（destination key）产生一个 sadd 事件。

• SPOP key 产生一个 spop 事件。如果执行 SPOP key 之后，集合键被清空，那么还会产生一个 del 通知。

• SINTERSTORE destination key [key …] 、 SUNIONSTORE destination key [key …] 和
  SDIFFSTORE destination key [key …] 分别产生 sinterstore 、 sunionostore 和 sdiffstore 三种通知。如果用于保存结果的键已经存在，那么还会产生一个 del 通知。

• ZINCRBY key increment member 产生一个 zincr 通知。（译注：非对称，请注意。）

• ZADD key score member [[score member] [score member] …] 产生一个 zadd 通知，即使有多个输入元素时，也是如此。

• ZREM key member [member …] 产生一个 zrem 通知，即使有多个输入元素时，也是如此。如果执行 ZREM key member [member …] 之后，有序集合键被清空，那么还会产生一个 del 通知。

• ZREMRANGEBYSCORE key min max 产生一个 zrembyscore 通知。（译注：非对称，请注意。）如果用于保存结果的键已经存在，那么还会产生一个 del 通知。

• ZREMRANGEBYRANK key start stop 产生一个 zrembyrank 通知。（译注：非对称，请注意。）如果用于保存结果的键已经存在，那么还会产生一个 del 通知。

• ZINTERSTORE destination numkeys key [key …] [WEIGHTS weight [weight …]] [AGGREGATE SUM|MIN|MAX] 和 ZUNIONSTORE destination numkeys key [key …] [WEIGHTS weight [weight …]] [AGGREGATE SUM|MIN|MAX] 分别产生 zinterstore 和 zunionstore 两种通知。如果用于保存结果的键已经存在，那么还会产生一个 del 通知。

三、客户端验证

此客户端验证只针对keyspace（键空间通知）

3.1 运行redis客户端

redis-cli

3.2 开启监听

PSUBSCRIBE __keyspace@*__:*

注意：因模糊匹配使用psubscribe命令

3.3 执行SETEX操作

新开启一个 redis客户端

SETEX mykey 10 redis

SETEX 为指定的 key 设置值及其过期时间。如果 key 已经存在， SETEX 命令将会替换旧的值。

3.4 查看监听

切换至开启监听的redis客户端

• 发生set事件
• 发生expire事件
• 发生expired时间

四、集成Redis实现keyspace监听

4.1 实现原理

请见 Spring Boot集成Redis实现keyevent监听 | Spring Cloud 33 文章 4.1 部分。

4.2 实现MessageListener监听器方式

4.2.1 自定义监听器RedisListener

com/gm/key/listen/listener/RedisListener.java：

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;@Slf4j
@Component
public class RedisListener implements MessageListener {@Autowiredprivate RedisTemplate<String, Object> redisTemplate;@Overridepublic void onMessage(Message message, byte[] pattern) {String key = message.toString();RedisSerializer<?> serializer = redisTemplate.getValueSerializer();String channel = String.valueOf(serializer.deserialize(message.getChannel()));log.info("redis key: {} , channel: {}", key, channel);}
}

4.2.2 开启监听及监听范围及事件类型

com/gm/key/listen/config/RedisListenerConfig.java：

package com.gm.key.listen.config;import com.gm.key.listen.listener.RedisListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.Topic;@Configuration
public class RedisListenerConfig {@AutowiredRedisListener redisListener;private static final Topic TOPIC_ALL_KEYSPACE = new PatternTopic("__keyspace@*__:*");@Beanpublic ThreadPoolTaskExecutor redisListenerTaskExecutor() {ThreadPoolTaskExecutor springSessionRedisTaskExecutor = new ThreadPoolTaskExecutor();springSessionRedisTaskExecutor.setCorePoolSize(12);springSessionRedisTaskExecutor.setMaxPoolSize(36);springSessionRedisTaskExecutor.setKeepAliveSeconds(60);springSessionRedisTaskExecutor.setThreadNamePrefix("redis-listener-");return springSessionRedisTaskExecutor;}@Beanpublic RedisMessageListenerContainer container(RedisConnectionFactory factory) {RedisMessageListenerContainer container = new RedisMessageListenerContainer();container.setConnectionFactory(factory);container.addMessageListener(redisListener, TOPIC_ALL_KEYSPACE);container.setTaskExecutor(redisListenerTaskExecutor());return container;}
}

请提前确认Redis服务端已开启keyspace（键空间通知）

RedisMessageListenerContainer 的默认使用线程池是SimpleAsyncTaskExecutor，每次消费都会创建一个线程来处理，这样就会有大量的新线程被创建。生产环境下建议使用自定义线程池，减少性能损耗。