为什么 RSocket 连接重试每次都使用多个（不同的）线程

Question

我有以下程序连接到 localhost:7999 上的 Spring 引导 rsocket 服务器运行。我已经配置了连接器 Retry.fixedDelay(Integer.MAX_VALUE, Duration.ofSeconds(5)) 如您所见，RSocketRequester 是 Mono，因此它应该保持单个连接。当连接失败并且重试开始时，我看到每次重试都是从不同的线程进行的，即如下 parallel-1---parallel-8。我可以知道这背后的原因吗？

12:08:24.463550|parallel-1|WARN |RSocketRefDataReceiver          |doAfterRetry===>attempt #1 (1 in a row), last failure={io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: no further information: localhost/127.0.0.1:7999}
12:08:30.470593|parallel-2|WARN |RSocketRefDataReceiver          |doAfterRetry===>attempt #2 (2 in a row), last failure={io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: no further information: localhost/127.0.0.1:7999}
12:08:36.475666|parallel-3|WARN |RSocketRefDataReceiver          |doAfterRetry===>attempt #3 (3 in a row), last failure={io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: no further information: localhost/127.0.0.1:7999}
12:08:42.494801|parallel-4|WARN |RSocketRefDataReceiver          |doAfterRetry===>attempt #4 (4 in a row), last failure={io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: no further information: localhost/127.0.0.1:7999}
12:08:48.499084|parallel-5|WARN |RSocketRefDataReceiver          |doAfterRetry===>attempt #5 (5 in a row), last failure={io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: no further information: localhost/127.0.0.1:7999}
12:08:54.503385|parallel-6|WARN |RSocketRefDataReceiver          |doAfterRetry===>attempt #6 (6 in a row), last failure={io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: no further information: localhost/127.0.0.1:7999}
12:09:00.509830|parallel-7|WARN |RSocketRefDataReceiver          |doAfterRetry===>attempt #7 (7 in a row), last failure={io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: no further information: localhost/127.0.0.1:7999}
12:09:06.545815|parallel-8|WARN |RSocketRefDataReceiver          |doAfterRetry===>attempt #8 (8 in a row), last failure={io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: no further information: localhost/127.0.0.1:7999}
12:09:12.553582|parallel-1|WARN |RSocketRefDataReceiver          |doAfterRetry===>attempt #9 (9 in a row), last failure={io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: no further information: localhost/127.0.0.1:7999}

我的程序如下：

RSocketStrategies strategies = RSocketStrategies.builder()
    .encoders(e -> e.add(new Jackson2CborEncoder()))
    .decoders(e -> e.add(new Jackson2CborDecoder()))
    .build();

Mono<RSocketRequester> requester = Mono.just(RSocketRequester.builder()
     .rsocketConnector(connector -> {
           connector.reconnect(
                     Retry.fixedDelay(Integer.MAX_VALUE, Duration.ofSeconds(5))
                     .doAfterRetry(e -> LOG.warn("doAfterRetry===>{}", e)))
             .acceptor(RSocketMessageHandler.responder(strategies,this))
             .payloadDecoder(PayloadDecoder.ZERO_COPY);
            })
      .dataMimeType(MediaType.APPLICATION_CBOR)
      .setupRoute("test")
      .setupData("test-123")
      .rsocketStrategies(strategies)
      .tcp("localhost",7999));

Answer 1

这篇文章 (Flight of the Flux 3) 很好地介绍了 Spring Reactor 线程模型。 Reactor 是在 rsocket-java.

中提供 Rx 功能实现的基础库

重点句子是

Schedulers.parallel() is good for CPU-intensive but short-lived tasks. It can execute N such tasks in parallel (by default N == number of CPUs)

另请阅读 https://projectreactor.io/docs/core/release/api/reactor/core/scheduler/Schedulers.html

如果保证所有操作都在单个线程上进行，那么它很可能会导致嘈杂的延迟，因为最初碰巧获得相同线程的两个不同客户端将在程序的整个生命周期内竞争线程。所以它比在有限的线程池之间平均分配一般工作要好。

Answer 2

感谢@Yuri @bruto @OlegDokuka 以及您的建议和回答。我已如下更改我的程序以在单线程上强制重试运行。

connector.reconnect(
        Retry.fixedDelay(Integer.MAX_VALUE, Duration.ofSeconds(5))
        .scheduler(Schedulers.single()) // <---- This enforces retry to run on a single thread
        .doAfterRetry(e -> LOG.warn("doAfterRetry===>{}", e)))
        .acceptor(RSocketMessageHandler.responder(strategies,this))
        .payloadDecoder(PayloadDecoder.ZERO_COPY);
      })

为什么 RSocket 连接重试每次都使用多个（不同的）线程

Why RSocket connection retry is using multiple (different) threads every time

rsocket

spring-boot-rsocket