redis在spring boot中异常退出的问题解决方案

2025-05-22 14:50

本文主要是介绍redis在spring boot中异常退出的问题解决方案,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

《redis在springboot中异常退出的问题解决方案》:本文主要介绍redis在springboot中异常退出的问题解决方案,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴...

问题:

Exception in thread "rtsp-consumer-3" org.springframework.data.redis.RedisConnectionFailureException: Unable to connect to Redis; nested exception is io.lettuce.core.RedisConnectionException: Unable to connect to localhost:6379
    at org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory$ExceptionTranslatingConnectionProvider.translateException(LettuceConnectionFactory.Java:1689)
    at org.springframework.pythondata.redis.connection.lettuce.LettuceConnectionFactory$ExceptionTranslatingConnectionProvider.getConnection(LettuceConnectionFactory.java:1597)
    at org.springframework.data.redis.connection.lettuce.LettuceConnection.doGetAsyncDedicatedConnection(LettuceConnection.java:1006)
    at org.springframework.data.redis.connection.lettuce.LettuceConnection.getOrCreateDedicatedConnection(LettuceConnection.java:1069)
    at org.springframework.data.redis.connection.lettuce.LettuceConnection.getAsyncDedicatedConnection(LettuceConnection.java:990)
    at org.springframework.data.redis.connection.lettuce.LettuceStreamCommands.getAsyncDedicatedConnection(LettuceStreamCommands.java:395)
    at org.springframework.data.redis.connection.lettuce.LettuceStreamCommands.xReadGroup(LettuceStreamCommands.java:346)
    at org.springframework.data.redis.connection.DefaultedRedisConnection.xReadGroup(DefaultedRedisConnection.java:592)
    at org.springframework.data.redis.core.DefaultStreamOperations$4.inRedis(DefaultStreamOperations.java:310)
    at org.springframework.data.redis.core.DefaultStreamOperations$RecordDeserializingRedisCallback.doInRedis(DefaultStreamOperations.java:387)
    at org.springframework.data.redis.core.DefaultStreamOperations$RecordDeserializingRedisCallback.doInRedis(DefaultStreamOperations.java:382)
    at org.springframework.data.redis.core.RedisTemplate.execute(RedisTemplate.java:222)
    at org.springframework.data.redis.core.RedisTemplate.execute(RedisTemplate.java:189)
    at org.springframework.data.redis.core.AbstractOperations.execute(AbstractOperations.java:96)
    at org.springframework.data.redis.core.DefaultStreamOperations.read(DefaultStreamOperations.java:305)
    at com.ruoyi.vedioFrame.utils.RedisStreamOperations.readGroup(RedisStreamOperations.java:70)
    at com.ruoyi.vedioFrame.service.impl.StreamConsumerService.consumeFrames(StreamConsumerService.java:200)
    at com.ruoyi.vedioFrame.service.impl.StreamConsumerService.lambda$null$0(StreamConsumerService.java:108)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: io.lettuce.core.RedisConnectionException: Unable to connect to localhost:6379
    at io.lettuce.core.RedisConnectionException.create(RedisConnectionException.java:78)
    at io.lettuce.core.RedisConnecwww.chinasem.cntionException.create(RedisConnectionException.java:56)
    at io.lettuce.core.AbstractRedisClient.getConnection(AbstractRedisClient.java:330)
    at io.lettuce.core.RedisClient.connect(RedisClient.java:216)
    at org.springframework.data.redis.connection.lettuce.StandaloneConnectionProvider.lambda$getConnection$1(StandaloneConnectionProvider.java:115)
    at java.util.Optional.orElseGet(Optional.java:267)
    at org.springframework.data.redis.connection.lettuce.StandaloneConnectionProvider.getConnection(StandaloneConnectionProvider.java:115)
    at org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory$ExceptionTranslatingConnectionProvider.getConnection(LettuceConnectionFactory.java:1595)
    ... 19 more
Caused by: io.netty.channel.AbstractChannel$AnnotatedSocketException: Address already in use: no further information: localhost/127.0.0.1:6379
Caused by: java.net.BindException: Address already in use: no further information
    at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
    at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717)
    at io.netty.channel.socket.nio.NIOSocketChannel.doFinishConnect(NioSocketChannel.java:337)
    at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:334)
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:776)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
    at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
    at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    at java.lang.Thread.run(Thread.java:745) 

解决

根据代码和现象分析,根本原因可能是处理时间过长导致Redis消息积压触发超时异常。以下是详细分析和解决方案:

问题根源

