Springboot Redis Session 消息监听遗漏Bug记录

发表于

场景

Springboot使用Redis作为分布式节点的共享Session, 使用Kafka传输日志到ELK，出现每小时百万条报警，报警日志均为WARN级别，如下：

1
2
RedisOperationsSessionRepository：Unable to publish SessionDestroyedEvent for session
RedisMessageListenerContainer：Execution of message listener failed, and no ErrorHandler has been set

Session失效监听失败, 报警源码如下：
RedisOperationsSessionRepository
RedisMessageListenerContainer

解决

紧急解决

日志报警撑爆了用于传输日志的Kafka，首先更新logback，新增如下，屏蔽日志，但问题仍在。

1
2
<logger name="org.springframework.session.data.redis.RedisOperationsSessionRepository" level="ERROR"/>
<logger name="org.springframework.data.redis.listener.RedisMessageListenerContainer" level="ERROR"/>

Bug修复

相关依赖为最新版本: spring-boot-starter-parent与spring-boot-starter-data-redis为1.5.18.RELEASE，spring-session-data-redis为1.3.4.RELEASE。
查源码(KeyExpirationEventMessageListener)可知, KeyExpirationEventMessageListener重写了父类KeyspaceEventMessageListener的doHandleMessage方法，只需重写KeyspaceEventMessageListener的onMessage方法即可解决问题，如下：
1.新增类KeyExpiredListener

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import java.nio.charset.StandardCharsets;
public class KeyExpiredListener extends KeyExpirationEventMessageListener {
    private static final Logger logger = LoggerFactory.getLogger(KeyExpiredListener.class);
    
    public KeyExpiredListener(RedisMessageListenerContainer listenerContainer) {
        super(listenerContainer);
    }
    @Override
    public void onMessage(Message message, byte[] pattern) {
        if (message == null || message.getChannel() == null || message.getBody() == null) {
            return;
        }
        String key = new String(message.getBody(), StandardCharsets.UTF_8);
        String channel = new String(message.getChannel(), StandardCharsets.UTF_8);
        logger.info("redis key expired, key: {}, channel: {}, time: {}", key, channel, System.currentTimeMillis());
    }
}

2.RedisConfig类新增RedisMessageListenerContainer、KeyExpiredListener的构建

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisSentinelConfiguration;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.session.data.redis.config.annotation.web.http.EnableRedisHttpSession;
import org.springframework.stereotype.Component;
import redis.clients.jedis.JedisPoolConfig;
@Component
@EnableRedisHttpSession(maxInactiveIntervalInSeconds = 86400)
public class RedisConfig {
    private final RedisCon redisCon;
    @Autowired
    public RedisConfig(RedisCon redisCon) {
        this.redisCon = redisCon;
    }
    @Bean
    public RedisSentinelConfiguration sentinelConfig() {
        RedisSentinelConfiguration sentinelConfig = new RedisSentinelConfiguration().master(redisCon.getMasterName());
        String[] sentinelNodes = redisCon.getSentinelAddress().split(",");
        for (String node : sentinelNodes) {
            String[] ipAndPort = node.split(":");
            String nodeIp = ipAndPort[0];
            Integer nodePort = Integer.valueOf(ipAndPort[1]);
            sentinelConfig.sentinel(nodeIp, nodePort);
        }
        return sentinelConfig;
    }
    @Bean
    public RedisConnectionFactory redisConnectionFactory() {
        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
        jedisPoolConfig.setMaxTotal(redisCon.getMaxActive());
        jedisPoolConfig.setMaxIdle(redisCon.getMaxIdle());
        jedisPoolConfig.setMinIdle(redisCon.getMinIdle());
        jedisPoolConfig.setMaxWaitMillis(redisCon.getMaxWaitTime());
        JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory(sentinelConfig(), jedisPoolConfig);
        jedisConnectionFactory.setDatabase(redisCon.getDbIndex());
        if (redisCon.getPassword() != null && !"".equals(redisCon.getPassword())) {
            jedisConnectionFactory.setPassword(redisCon.getPassword());
        }
        jedisConnectionFactory.setTimeout(redisCon.getTimeout());
        jedisConnectionFactory.afterPropertiesSet();
        return jedisConnectionFactory;
    }
    @Bean
    public RedisTemplate<String, Object> redisTemplate() {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(redisConnectionFactory());
        return template;
    }
    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer() {
        RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
        redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory());
        return redisMessageListenerContainer;
    }
    @Bean
    public KeyExpiredListener keyExpiredListener() {
        return new KeyExpiredListener(this.redisMessageListenerContainer());
    }
}