Redis Stream消费异常致订阅取消，求保留订阅的处理方案

方案1：在消费逻辑内捕获所有异常 #

这是最直接有效的方式——把你的消费业务逻辑完全包裹在try-catch块里，不让任何未处理的异常抛到上层的StreamMessageListenerContainer。只要异常不逃逸，就不会触发cancelSubscriptionOnError机制，订阅自然不会中断。

示例代码：

@Override
public void onMessage(StreamMessage<String, String> message) {
    try {
        // 这里写你的消费业务逻辑
        String payload = message.getValue().get("content");
        handleBusinessLogic(payload);
    } catch (Exception e) {
        // 异常处理：打日志、记录到监控、或者转至死信队列
        log.error("消费消息失败，消息ID: {}", message.getId(), e);
        sendToDeadLetterQueue(message); // 可选：把失败消息转存，后续重试
    }
}

不管是运行时异常还是检查型异常，都要捕获处理，确保没有异常能触发容器的取消订阅逻辑。

方案2：配置全局ErrorHandler兜底 #

如果担心业务层漏捕异常，可以给StreamMessageListenerContainer配置全局错误处理器，统一处理所有消费过程中抛出的异常，并且控制是否取消订阅。

示例代码：

// 构建容器时配置ErrorHandler
StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, String> options =
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
                .pollTimeout(Duration.ofMillis(500))
                .errorHandler(throwable -> {
                    // 全局异常处理逻辑
                    log.error("消费全局异常", throwable);
                    // 返回true表示异常已处理，不会触发取消订阅；返回false则会走默认逻辑
                    return true;
                })
                .build();

StreamMessageListenerContainer<String, String> container =
        StreamMessageListenerContainer.create(redisConnectionFactory, options);

这里的关键是ErrorHandler的返回值：返回true，容器就会忽略该异常，继续保持订阅；返回false则会触发默认的取消订阅行为。

注意事项 #

• 优先用方案1，因为在业务层捕获异常能更精准地处理不同场景（比如区分可重试异常和不可重试异常），避免全局处理的粗糙性。
• 如果用方案2，一定要确保ErrorHandler里的逻辑不会抛出新的异常，否则还是会导致订阅中断。

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