同步阻塞模型

  • frameProcessor.process()调用外部接口时存在网络延迟,导致消息处理时间超过Redis的block超时(2000ms)。
  • 未及时ACK会导致消息重新进入Pending状态,积累后触发READONLY错误或消费者组失效。

Thread.sleep(1000)加剧阻塞

  • 即使没有消息,每次循环强制休眠1秒,降低消费吞吐量,无法及时处理积压消息。

异常未妥善处理

  • 注释掉的try-catch导致处理过程中的异常(如网络超时)未被捕获,中断循环后Redis连接未正确恢复。

️ 解决方案

1. 异步处理 + 提前ACK(关键步骤)

将消息处理与ACK分离,处理前立即ACK避免消息积压,异步处理业务逻辑:

ExecutorService executor = Executors.newCachedThreadPool(); // 全局线程池
while (!Thread.currentThread().isInterrupted()) {
    List<MapRecord<String, Object, Object>> records = redisStreamOperations.readGroup(...);
    if (records != null) {
        for (MapRecord<String, Object, Object> record : records) {
            // 提前ACK释放消息
            redisStreamOperations.ack(streamKey, groupName, record.getId());
            // 提交异python步任务处理业务
            executor.submit(() -> {
                try {
                    String strdata = (String) record.getValue().get("frameData");
                    byte[] frameData = Base64.getDecoder().decode(strdata);
                    BufferedImage image = ImageIO.read(new ByteArrayInputStream(frameData));
                    frameProcessor.process(image, streamId, commonQueues, regions);
                } catch (Exception e) {
                    log.error("处理帧失败: {}", e.getMessage());
                }
            });
        }
    }
    // 移除Thread.sleep(1000) 提升吞吐量
}

2. 调整Redis消费者组参数

增加XGROUP CREATE时的BLOCKIDLE超时:

XGROUP CREATE stream:1234 mygroup $ MKSTREAM BLOCK 5000 IDLE 30000

3. 优化Redis连接配置

确保RedisTemplate的读写超时大于业务处理最长时间:

@Bean
public RedisTemplate<String, Object> redisTemplate() {
    RedisTemplate<String, Object> template = new RedisTemplate<>();
    template.setConnectionFactory(redisConnectionFactory());
    // 设置超时时间(示例:10秒)
    RedisConnectionFactory factory = template.getConnectionFactory();
    if (factory instanceof LettuceConnectionFactory) {
        ((LettuceConnectionFactory) factory).setTimeout(Duration.ofSeconds(10));
    }
    return template;
}

4. 监控Pending队列

定期检查Pending消息数量,异常时重置消费者:

// 在循环中添加检查
Long pending = redisTemplate.opsForStream().pending(streamKey, groupName);
if (pending > 1000) { // 阈值根据业务设定
    redisTemplate.opsForStream().destroyGroup(streamKey, groupName);
    redisTemplate.opsForStream().createGroup(streamKey, groupName);
}

5. 添加熔断机制

当外部接口连续失败时,暂停消费避免雪崩:

// 全局熔断器
CircuitBreaker breaker = CircuitBreaker.ofDefaults("frameProcessor");
// 处理逻辑改为
breaker.executeSupplier(() -> {
    frameProcessor.process(...);
    return null;
});

验证方案

日志监控

// 在异步任务中添加耗时日志
long start = System.currentTimeMillis();
frameProcessor.process(...);
log.info("处理耗时: {}ms", System.currentTimeMillis() - start);

确保单次处理时间低于Redis超时阈值。

Redis CLI检查

XINFO GROUPS stream:1234  # 查看消费者组状态
XPENDING stream:1234 mygroup # 查看Pending消息

关键优化点总结

措施作用
异步处理 + 提前ACK彻底解耦处理耗时与Redis消费
增大Redis超时参数避免服务端主动断开连接
线程池隔离防止外部接口故障拖垮整个消费服务
熔断机制快速失败,保php系统稳定性

通过以上改造,即使frameProcessor.process()调用外部平台接口,也能通过异步化和资源隔离保证Redis消费链路的稳定性。最终只需确保线程池和熔断参数适配业务负载即可。

代码具体添加位置:

以下http://www.chinasem.cn是针对 优化Redis连接配置监控Pending队列添加熔断机制 的代码具体添加位置的详细说明:

一、优化Redis连接配置

位置:Redis配置类
在Spring Boot的Redis配置类中调整超时参数(通常为RedisConfig.java):

