Redis Stream监听器断连后失效：如何避免监听器因网络问题停止工作？

Redis Stream监听器断连后自动恢复：保障持续消息监听

使用spring data redis监听redis stream消息时，存在一个常见问题：监听器在长时间运行后（例如数周）停止接收消息，这通常是由于网络或连接问题导致程序与redis服务器断开连接。本文分析此问题，并提供一种解决方案，确保监听器在连接中断后自动恢复。

问题描述：

应用使用StreamMessageListenerContainer监听Redis Stream消息，但会间歇性地失效，无法接收新消息。怀疑是网络连接或连接数问题导致连接中断。

解决方案：

问题在于StreamMessageListenerContainer在默认情况下，遇到错误会停止监听。我们需要修改配置，使其在连接中断后能够自动重连并继续监听。

核心改进：

通过StreamReadRequest.builder().cancelOnError()方法自定义错误处理逻辑。将cancelOnError设置为返回false，即可在发生异常时阻止监听器停止工作，并尝试重新连接。

幻导航网

发现优质实用网站,开启网络探索之旅！

下载

改进后的代码片段：

@Bean
public List subscription(RedisConnectionFactory factory) {
    List resultList = new ArrayList<>();
    for (String redisStreamName : redisStreamNames) {
        initStream(redisStreamName, groups[0]);
        StreamReadRequest options = StreamReadRequest.builder(StreamOffset.create(redisStreamName, ReadOffset.lastConsumed()))
                .cancelOnError(throwable -> {
                    System.err.println("Redis连接错误: " + throwable.getMessage()); // 记录错误信息到日志
                    return false; // 不取消监听，尝试重新连接
                })
                .consumer(Consumer.from(groups[0], this.getClass().getName()))
                .autoAck(true) // 自动确认消息
                .build();

        StreamMessageListenerContainer> listenerContainer = 
            StreamMessageListenerContainer.create(factory, options); // 使用泛型指定消息类型
        Subscription subscription = listenerContainer.receive(options, streamListener);
        resultList.add(subscription);
        listenerContainer.start();
    }
    return resultList;
}

改进说明：

• 使用StreamReadRequest.builder()构建StreamReadRequest对象，并设置cancelOnError函数，在发生异常时返回false，避免监听器停止。
• 添加了错误日志记录，方便排查问题。
• 使用了泛型StreamMessageListenerContainer>，假设消息键为String，消息体为Map，请根据实际情况修改。 这更清晰地定义了消息类型。
• receiveAutoAck 方法已被替换为 receive 方法，并显式设置 autoAck 为 true，确保消息自动确认。

额外建议：

虽然此方法可防止监听器停止，但建议结合其他监控手段，例如监控Redis连接状态，以便及时发现并处理更深层次的网络或Redis服务问题。 定期检查日志中的错误信息，有助于诊断潜在问题。