SpringBoot 整合 Redis 监听 Key 过期策略

最近遇到的一个需求，时间到期或者超时了可以自动处理一些逻辑，第一个想到的就是 MQ , 想 MQ 不就是专门干这事的吗？但是为了考虑时间学习成本，并且考虑访问量、并发量都不大，就放弃了这种方式，采用的则是 redis 过期 key 监听事件，是因为现在的项目都会使用 redis 做一些缓存，存储一些常用的数据。根据失效的 key 去处理一些逻辑。

使用 redis 的 Keyspace Notifications，中文翻译就是键空间机制，就是利用该机制可以在 key 失效之后，提供一个回调，实际上是 redis 会给客户端发送一个消息。是需要 redis 版本 2.8 以上。

优点：

由于使用 Redis 作为消息通道，消息都存储在 Redis 中。如果发送程序或者任务处理程序挂了，重启之后，还有重新处理数据的可能性。
做集群扩展相当方便
时间准确度高

缺点:

需要额外进行 redis 维护

实现步骤

Redis 安装操作步骤可看这篇： 链接

默认情况下，Redis 并未开启键空间消息提醒功能。修改 redis.conf 配置，在 redis.conf 中，加入一条配置：
```
notify-keyspace-events Ex
```
启动 redis 服务，测试链接是否成功
SpringBoot 整合 redis ，可参考这篇文章：链接

在 RedisConfig.java 配置类中配置 redis 监听的 bean 对象，增加以下配置

// redis key 过期监听事件
@Bean
RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
    RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    container.setConnectionFactory(connectionFactory);
    return container;
}

添加 redis key 过期监听类，监听过期 key

import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Component;

/**
 * redis 过期事件监听器
 */
@Component
public class RedisKeyExpireListener extends KeyExpirationEventMessageListener {

    public RedisKeyExpireListener(RedisMessageListenerContainer listenerContainer) {
        super(listenerContainer);
    }

    /**
     * 重写 onMessage方法
     */
    @Override
    public void onMessage(Message message, byte[] pattern) {
        // 过期的key
        String expireKey = message.toString();
        System.out.println("过期的key：" + expireKey);
        // 根据key,执行自己需要实现的功能。
    }
}

最后测试

注意：redis 集群监听配置和单机的有很大区别，以下是集群配置方式

修改 yml 配置文件，redis 改成集群模式

  # redis 配置
  redis:
    # 集群连接地址
    cluster:
      nodes:
        - 127.0.0.1:7000
        - 127.0.0.1:7001
        - 127.0.0.1:7002
        - 127.0.0.1:7003
        - 127.0.0.1:7004
        - 127.0.0.1:7005
    # 端口，默认为6379
    port: 6379
    # 数据库索引
    database: 0
    # 密码
    password: password
    # 连接超时时间
    timeout: 10s
    lettuce:
      pool:
        # 连接池中的最小空闲连接
        min-idle: 0
        # 连接池中的最大空闲连接
        max-idle: 8
        # 连接池的最大数据库连接数
        max-active: 8
        # #连接池最大阻塞等待时间（使用负值表示没有限制）
        max-wait: -1ms

添加 RedisProperties 配置类

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
import java.util.List;

@Data
@Component
@ConfigurationProperties(prefix = "spring.redis.cluster") // 前缀
public class RedisProperties {
    /**
     * 对应属性名,与配置中相同
     */
    private List<String> nodes;

}

redis 过期 key 订阅实现

import io.lettuce.core.RedisURI;
import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.models.partitions.RedisClusterNode;
import io.lettuce.core.cluster.pubsub.RedisClusterPubSubAdapter;
import io.lettuce.core.cluster.pubsub.StatefulRedisClusterPubSubConnection;
import io.lettuce.core.cluster.pubsub.api.async.NodeSelectionPubSubAsyncCommands;
import io.lettuce.core.cluster.pubsub.api.async.PubSubAsyncNodeSelection;
import io.lettuce.core.pubsub.RedisPubSubAdapter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.List;

/**
 * redis 过期key监听
 */
@SuppressWarnings("rawtypes")
@Component
@Slf4j
public class ClusterSubscriber extends RedisPubSubAdapter implements ApplicationRunner {

    // 过期事件监听
    private static final String EXPIRED_CHANNEL = "__keyevent@0__:expired";

    @Autowired
    private RedisProperties redisProperties;

    @Value("${spring.redis.password}")
    private String password;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        log.info("过期事件，启动监听......");
        //项目启动后就运行该方法
        startListener();
    }

    /**
     * 启动监听
     */
    @SuppressWarnings("unchecked")
    public void startListener() {
        //redis集群监听
        final String[] redisNodes = redisProperties.getNodes().toArray(new String[0]);
        //监听其中一个端口号即可
        List<RedisURI> redisURIList = new ArrayList<>();
        for (int i = 0; i < redisNodes.length; i++) {
            RedisURI redisURI = RedisURI.create("redis://" + redisNodes[i]);
            redisURI.setPassword(password);
            redisURIList.add(redisURI);
        }
        RedisClusterClient clusterClient = RedisClusterClient.create(redisURIList);

        StatefulRedisClusterPubSubConnection<String, String> pubSubConnection = clusterClient.connectPubSub();
        //redis 节点间消息的传播为true
        pubSubConnection.setNodeMessagePropagation(true);
        //过期消息的接受和处理
        pubSubConnection.addListener(new RedisClusterPubSubAdapter() {
            @Override
            public void message(RedisClusterNode node, Object channel, Object message) {
                // 注意这里只能获取到key，不能获取到key对应的值
                String msgKey = message.toString();
                System.out.println("过期key：" + msgKey);
                // 根据 key 处理业务逻辑
            }
        });

        //异步操作
        PubSubAsyncNodeSelection<String, String> masters = pubSubConnection.async().masters();
        NodeSelectionPubSubAsyncCommands<String, String> commands = masters.commands();
        //设置订阅消息类型，一个或多个
        commands.subscribe(EXPIRED_CHANNEL);
    }
}