那些遇到过的问题

Spring Boot WebClient 停止发送请求

wxk*_*ker 11 java spring-boot spring-webclient

我正在运行一个 Spring Boot 应用程序，它使用 WebClient 来处理非阻塞和阻塞 HTTP 请求。应用程序运行一段时间后，所有传出的 HTTP 请求似乎都会被卡住。

WebClient 用于向多个主机发送请求，但作为示例，以下是它的初始化和用于向 Telegram 发送请求的方式：

网络客户端配置：

    @Bean
    public ReactorClientHttpConnector httpClient() {
        HttpClient.create(ConnectionProvider.builder("connectionProvider").build())
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeout)
                .responseTimeout(Duration.ofMillis(responseTimeout));
        return new ReactorClientHttpConnector(httpClient);
    }
Run Code Online (Sandbox Code Playgroud)

所有 WebClient 使用相同的 ReactorClientHttpConnector。

电报客户端：

    @Autowired
    ReactorClientHttpConnector httpClient;

    WebClient webClient;

    RateLimiter rateLimiter;

    @PostConstruct
    public void init() {
        webClient = WebClient.builder()
                .clientConnector(httpClient)
                .baseUrl(telegramUrl)
                .build();

        rateLimiter = RateLimiter.of("telegram-rate-limiter",
                RateLimiterConfig.custom()
                        .limitRefreshPeriod(Duration.ofMinutes(1))
                        .limitForPeriod(20)
                        .build());
    }

    public void sendMessage(@PathVariable("token") String token, @RequestParam("chat_id") long chatId, @RequestParam("text") String message) {
        webClient.post().uri(String.format("/bot%s/sendMessage", token))
                .contentType(MediaType.APPLICATION_JSON)
                .body(BodyInserters.fromFormData("chat_id", String.valueOf(chatId))
                        .with("text", message))
                .retrieve()
                .bodyToMono(Void.class)
                .transformDeferred(RateLimiterOperator.of(rateLimiter))
                .block();
    }
Run Code Online (Sandbox Code Playgroud)

RateLimiter 用于确保请求数量不超过 Telegram API 中指定的每分钟 20 个。

当应用程序启动时，所有请求都会按预期正常解决。但过了一段时间后，所有请求似乎都被卡住了。发生这种情况所需的时间可能从几个小时到几天不等。对于向不同主机发出的所有请求都会发生这种情况，并且当来自 TelegramBot 的消息停止时很容易注意到。一旦请求被卡住，它们就会无限期地被卡住，我必须重新启动应用程序才能使其再次工作。

日志中没有任何异常似乎导致了这种情况。由于我为电报消息维护了一个队列，因此当队列中的消息数量稳定增加以及等待请求解决的其他进程中发生错误时，我可以看到请求停止的时间点。

由于我设置的连接超时和响应超时没有生效，似乎请求甚至没有发送出去。

我之前也尝试过将空闲时间设置为0，但这并没有解决问题

    @Bean
    public ReactorClientHttpConnector httpClient() {
        HttpClient httpClient = HttpClient.create(ConnectionProvider.builder("connectionProvider").maxConnections(1000).maxIdleTime(Duration.ofSeconds(0)).build())
        HttpClient httpClient = HttpClient.newConnection()
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeout)
                .responseTimeout(Duration.ofMillis(responseTimeout));
        return new ReactorClientHttpConnector(httpClient);
    }
Run Code Online (Sandbox Code Playgroud)

更新：

我启用了指标，并在卡住时使用千分尺查看它。有趣的是，它显示有一个 Telegram 连接，但也显示没有空闲、挂起或活动的连接。

reactor_netty_connection_provider_idle_connections{id="-1268283746",name="connectionProvider",remote_address="api.telegram.org:443",} 0.0
reactor_netty_connection_provider_pending_connections{id="-1268283746",name="connectionProvider",remote_address="api.telegram.org:443",} 0.0
reactor_netty_connection_provider_active_connections{id="-1268283746",name="connectionProvider",remote_address="api.telegram.org:443",} 0.0
reactor_netty_connection_provider_total_connections{id="-1268283746",name="connectionProvider",remote_address="api.telegram.org:443",} 1.0
Run Code Online (Sandbox Code Playgroud)

问题可能是缺少连接吗？

更新2：

我认为这可能与另一个问题有关：Closing Reactor Netty connection on error status code

所以我将 HttpClient 更新为：

reactor_netty_connection_provider_idle_connections{id="-1268283746",name="connectionProvider",remote_address="api.telegram.org:443",} 0.0
reactor_netty_connection_provider_pending_connections{id="-1268283746",name="connectionProvider",remote_address="api.telegram.org:443",} 0.0
reactor_netty_connection_provider_active_connections{id="-1268283746",name="connectionProvider",remote_address="api.telegram.org:443",} 0.0
reactor_netty_connection_provider_total_connections{id="-1268283746",name="connectionProvider",remote_address="api.telegram.org:443",} 1.0
Run Code Online (Sandbox Code Playgroud)

