java - Lettuce RedisCache 抛出 java.util.concurrent.RejectedExecutionException 线程限制超出替换阻塞的工作线程

Question

我使用 Spring Boot 和 Redis-Cache，使用 Lettuce 默认配置，并在服务器启动几分钟后收到以下 RejectedExecutionException:

org.springframework.data.redis.RedisSystemException: Unknown redis exception; nested exception is java.util.concurrent.RejectedExecutionException: Thread limit exceeded replacing blocked worker
        at org.springframework.data.redis.FallbackExceptionTranslationStrategy.getFallback(FallbackExceptionTranslationStrategy.java:53)
        at org.springframework.data.redis.FallbackExceptionTranslationStrategy.translate(FallbackExceptionTranslationStrategy.java:43)
        at org.springframework.data.redis.connection.lettuce.LettuceConnection.convertLettuceAccessException(LettuceConnection.java:257)
        at org.springframework.data.redis.connection.lettuce.LettuceStringCommands.convertLettuceAccessException(LettuceStringCommands.java:718)
        at org.springframework.data.redis.connection.lettuce.LettuceStringCommands.get(LettuceStringCommands.java:63)
        at org.springframework.data.redis.connection.DefaultedRedisConnection.get(DefaultedRedisConnection.java:210)
        at org.springframework.data.redis.cache.DefaultRedisCacheWriter.lambda$get$1(DefaultRedisCacheWriter.java:109)
        at org.springframework.data.redis.cache.DefaultRedisCacheWriter.execute(DefaultRedisCacheWriter.java:242)
        at org.springframework.data.redis.cache.DefaultRedisCacheWriter.get(DefaultRedisCacheWriter.java:109)
        at org.springframework.data.redis.cache.RedisCache.lookup(RedisCache.java:82)
        at org.springframework.cache.support.AbstractValueAdaptingCache.get(AbstractValueAdaptingCache.java:58)
        at org.springframework.cache.interceptor.AbstractCacheInvoker.doGet(AbstractCacheInvoker.java:73)
        at org.springframework.cache.interceptor.CacheAspectSupport.findInCaches(CacheAspectSupport.java:525)
        at org.springframework.cache.interceptor.CacheAspectSupport.findCachedItem(CacheAspectSupport.java:490)
        at org.springframework.cache.interceptor.CacheAspectSupport.execute(CacheAspectSupport.java:372)
        at org.springframework.cache.interceptor.CacheAspectSupport.execute(CacheAspectSupport.java:316)
        at org.springframework.cache.interceptor.CacheInterceptor.invoke(CacheInterceptor.java:61)
        at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185)
        at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688)
        at my.controller.MyController.lambda$myFunc$0(MyController.java:60)
        at java.util.concurrent.ForkJoinTask$AdaptedRunnableAction.exec(ForkJoinTask.java:1386)
        at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
        at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
        at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
        at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
Caused by: java.util.concurrent.RejectedExecutionException: Thread limit exceeded replacing blocked worker
        at java.util.concurrent.ForkJoinPool.tryCompensate(ForkJoinPool.java:2011)
        at java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3310)
        at java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1775)
        at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1915)
        at io.lettuce.core.protocol.AsyncCommand.await(AsyncCommand.java:81)
        at io.lettuce.core.LettuceFutures.awaitOrCancel(LettuceFutures.java:112)
        at io.lettuce.core.FutureSyncInvocationHandler.handleInvocation(FutureSyncInvocationHandler.java:62)
        at io.lettuce.core.internal.AbstractInvocationHandler.invoke(AbstractInvocationHandler.java:80)
        at com.sun.proxy.$Proxy168.get(Unknown Source)
        at org.springframework.data.redis.connection.lettuce.LettuceStringCommands.get(LettuceStringCommands.java:61)
        ... 21 common frames omitted

Lettuce 似乎使用了常见的 ForkJoinPool，DeferredResult 也使用了它，并且所有请求和连接都阻塞了池(如果我错了，请纠正我)。推荐的方法是什么？我应该将 Lettuce 移至其他池吗？如果是这样怎么办？如果我可以提供任何其他配置或其他信息，请告诉我。

Answer 1

Lettuce 使用 Netty 的 EventLoop 作为其线程基础设施。

这里发生的情况是您的任务在 ForkJoin 池上执行。 Lettuce 使用 CompletableFuture 返回异步结果处理的句柄，同步 API 调用 CompletableFuture.get(timeout, TimeUnit) 等待命令完成。在 ForkJoin 池上调用阻塞方法涉及到 ManagedBlocker 可能会切换到可以继续工作的不同线程。

如果太多线程等待命令完成，最终会出现 RejectedExecutionException。

我建议为您调用的 lambda 使用不同的执行池。