标签: project-reactor

我必须重新实现一些后端服务，主要要求之一是使整个流程具有反应性。以前，服务使用 PostgreSQL 休眠，因此上述连接是由框架提供的。

由于我必须保留原始数据库并更改服务实现，因此我必须使用 r2dbc-postgresql。我找不到关于这个主题的任何资源，但我最好的猜测是做一些类似于我会用 JDBC 做的事情，并在我的实体之间引入一些新的连接表。

1. 这是一种正确的方法还是我应该考虑一些不同的解决方案？
2. 实现上述连接的步骤是什么？

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

Mono result = someMethodThatReturnMono().cache();

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

我看过Spring Tips:使用Spring Framework 5.0的功能反应端点,并阅读了一些关于spring reactor的内容,但我无法理解它.

鉴于我有netty和spring reactor有效,让端点返回Flux/ Mono实例(jacksonified)而不是直接dto对象(jacksonified)有什么好处？我最初假设在http请求/响应上下文中,反应流将更像是websockets,其中服务器通过开放通道将数据推送到接收器,但似乎并非如此.

另外netty在反应式编程中实际上比tomcat做得更好？

如果这些问题看起来很愚蠢,我很抱歉,但我不太明白这个新框架方向的目的.它为什么会出现,它是如何工作的以及它解决了哪些问题？

如何将1个元素的Flux转换为Mono？

Flux.fromArray(arrayOf(1,2,1,1,1,2))
                .distinct()
                .take(1)

我如何使这成为单声道(1)？

我无法理解在构建WebClient请求时我做错了什么.我想了解实际的HTTP请求是什么样的.(例如,将原始请求转储到控制台)

POST /rest/json/send HTTP/1.1
Host: emailapi.dynect.net
Cache-Control: no-cache
Postman-Token: 93e70432-2566-7627-6e08-e2bcf8d1ffcd
Content-Type: application/x-www-form-urlencoded

apikey=ABC123XYZ&from=example%40example.com&to=customer1%40domain.com&to=customer2%40domain.com&to=customer3%40domain.com&subject=New+Sale+Coming+Friday&bodytext=You+will+love+this+sale.

我正在使用Spring5的反应工具来构建API.我有一个实用工具类,它将使用Dyn的电子邮件api发送电子邮件.我想使用新的WebClient类来完成这个(org.springframework.web.reactive.function.client.WebClient)

以下命令取自:https://help.dyn.com/email-rest-methods-api/sending-api/#postsend

curl --request POST "https://emailapi.dynect.net/rest/json/send" --data "apikey=ABC123XYZ&from=example@example.com&to=customer1@domain.com&to=customer2@domain.com&to=customer3@domain.com&subject=New Sale Coming Friday&bodytext=You will love this sale."

当我使用真实值在curl中进行调用时,电子邮件正确发送,因此我觉得我正在生成错误的请求.

我的发送命令

public Mono<String> send( DynEmailOptions options )
{
    WebClient webClient = WebClient.create();
    HttpHeaders headers = new HttpHeaders();
    // this line causes unsupported content type exception :(
    // headers.setContentType( MediaType.APPLICATION_FORM_URLENCODED );
    Mono<String> result = webClient.post()
        .uri( "https://emailapi.dynect.net/rest/json/send" )
        .headers( headers )
        .accept( MediaType.APPLICATION_JSON )
        .body( BodyInserters.fromObject( options …

我在一些Spring webflux代码中遇到了Mono.defer（）

我在文档中查找了该方法，但不理解其中的解释：

“创建一个Mono提供程序，该提供程序将提供目标Mono以便为每个下游的订阅者进行订阅”

请给我一个解释和一个例子。我可能会参考一堆Reactor示例代码（它们的单元测试？）的地方。

谢谢

在 spring 项目反应堆中，onErrorResume和之间有什么区别doOnError？什么时候我应该每个人？

我们在生产中使用spring-data-r2dbc:1.3.2并dev.miku:r2dbc-mysql:0.8.2.RELEASE遇到了一个奇怪的问题。

我们不明白根本原因是什么，也不知道它是否是可恢复的，或者是否真的会产生不可预测的结果（如日志所示）。

我们定期看到错误日志如下：

• 记录器：dev.miku.r2dbc.mysql.client.ReactorNettyClient
• 级别：错误
• 消息：交换已取消，交换处于活动状态。这可能是一个导致不可预测结果的错误。

我们每天都会看到它几次，这让我们感到紧张，因为它明确指出：这可能是一个导致不可预测结果的错误

我们在日志中没有得到任何附加信息。

几个问题：

1. 这真的很危险吗？还是我们可以将其视为警告？
2. 我们如何调试它以了解其原因？
3. 从源代码中我们可以看到，它假设交换请求如果未完成则无法取消，并认为这种情况是一个BUG。是否有意义？如果反应流由于“连接丢失”或任何其他原因被取消怎么办？

当响应为 5xx 时，我想在等待 10 秒后重试请求 3 次。但我没有看到可以使用的方法。在对象上

WebClient.builder()
                .baseUrl("...").build().post()
                .retrieve().bodyToMono(...)

我可以看到方法：

在重试次数但没有延迟的条件下重试

.retry(3, {it is WebClientResponseException && it.statusCode.is5xxServerError} )

重试退避和次数但没有条件

.retryBackoff 

还有一个，retryWhen但我不知道如何使用它

我有以下代码：

import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;
import reactor.core.publisher.Mono;

@Component
public class GreetingHandler 
    public Mono<ServerResponse> hello(ServerRequest request) {
        return ServerResponse.ok().contentType(MediaType.TEXT_PLAIN)
        .body(BodyInserters.fromValue("Hello Spring!"));
    }
}

我理解这段代码，除了 Mono 类做什么以及它的功能是什么。我进行了大量搜索，但并没有直截了当：Mono 类是什么以及何时使用它？