@Configuration
public class RedisConfig {
    @Bean
    public RedisTemplate<String, Object> redisTemplate(LettuceConnectionFactory lettuceConnectionFactory) {
        // 设置连接超时和读写超时(关键参数)
        lettuceConnectionFactory.setTimeout(Duration.ofSeconds(10));  // 命令超时时间
        lettuceConnectionFactory.setShareNativeConnection(false);    // 禁用共享连接,避免阻塞
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(lettuceConnectionFactory);
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new GenericJackson2jsonRedisSerializer());
        return template;
    }
}

关键参数说明

  • setTimeout(10秒):确保超时时间大于frameProcessor.process()的最长处理时间
  • setShareNativeConnection(false):避免多个线程共享同一个连接导致阻塞。

二、监控Pending队列

位置:consumeFrames方法内的循环中
在消费消息的主循环中定期检查Pending队列:

private void consumeFrames(String streamId, String groupName, String consumerName,
                           CommonQueues commonQueues, String regions) throws InterruptedException, IOException {
    // ... 其他初始化代码 ...
    int checkPendingInterval = 10; // 每处理10次循环检查一次Pending队列
    int loopCount = 0;
    while (!Thread.currentThread().isInterrupted()) {
        // ... 原有代码读取消息 ...
        // 监控Pending队列的逻辑(添加位置)
        loopCount++;
        if (loopCount % checkPendingInterval == 0) {
            String streamKey = "stream:" + streamId;
            PendingMessages pending = redisStreamOperations.pending(streamKey, groupName);
            if (pending != null && pending.getTotalPendingMessages() > 1000) { // 阈值根据业务调整
                log.warn("检测到Pending消息积压 {} 条,重置消费者组", pending.getTotalPendingMessages());
                redisStreamOperations.destroyGroup(streamKey, groupName);
                redisStreamOperations.createGroup(StreamKey.of(streamKey), groupName);
            }
        }
        // ... 后续处理代码 ...
    }
}

说明

  • 通过redisStreamOperations.pending()获取当前Pending消息数。
  • 当Pending消息超过阈值时,强制销毁并重建消费者组,避免消息卡死。

三、添加熔断机制

位置:处理消息的业务逻辑外层
使用Resilience4j熔断器包裹frameProcessor.process()调用:

1. 熔断器配置类

@Configuration
public class CircuitBreakerConfig {
    @Bean
    public CircuitBreaker frameProcessorCircuitBreaker() {
        CircuitBreakerConfig config = CircuitBreakerConfig.custom()
            .failureRateThreshold(50)          // 失败率阈值50%
            .slidingWindowType(SlidingWindowType.COUNT_BASED)
            .slidingWindowsize(10)             // 基于最近10次调用统计
            .minimumNumberOfCalls(5)           // 最少5次调用后开始计算
            .waitDurationInOpenState(Duration.ofSeconds(30)) // 熔断后30秒进入半开状态
            .build();
        return CircuitBreakerRegistry.of(config).circuitBreaker("frameProcessor");
    }
}

2. 在消费代码中使用熔断器

public class YourConsumerClass {
    @Autowired
    private CircuitBreaker frameProcessorCircuitBreaker; // 注入熔断器
    private void consumeFrames(...) {
        // ... 原有代码 ...
        for (MapRecord<String, Object, Object> record : records) {
            redisStreamOperations.ack(...); // 提前ACK
            // 使用熔断器保护处理逻辑(添加位置)
            Try.runRunnable(() -> frameProcessorCircuitBreaker.executeRunnable(() -> {
                String strdata = (String) record.getValue().get("frameData");
                byte[] frameData = Base64.getDecoder().decode(strdata);
                BufferedImage image = ImageIO.read(new ByteArrayInputStream(frameData));
                frameProcessor.process(image, streamId, commonQueues, regions);
            })).onFailure(e -> log.error("处理失败且熔断: {}", e.getMessage()));
        }
        // ... 后续代码 ...
    }
}

熔断逻辑说明

  • frameProcessor.process()连续失败触发阈值时,熔断器会暂时阻止后续调用,避免雪崩效应。
  • 熔断期间直接跳过处理,但仍会ACK消息(根据业务需求选择是否重试)。

四、代码集成位置总结

