集群模式下Redis键空间通知及Spring Data Redis索引清理问题咨询

问题根源 #

在Amazon ElastiCache集群模式下，Redis的键会分布在不同主节点上。Spring Data Redis默认仅随机选择一个集群节点订阅键空间事件，导致其他主节点上的主条目过期/删除事件无法被捕获，进而关联的索引条目无法同步清除。

解决方案：订阅所有Redis主节点 #

要确保能接收所有节点的键空间事件，需为集群中的每个主节点单独创建订阅连接并注册事件监听器，具体实现步骤如下：

1. 获取Redis集群所有主节点信息 #

通过RedisClusterConnection获取集群节点元数据，筛选出主节点：

@Autowired
private RedisConnectionFactory connectionFactory;

private Set<RedisNode> getMasterNodes() {
    try (RedisClusterConnection connection = connectionFactory.getClusterConnection()) {
        ClusterTopology topology = connection.clusterGetClusterInfo();
        return topology.getNodes().stream()
                .filter(RedisNode::isMaster)
                .collect(Collectors.toSet());
    }
}

2. 为每个主节点创建独立的消息监听器容器 #

自定义配置，为每个主节点初始化RedisMessageListenerContainer并注册事件监听器：

@Bean
public List<RedisMessageListenerContainer> redisMessageListenerContainers(KeyExpirationEventMessageListener expirationListener) {
    Set<RedisNode> masterNodes = getMasterNodes();
    List<RedisMessageListenerContainer> containers = new ArrayList<>();

    for (RedisNode node : masterNodes) {
        LettuceConnectionFactory nodeConnectionFactory = new LettuceConnectionFactory(node.getHost(), node.getPort());
        nodeConnectionFactory.afterPropertiesSet();

        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(nodeConnectionFactory);
        // 订阅键过期事件，可根据需求调整事件类型（如删除事件）
        container.addMessageListener(expirationListener, new PatternTopic("__keyevent@*__:expired"));
        container.afterPropertiesSet();
        containers.add(container);
    }
    return containers;
}

3. 实现键过期事件监听器 #

自定义监听器处理主条目过期逻辑，同步删除对应索引：

@Component
public class CustomKeyExpirationListener extends KeyExpirationEventMessageListener {

    private final RedisTemplate<String, Object> redisTemplate;

    public CustomKeyExpirationListener(RedisMessageListenerContainer listenerContainer, RedisTemplate<String, Object> redisTemplate) {
        super(listenerContainer);
        this.redisTemplate = redisTemplate;
    }

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String expiredKey = new String(message.getBody());
        // 根据业务规则生成索引键并删除
        String indexKey = "index:" + expiredKey;
        redisTemplate.delete(indexKey);
    }
}

注意事项 #

• 节点动态变动处理：Redis集群可能发生主节点故障转移、扩容缩容，需定期刷新主节点列表并重建订阅容器，可通过定时任务或集群拓扑变化监听实现。
• 资源占用控制：每个主节点的订阅会占用一个连接，需根据集群规模评估资源消耗，确保ElastiCache连接数配额充足。
• 事件配置验证：确认Redis集群已开启对应键空间事件（通过notify-keyspace-events配置，如设置为Ex开启过期事件）。

内容的提问来源于stack exchange，提问作者saurav