但这似乎只会加速问题的发生。就像以前一样，活动连接、挂起连接和空闲连接不会合计到总连接数。总和始终大于其他 3 个指标的总和。

更新 3： 问题发生时我进行了线程转储。总共有 74 个线程，所以我认为应用程序没有用完线程。

Telegram 线程的转储：

"TelegramBot" #20 daemon prio=5 os_prio=0 cpu=14.65ms elapsed=47154.24s tid=0x00007f6b28e73000 nid=0x1c waiting on condition  [0x00007f6aed6fb000]
   java.lang.Thread.State: WAITING (parking)
        at jdk.internal.misc.Unsafe.park(java.base@11.0.13/Native Method)
        - parking to wait for  <0x00000000fa865c80> (a java.util.concurrent.CountDownLatch$Sync)
        at java.util.concurrent.locks.LockSupport.park(java.base@11.0.13/LockSupport.java:194)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(java.base@11.0.13/AbstractQueuedSynchronizer.java:885)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(java.base@11.0.13/AbstractQueuedSynchronizer.java:1039)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(java.base@11.0.13/AbstractQueuedSynchronizer.java:1345)
        at java.util.concurrent.CountDownLatch.await(java.base@11.0.13/CountDownLatch.java:232)
        at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:87)
        at reactor.core.publisher.Mono.block(Mono.java:1707)
        at com.moon.arbitrage.cm.feign.TelegramClient.sendMessage(TelegramClient.java:59)
        at com.moon.arbitrage.cm.service.TelegramService.lambda$sendArbMessage$0(TelegramService.java:53)
        at com.moon.arbitrage.cm.service.TelegramService$$Lambda$1092/0x000000084070f840.run(Unknown Source)
        at com.moon.arbitrage.cm.service.TelegramService.task(TelegramService.java:82)
        at com.moon.arbitrage.cm.service.TelegramService$$Lambda$920/0x0000000840665040.run(Unknown Source)
        at java.lang.Thread.run(java.base@11.0.13/Thread.java:829)

   Locked ownable synchronizers:
        - None
Run Code Online (Sandbox Code Playgroud)

反应堆工作线程：

"reactor-http-epoll-1" #15 daemon prio=5 os_prio=0 cpu=810.44ms elapsed=47157.07s tid=0x00007f6b281c4000 nid=0x17 runnable  [0x00007f6b0c46c000]
   java.lang.Thread.State: RUNNABLE
        at io.netty.channel.epoll.Native.epollWait0(Native Method)
        at io.netty.channel.epoll.Native.epollWait(Native.java:177)
        at io.netty.channel.epoll.EpollEventLoop.epollWait(EpollEventLoop.java:286)
        at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:351)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)
        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(java.base@11.0.13/Thread.java:829)

   Locked ownable synchronizers:
        - None

"reactor-http-epoll-2" #16 daemon prio=5 os_prio=0 cpu=1312.16ms elapsed=47157.07s tid=0x00007f6b281c5000 nid=0x18 waiting on condition  [0x00007f6b0c369000]
   java.lang.Thread.State: WAITING (parking)
        at jdk.internal.misc.Unsafe.park(java.base@11.0.13/Native Method)
        - parking to wait for  <0x00000000fa865948> (a java.util.concurrent.CompletableFuture$Signaller)
        at java.util.concurrent.locks.LockSupport.park(java.base@11.0.13/LockSupport.java:194)
        at java.util.concurrent.CompletableFuture$Signaller.block(java.base@11.0.13/CompletableFuture.java:1796)
        at java.util.concurrent.ForkJoinPool.managedBlock(java.base@11.0.13/ForkJoinPool.java:3128)
        at java.util.concurrent.CompletableFuture.waitingGet(java.base@11.0.13/CompletableFuture.java:1823)
        at java.util.concurrent.CompletableFuture.get(java.base@11.0.13/CompletableFuture.java:1998)
        at com.moon.arbitrage.cm.service.OrderService.reconcileOrder(OrderService.java:103)
        at com.moon.arbitrage.cm.service.BotService$BotTask.lambda$task$1(BotService.java:383)
        at com.moon.arbitrage.cm.service.BotService$BotTask$$Lambda$1161/0x00000008400af440.accept(Unknown Source)
        at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onNext(MonoPeekTerminal.java:171)
        at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onNext(MonoPeekTerminal.java:180)
        at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1816)
        at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:249)
        at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:79)
        at reactor.core.publisher.FluxOnAssembly$OnAssemblySubscriber.onNext(FluxOnAssembly.java:539)
        at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1816)
        at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:151)
        at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onNext(FluxContextWrite.java:107)
        at reactor.core.publisher.FluxMapFuseable$MapFuseableConditionalSubscriber.onNext(FluxMapFuseable.java:295)
        at reactor.core.publisher.FluxFilterFuseable$FilterFuseableConditionalSubscriber.onNext(FluxFilterFuseable.java:337)
        at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1816)
        at reactor.core.publisher.MonoCollect$CollectSubscriber.onComplete(MonoCollect.java:159)
        at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:142)
        at reactor.core.publisher.FluxPeek$PeekSubscriber.onComplete(FluxPeek.java:260)
        at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:142)
        at reactor.netty.channel.FluxReceive.onInboundComplete(FluxReceive.java:400)
        at reactor.netty.channel.ChannelOperations.onInboundComplete(ChannelOperations.java:419)
        at reactor.netty.channel.ChannelOperations.terminate(ChannelOperations.java:473)
        at reactor.netty.http.client.HttpClientOperations.onInboundNext(HttpClientOperations.java:702)
        at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:93)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
        at io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:286)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
        at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
        at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
        at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:296)
        at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
        at io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1372)
        at io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1235)
        at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1284)
        at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:507)
        at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:446)
        at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
        at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
        at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
        at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:795)
        at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:480)
        at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:378)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)
        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(java.base@11.0.13/Thread.java:829)

   Locked ownable synchronizers:
        - None