优化措施代码位置关键注解
Redis连接配置Redis配置类(如RedisConfig.java调整超时时间和连接池参数
Pending队列监控consumeFrames方法的主循环内定期检查+自动重置消费者组
熔断机制业务处理代码外层(包裹frameProcessor.process依赖熔断器库(如Resilience4j)

五、参数调整建议

Redis超时

  • lettuceConnectionFactory.setTimeout应大于frameProcessor.process()的最大处理时间 + 网络抖动余量(如设置为实际最大处理时间的2倍)。

Pending队列阈值

  • 如果每秒处理100条消息,阈值可设置为1000(相当于10秒积压量)。

熔断器参数

  • failureRateThreshold:根据外部接口的稳定性调整(如频繁超时可设为70%)。
  • waitDurationInOpenState:根据外部服务恢复时间调整(如30秒到5分钟)。

通过以上改造,即使frameProcessor.process()调用外部平台接口,也能通过资源隔离、快速失败和自动恢复机制保障Redis消费链路的稳定性。

到此这篇关于redis在spring boot中异常退出的文章就介绍到这了,更多相关redis spring boot异常退出内容请搜索China编程(www.chinasem.cn)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程China编程(www.chinasem.cn)!

这篇关于redis在spring boot中异常退出的问题解决方案的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1154734

相关文章

Spring Boot中WebSocket常用使用方法详解

《SpringBoot中WebSocket常用使用方法详解》本文从WebSocket的基础概念出发,详细介绍了SpringBoot集成WebSocket的步骤,并重点讲解了常用的使用方法,包括简单消... 目录一、WebSocket基础概念1.1 什么是WebSocket1.2 WebSocket与HTTP

Knife4j+Axios+Redis前后端分离架构下的 API 管理与会话方案(最新推荐)

《Knife4j+Axios+Redis前后端分离架构下的API管理与会话方案(最新推荐)》本文主要介绍了Swagger与Knife4j的配置要点、前后端对接方法以及分布式Session实现原理,... 目录一、Swagger 与 Knife4j 的深度理解及配置要点Knife4j 配置关键要点1.Spri

SpringBoot+Docker+Graylog 如何让错误自动报警

《SpringBoot+Docker+Graylog如何让错误自动报警》SpringBoot默认使用SLF4J与Logback,支持多日志级别和配置方式,可输出到控制台、文件及远程服务器,集成ELK... 目录01 Spring Boot 默认日志框架解析02 Spring Boot 日志级别详解03 Sp

java中反射Reflection的4个作用详解

《java中反射Reflection的4个作用详解》反射Reflection是Java等编程语言中的一个重要特性,它允许程序在运行时进行自我检查和对内部成员(如字段、方法、类等)的操作,本文将详细介绍... 目录作用1、在运行时判断任意一个对象所属的类作用2、在运行时构造任意一个类的对象作用3、在运行时判断

java如何解压zip压缩包

《java如何解压zip压缩包》:本文主要介绍java如何解压zip压缩包问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录Java解压zip压缩包实例代码结果如下总结java解压zip压缩包坐在旁边的小伙伴问我怎么用 java 将服务器上的压缩文件解压出来,

SpringBoot中SM2公钥加密、私钥解密的实现示例详解

《SpringBoot中SM2公钥加密、私钥解密的实现示例详解》本文介绍了如何在SpringBoot项目中实现SM2公钥加密和私钥解密的功能,通过使用Hutool库和BouncyCastle依赖,简化... 目录一、前言1、加密信息(示例)2、加密结果(示例)二、实现代码1、yml文件配置2、创建SM2工具

Spring WebFlux 与 WebClient 使用指南及最佳实践

《SpringWebFlux与WebClient使用指南及最佳实践》WebClient是SpringWebFlux模块提供的非阻塞、响应式HTTP客户端,基于ProjectReactor实现,... 目录Spring WebFlux 与 WebClient 使用指南1. WebClient 概述2. 核心依

Spring Boot @RestControllerAdvice全局异常处理最佳实践

《SpringBoot@RestControllerAdvice全局异常处理最佳实践》本文详解SpringBoot中通过@RestControllerAdvice实现全局异常处理,强调代码复用、统... 目录前言一、为什么要使用全局异常处理?二、核心注解解析1. @RestControllerAdvice2

Spring IoC 容器的使用详解(最新整理)

《SpringIoC容器的使用详解(最新整理)》文章介绍了Spring框架中的应用分层思想与IoC容器原理,通过分层解耦业务逻辑、数据访问等模块,IoC容器利用@Component注解管理Bean... 目录1. 应用分层2. IoC 的介绍3. IoC 容器的使用3.1. bean 的存储3.2. 方法注

Spring事务传播机制最佳实践

《Spring事务传播机制最佳实践》Spring的事务传播机制为我们提供了优雅的解决方案,本文将带您深入理解这一机制,掌握不同场景下的最佳实践,感兴趣的朋友一起看看吧... 目录1. 什么是事务传播行为2. Spring支持的七种事务传播行为2.1 REQUIRED(默认)2.2 SUPPORTS2