标签: reactor-netty

在调用下游RESTful服务时,我们看到太多"连接过早关闭并由同行重置"错误,即使在将Springboot版本升级到最新的专家建议之后也是如此.

我们试图重现它,但无法在dev中重现.它发生在负载测试中,当应用程序负载过重时.

堆栈跟踪:

Suppressed: java.io.IOException: Connection closed prematurely
    at reactor.ipc.netty.http.client.HttpClientOperations.onInboundClose(HttpClientOperations.java:269)
    at reactor.ipc.netty.channel.ChannelOperationsHandler.channelInactive(ChannelOperationsHandler.java:113)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:245)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:231)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:224)
    at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelInactive(CombinedChannelDuplexHandler.java:420)
    at io.netty.handler.codec.ByteToMessageDecoder.channelInputClosed(ByteToMessageDecoder.java:377)
    at io.netty.handler.codec.ByteToMessageDecoder.channelInactive(ByteToMessageDecoder.java:342)
    at io.netty.handler.codec.http.HttpClientCodec$Decoder.channelInactive(HttpClientCodec.java:282)
    at io.netty.channel.CombinedChannelDuplexHandler.channelInactive(CombinedChannelDuplexHandler.java:223)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:245)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:231)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:224)
    at io.netty.handler.codec.ByteToMessageDecoder.channelInputClosed(ByteToMessageDecoder.java:377)
    at io.netty.handler.codec.ByteToMessageDecoder.channelInactive(ByteToMessageDecoder.java:342)
    at io.netty.handler.ssl.SslHandler.channelInactive(SslHandler.java:1028)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:245)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:231)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:224)
    at io.netty.channel.DefaultChannelPipeline$HeadContext.channelInactive(DefaultChannelPipeline.java:1429)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:245)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:231)
    at io.netty.channel.DefaultChannelPipeline.fireChannelInactive(DefaultChannelPipeline.java:947)
    at io.netty.channel.AbstractChannel$AbstractUnsafe$8.run(AbstractChannel.java:822)
    at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163)
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:404)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:465)
    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:884)
    ... 1 common frames omitted

任何帮助和建议表示赞赏,提前谢谢.

我使用 Spring Webflux（带有 spring-reactor-netty）2.1.0.RC1 和 Lettuce 5.1.1.RELEASE。

当我使用 Reactive Lettuce API 调用任何 Redis 操作时，执行总是切换到同一个单独的线程 (lettuce-nioEventLoop-4-1)。

这导致性能不佳，因为所有执行都在该单个线程中遇到瓶颈。

我知道publishOn每次调用 Redis 时我都可以使用它来切换到另一个线程，但这很容易出错并且仍然不是最佳的。

有什么办法可以改善吗？我看到 Lettuce 提供了 ClientResources 类来自定义线程分配，但我找不到任何方法将其与 Spring webflux 集成。

此外，对于粗心的开发人员来说，当前的行为会不会很危险？也许默认值应该稍微调整一下。我想理想的情况是 Lettuce 可以重用来自 webflux 的相同事件循环。

我正在添加这个 spring boot 单类片段，可用于重现我所描述的内容：

@SpringBootApplication
public class ReactiveApplication {
    public static void main(String[] args) {
        SpringApplication.run(ReactiveApplication.class, args);
    }
}

@Controller
class TestController {

    private final RedisReactiveCommands<String, String> redis = RedisClient.create("redis://localhost:6379").connect().reactive();