"reactor-http-epoll-3" #17 daemon prio=5 os_prio=0 cpu=171.84ms elapsed=47157.07s tid=0x00007f6b28beb000 nid=0x19 runnable  [0x00007f6b0c26a000]
   java.lang.Thread.State: RUNNABLE
        at io.netty.channel.epoll.Native.epollWait0(Native Method)
        at io.netty.channel.epoll.Native.epollWait(Native.java:177)
        at io.netty.channel.epoll.EpollEventLoop.epollWait(EpollEventLoop.java:281)
        at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:351)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)
        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(java.base@11.0.13/Thread.java:829)

   Locked ownable synchronizers:
        - None

"reactor-http-epoll-4" #18 daemon prio=5 os_prio=0 cpu=188.10ms elapsed=47157.07s tid=0x00007f6b28b7d800 nid=0x1a runnable  [0x00007f6b0c169000]
   java.lang.Thread.State: RUNNABLE
        at io.netty.channel.epoll.Native.epollWait0(Native Method)
        at io.netty.channel.epoll.Native.epollWait(Native.java:177)
        at io.netty.channel.epoll.EpollEventLoop.epollWait(EpollEventLoop.java:281)
        at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:351)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986)
        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(java.base@11.0.13/Thread.java:829)

   Locked ownable synchronizers:
        - None
Run Code Online (Sandbox Code Playgroud)

似乎其中一个任务被另一个任务阻塞（甚至不是来自 Telegram 服务），但这不应该成为问题，因为其他三个工作线程是可以运行的，对吧？

wxk*_*ker 6

终于找到并解决了问题。问题是我有一个阻塞任务被反应器线程阻塞。由于线程转储，我才注意到这一点。阻塞任务正在等待事件，因此可能需要很长时间才能解决。因此，最终，当所有四个反应器线程都被阻塞时，所有请求自然会被卡住，没有线程来处理它们。

简而言之：不要阻塞你的反应器线程。

归档时间：

查看次数：

4295 次

最近记录：

2 年，10 月 前
相关归档
在准备好的声明中使用"like"通配符 159
隐藏实用程序类构造函数:实用程序类不应具有公共或默认构造函数 85
"final int i"如何在Java for循环中工作？ 60
新的/奇怪的Java"try()"语法？ 53
Spring @ContextConfiguration如何为xml放置正确的位置 50
将HashMap转换为ArrayList 44
使用Spring数据创建一个上限集合MongoDB @Document 15
连接PostgreSQL数据库时Spring启动应用程序错误 7
Spring boot似乎找不到mybatis映射器配置（Mapped statements集合不包含值） 5
Axon 4 XStream 配置 5
难疑归档
可以在JSON中使用注释吗？ 7104
C++中指针变量和引用变量之间有什么区别？ 3115
何时在Java中使用LinkedList而不是ArrayList？ 2974
提高SQLite的每秒INSERT性能？ 2880
如何"git克隆"包括子模块？ 1864
如何在C#中生成随机int数？ 1792
<meta http-equiv ="X-UA-Compatible"content ="IE = edge">是做什么用的？ 1378
使用*args和**kwargs 1367
从字典中删除元素 1243
如何删除选择框的所有选项,然后添加一个选项并使用jQuery选择它？ 1028