踩坑周折Redis过期监听之路（redis过期监听坑）

在开发正在正火热的分布式服务中，Redis作为一个关键的存储组件经常被使用。

比如在项目中经常会用Redis来储存账号的Session以及超时的定时任务等等，而常用的会话或任务都会有一个时间的概念：**过期时间**。那么如何监听准确的Redis过期时间一直是持续困扰着大家的问题。

对于Redis有两种比较常见的过期时间侦听方案：

1. 将Redis中存储的内容设置有过期时间，然后再由程序定时向Redis中轮询查看有没有过期键值；

2. 采用Redis自身提供的 Keyspace Notification 的事件机制，即 Redis Log 的 Keyspace[expired] 事件，当发生键过期时，会自动发出一个 ‘__keyspace@xx__:keyname’ 的 Channel，接收这样的 Channel 信息则知道key的过期了。

而我拿到这个任务的时候恰好是COVID后的大年初一，本以为会42度的热情迎接新年，结果理想远异于现实：加起来的坑也有数十个，实现程序的Redis过期监听可谓困难重重。

首先尝试的方案便是大家常用的轮询检查Redis中过期键值，但显而易见的是，当Redis服务器中过期键值数量较大时，轮询检查压力肯定会很大，并且比较耗时。再如检查间隔只有一分钟，可谓是一种浪费时间，尤其是有可能某些键值在一分钟内被重新设置，但未来也无法得知。

因此，我又转向采用Redis Log 的 Keyspace[expired] 事件来进行过期监听的尝试，于是安装了Redis的源码，通过Flume将Redis的 Log 事件引入Flume实时传输给Kafka，然后再利用Kafka自带的Offset提取想要的信息。期间也折腾了不少，比如Flume无法将接收原始信息传输给Kafka，排查到久参数没有正确设置；比如Flume把Redis Log事件传给Kafka之后，超过32G大小的log被跳过分发而导致实际的过期监听无法展开，排查到这儿原因是没有正确限定长度模板并且类型是StringIndex。

最终，经过大量的坑和折腾，终于在一个清凉的night中完成了Redis的过期监听任务，哈哈，网上查了各种资料，总算是能把Redis的过期监听实现了！如今，有了这招，可以为我们实现中byte级别准确及时的检测。

“` java

// Flume 配置文件。

# redis source 定义

[agent-name]

maxBatchSize = 1000

maxFileSize = 10485760

# tl source 定义

[agent-name]

type = exec

command = tl -F /home/monit_rpdp/usr/local/redis/logs/redis.log

# sink　定义

[agent-name]

type = org.apache.flume.sink.kafka.KafkaSink

topic = redis_expired

kafka.bootstrap.servers = localhost:9092

kafka.metadata.broker.list = localhost:9092

kafka.request.required.acks = 1

[agent-name]

key.serializer = org.apache.kafka.common.serialization.StringSerializer

# key.serializer = org.apache.kafka.common.serialization.StringIndexSerializer

value.serializer = org.apache.kafka.common.serialization.StringSerializer

# value.serializer = org.apache.kafka.common.serialization.StringIndexSerializer

# value.serializer = org.apache.kafka.common.serialization.ByteArraySerializer

# value.serializer = org.apache.kafka.common.serialization.ByteArrayIndexSerializer

最终，Redis过期时间监听的实现可谓是一次漫长却又有成就感的探索之旅，相信有了这样的经历，以后接下来的类似问题难度也不大，温故而知新，只有不断地折腾才能得出更