标签: webflux

我是 spring webflux 的初学者。在研究时我发现了一些代码，例如：

Mono result = someMethodThatReturnMono().cache();

“缓存”这个名字告诉我关于缓存某些东西，但是缓存在哪里以及如何检索缓存的东西？是咖啡因之类的东西吗？

我试图了解反应式编程以及事件循环的工作原理。只要我能理解何时向应用程序发送新的 HTTP 请求，该请求就会分解为事件。每个事件都由该事件循环处理，并且为每个事件注册一个回调。此事件完成后，结果将返回到请求。但是有多少线程处理此请求以及此事件循环中有多少线程。

我目前正在使用 spring boot webflux 进行开发。我正在使用 websocket 客户端和 API 客户端，但此错误在部署后几个小时发生。我无法弄清楚这个错误是从哪里来的。这个问题从何而来？

错误日志如下。

[reactor-http-epoll-2] r.n.http.server.HttpServerOperations     : [9c0c3ede, L:/172.25.0.8:8080 - R:/94.232.43.63:34534] Decoding failed: DefaultFullHttpRequest(decodeResult: failure(java.lang.IllegalArgumentException: text is empty (possibly HTTP/0.9)), version: HTTP/1.0, content: UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeHeapByteBuf(ridx: 0, widx: 0, cap: 0))
GET /bad-request HTTP/1.0 :

java.lang.IllegalArgumentException: text is empty (possibly HTTP/0.9)
        at io.netty.handler.codec.http.HttpVersion.valueOf(HttpVersion.java:65) ~[netty-codec-http-4.1.72.Final.jar:4.1.72.Final]
        at io.netty.handler.codec.http.HttpRequestDecoder.createMessage(HttpRequestDecoder.java:126) ~[netty-codec-http-4.1.72.Final.jar:4.1.72.Final]
        at io.netty.handler.codec.http.HttpObjectDecoder.decode(HttpObjectDecoder.java:273) ~[netty-codec-http-4.1.72.Final.jar:4.1.72.Final]
        at io.netty.handler.codec.http.HttpServerCodec$HttpServerRequestDecoder.decode(HttpServerCodec.java:140) ~[netty-codec-http-4.1.72.Final.jar:4.1.72.Final]
        at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:507) ~[netty-codec-4.1.72.Final.jar:4.1.72.Final]
        at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:446) ~[netty-codec-4.1.72.Final.jar:4.1.72.Final]

有时，还会发现破损的字符以及“文本为空”这句话。如果这是编码问题，我应该如何设置？

在我们的 kubernetes 环境中，我们的 pod 通常保留的 cpu 核心少于 1 个。知道 spring webflux 与 eventloop+workers 的概念一起工作，那它是如何工作的呢？是否建议我们为该 Pod 预留至少 1 个 cpu 核心？

如果我仍然在 kubernetes 中使用少于 1cpu 请求的 webflux，我的 eventloop 会表现不佳吗？

在单元测试重试期间，模拟的响应似乎已缓存，或者很可能我做错了什么。

我正在尝试请求某些内容，如果发生错误，则重试两次，延迟 1 秒。

  public Mono<Object> someMethod(String someParam) {
    return someInjectedService.doSomething(someParam)
        .doOnError(ex -> System.out.println(ex + ": " + System.currentTimeMillis()))
        .retryWhen(Retry.fixedDelay(2, Duration.ofSeconds(1)).filter(ex -> ex instanceof SomeCustomException))
        .doOnSuccess(result -> doSomethingOnSuccess(result));
  }

我的测试：

  @Test
  void testshouldRequestThrice_whenErrorOccurs() {
    // Given
    String someParam = "testParam";
    when(someInjectedService.doSomething(someParam))
        .thenReturn(Mono.error(new SomeCustomException("SomeCustomException"))) // 1st response
        .thenReturn(Mono.error(new SomeCustomException("SomeCustomException"))) // 2nd response
        .thenReturn(Mono.just("SomeValidResponse")); // 3rd valid response

    // When
    var result = testService.someMethod(someParam).block();

    // Then
    // Initial request, followed by two retries
    verify(someInjectedService, times(3)).doSomething(someParam);
  }

这someInjectedService是一个模拟。我的计划是返回异常两次，并在第三次请求时返回有效响应。但我得到的是：

org.mockito.exceptions.verification.TooFewActualInitations: someInjectedService.doSomething("testParam");

通缉 …

我已经研究 rsocket 和响应式编程有一段时间了，经常提到 spring webflux，我想问 rsocket 和 spring webflux 之间有什么区别，或者它们是相同的东西。谢谢

我正在使用 r2dc 作为 Spring Boot Java 应用程序。

我在想是否可以将 Flux 转换为 Mono 以进行某些计算。

伪示例：

static PseudoMagic calculate(List<Foo> foos) {
   return callTheMagicRutine(foos)
}

Mono<PseudoMagic> getMyMagic() {
   Flux<Foo> foos = getMyFoos()
   foos.transformToMagic(f -> calculator(f))
}