    @RequestMapping("/test")
    public Mono<Void> test() {
        return redis.exists("key")
            .doOnSubscribe(subscription -> System.out.println("\nonSubscribe called on thread " + Thread.currentThread().getName())) …

我在 AWS 负载均衡器后面运行 spring cloud gateway（我理解它是在 Spring Webflux 上构建的），并且我收到间歇性 502 错误。经过调查，问题似乎与负载均衡器和我的节点之间的连接超时有关。从一些调查来看，底层 netty 服务器的默认超时时间为 10 秒。我使用以下命令确定了这一点...

time nc -vv 10.10.xx.xxx 5100
Connection to 10.10.xx.xxx 5100 port [tcp/*] succeeded!

real    0m10.009s
user    0m0.000s
sys     0m0.000s

虽然我可以将负载均衡器上的 idleTimeout 设置在 10 秒以内，但这感觉非常低效。如果可能，我想将其保持在 30 秒以上。相反，我想增加 netty 服务器上的连接超时。我试图在我的 application.yml 中设置 server.connection-timeout 属性...

server:
  connection-timeout: 75000

也通过指定秒...

server:
  connection-timeout: 75s

但是当我运行 time 命令以查看我的连接持续多长时间时，超时没有变化，它仍然以 10 秒结束......

time nc -vv 10.10.xx.xxx 5100
Connection to 10.10.xx.xxx 5100 port [tcp/*] succeeded!

real    0m10.009s
user    0m0.000s
sys     0m0.000s

我在这里缺少什么？

我在 Spring Boot MVC 2.1 项目中使用 WebClient，发现客户端发出的第一个请求最多需要 6 秒。后续请求速度更快（~30 毫秒）。

Spring 的 JIRA中有一个已解决的问题，建议使用 Jetty 作为 WebClient Http 连接器。我已经尝试过这种方法，以大约 800 毫秒的第一个请求改进了数字。这次是一个改进，但距离通常需要 < 200 毫秒的 RestTemplate 还很远。

Netty 方法（5s 第一个请求）：

会议：

@Bean
public WebClient webClient() {
    return WebClient.create();
}

用法：

private final WebClient webClient;

@GetMapping(value="/wc", produces = APPLICATION_JSON_UTF8_VALUE)
public Mono<String> findWc() throws URISyntaxException {
    URI uri = new URI("http://xxx");
    final Mono<String> response = webClient.get().uri(uri).retrieve().bodyToMono(String.class);
    return response;
}

码头方法（800ms 第一个请求）：

会议：

@Bean
public JettyResourceFactory resourceFactory() {
    return …

我正在使用 springboot webclient 从远程服务器调用 rest api。第一个请求工作正常。如果我在一段时间后发出后续请求，服务器会抛出 500 服务器错误。我得到的错误是“onError(java.io.IOException:一个现有的连接被远程主机强行关闭)”。

我想通过禁用连接池来测试行为，因为我相信它使用以前的连接。你能帮我在创建 webclient 时如何禁用连接池吗？

TcpClient tcpClient = TcpClient.create()
        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 30000)
        .option(ChannelOption.SO_KEEPALIVE, false)
        .doOnConnected(connection ->
                connection.addHandlerLast(new ReadTimeoutHandler(30))
                        .addHandlerLast(new WriteTimeoutHandler(30))
        );

ReactorClientHttpConnector httpConnector = new ReactorClientHttpConnector(HttpClient.from(tcpClient));

final WebClient webClient = WebClient
        .builder()
        .clientConnector(httpConnector)
        .baseUrl("http://customer.service.api.internal.cloud.qa.intranet.pagseguro.uol")
        .exchangeStrategies(strategies)
        .build()

Springboot Webclient 在尝试调用远程服务器中的 rest api 时抛出“一个现有连接被远程主机强行关闭”错误。

当服务器加载时，第一个服务器请求加载正常。当我在一段时间后（比如 5 分钟）发送第二个请求时，出现错误。

网页客户端创建代码：

WebClient webClient() {
           TcpClient tcpClient = TcpClient.create()
               .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 30000)
               .option(ChannelOption.SO_KEEPALIVE, false)
               .doOnConnected { connection ->
               connection.addHandlerLast(new LoggingHandler(LogLevel.TRACE))
           connection.addHandlerLast(new ReadTimeoutHandler(30))
                   .addHandlerLast(new WriteTimeoutHandler(30))
               .addHandlerLast(new IdleStateHandler(30,30,30))

       }

       ReactorClientHttpConnector httpConnector = new ReactorClientHttpConnector(HttpClient.from(tcpClient))

       return WebClient.builder()
               .clientConnector(httpConnector)
               .build()

   }

请在下面找到日志。看起来连接没有正确关闭。

你能告诉我如何解决吗？

2019-04-10 15:26:31.534  INFO 235344 --- [nio-8080-exec-2] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring DispatcherServlet 'dispatcherServlet'
2019-04-10 15:26:31.534  INFO 235344 --- [nio-8080-exec-2] o.s.web.servlet.DispatcherServlet        : Initializing Servlet 'dispatcherServlet'
2019-04-10 15:26:31.544  INFO 235344 --- [nio-8080-exec-2] o.s.web.servlet.DispatcherServlet        : Completed initialization in …

在Spring 5中，我们使用webclient通过SSL调用其他REST端点。当我们在wireshark上记录日志时，我们看到对于每个请求，SSL握手都在发生，并且花费大量时间。

这些是我的查询。

查询1：Webclient执行内部连接池。连接失效之前的到期时间是多少？

我们是否可以设置打开连接的时间，以便在已经定义的连接上发送新请求以避免SSL握手？

是否存在避免每个请求进行SSL握手的设置？

查询2：我们正在创建Webclient的bean，并在整个应用程序中使用它。我们需要做一些设置来启用连接池和重用现有连接以发送新请求。还有避免每次都进行SSL握手的任何设置。

    @Bean(name="productDefWebClient")
    public WebClient productDefWebClient() {
        ConnectionProvider elasticPool = ConnectionProvider.elastic("productDef-pool");
        HttpClient httpClient = HttpClient.create(elasticPool).wiretap(true);
        return WebClient.builder()
                .clientConnector(new ReactorClientHttpConnector(httpClient))
                .defaultHeader("Accept","application/json")
                .baseUrl("https://test-qa.test.com")
                .build();
    }

我一直在阅读有关 Java Fibers 作为映射到线程的小工作单元的内容。如果发生阻塞调用，不同的纤程将被映射到同一个线程。由于 Java 中的线程是内核级线程，因此这可以防止线程耗尽。

我一直在使用Spring Web-Flux，所以只是想了解当Netty服务器每秒接收100个请求时内部会发生什么，每个请求都包括反应式数据库访问，这些请求如何映射到Netty服务器默认生成的40个线程？

通量与光纤有何不同？Flux 如何保证线程数量有限的异步行为？

我在使用 Spring Data R2DBC DatabaseClient 将条件参数绑定到 SQL 查询时遇到困难。两个参数可以为空。由于DatabaseClient需要明确指定参数为空，因此我尝试了以下语法，但条件参数未附加到现有参数中：

public Mono<Void> createAddress(Address address) {
    DatabaseClient.GenericExecuteSpec bindings = databaseClient.execute(addressesQueries.getProperty("addresses.insert"))
            .bind("line1", address.getLine1())
            .bind("zipCode", address.getZipCode())
            .bind("city", address.getCity())
            .bind("countryId", address.getCountry())
            .bind("id", address.getId()); // UUID

            if(address.getLine2() == null) {
                bindings.bindNull("line2", String.class);
            } else {
                bindings.bind("line2", address.getLine2());
            }
            if(address.getState() == null) {
                bindings.bindNull("state", String.class);
            } else {
                bindings.bind("state", address.getState());
            }

    return bindings.fetch().rowsUpdated().then();
}

SQL查询：

INSERT INTO addresses(id,line1,line2,zip_code,city,state,country) VALUES(:id,:line1,:line2,:zipCode,:city,:state,:countryId)

我知道我可以拆分 SQL 查询来处理带/不带空参数的情况，但如果我有多个条件参数，就会有点复杂。

你知道一种解决方案可以帮助我保留一个 SQL 查询并在 Java 代码中处理条件参数吗？

我看到了

HttpClient.from(TcpClient.create().option(ChannelOption.SO_KEEPALIVE, true)) 

from方法已被弃用。

目前我应该如何设置SO_KEEPALIVE使用HttpClient.create()？