问题标签 [mutiny]

更新！

在解决了一些与主要问题无关的问题后，我修复了示例代码中的小错误，主要问题仍然是关于服务之间的非阻塞流。

背景资料：

我在 Quarkus 下移植了一个 Spring WebFlux 服务。该服务对多个大型数据集运行长时间搜索，并在可用时以 Flux（文本/事件流）的形式返回部分结果。

问题：

现在我正在尝试在 Quarkus 下将 Mutiny Multi 与 Vert.x 一起使用，但无法弄清楚消费者服务如何在不阻塞的情况下接收此流。

在所有示例中，消费者要么是 JS 前端页面，要么生产者的内容类型是 application/json，这似乎在 Multi 完成之前一直在发送，然后再将其发送到一个 JSON 对象中（这在我的应用程序中没有意义）。

问题：

1. 如何使用 Mutiny 风格的 Vert.x WebClient 接收文本/事件流？
2. 如果问题是 WebClient 无法接收连续流：在两个 Quarkus 服务之间流式传输数据的标准方式是什么？

这是一个简化的例子

测试实体

生产者 1. 简单的无限流 -> 挂起

Producer 2. 使用 Vertx Paths 无限流 -> 挂起

Producer 3. 简单的有限流 -> 阻塞直到完成

消费者：

在生产者和消费者方面尝试了多种不同的解决方案，但在每种情况下，流都会阻塞，直到它完成或无限期挂起，而不为无限流传输数据。我用 httpie 得到了相同的结果。这是最新的迭代：

在 quarkus / kotlin 应用程序中，我想同时启动多个数据库请求。我是 quarkys 的新手，我不确定我是否做对了：

真的会request()并行吗？这是在quarkus中做的正确方法吗？

从一个 quarkus (kotlin) 应用程序，我并行请求多个端点（相同的端点但具有不同的基本 URL），然后我组合 uni 以便不按顺序等待每个响应。

这是一个示例：

我遇到的问题是ifNoItem在超时的情况下不使用。如果单个 uni 花费超过 1 毫秒（此值仅用于测试...），则最后一个块在combinedWith(...).await().

我想在单个 uni 级别处理超时，并将此错误视为失败（并记录和恢复）。但这似乎不起作用。我的实现来自https://smallrye.io/smallrye-mutiny/#_how_do_i_handle_timeout）。

如何正确地做到这一点？

其他响应式库（如项目反应器）为发布者提供排序方法，但在兵变中没有这种方法。他们的文档甚至没有谈论它。

https://smallrye.io/smallrye-mutiny

现在我正在通过这样做来实现上述功能

有没有更好的方法来做同样的事情？

在Quarkus和SmallRye Mutiny工作，我不知道如何处理补偿。

我有一个员工列表，我必须更新数据库中的所有员工。数据访问对象 (EmployeeDao) 具有以下方法：public Uni<Boolean> update(Long id, String name, Long deptId). 它期望接收员工的字段。

另外，我必须返回一个Uni<Boolean>通知所有更新是否正常。如下所示：

我的问题是，我怎样才能一一迭代执行更新过程并处理补偿？

如果任何员工更新过程失败，我想尝试几次，然后，如果我无法更新员工，我必须在返回任何响应之前撤消所有员工。这个想法是使数据库保持一致的状态。

如何使用 Mutiny 处理全部补偿？

我尝试使用 QUARKUS 和 SmallRye 学习反应式编程/反应式消息传递。但是我仍然很难理解 org.eclipse.microprofile.reactive.messaging 相对于 javax.enterprise.event 的优势，如果有的话？

在一边 ：

另一方面 ：

我可能会错过一些东西......

我一直在使用 Quarkus 并尝试使用 Mutiny Vertx WebClient。我的代码有效，但我不想依赖不安全/未经检查的分配，这就是我目前在 HttpResponse 上使用 bodyAsJson 方法编写代码的方式。是否有更好的方法或更标准的方法从 Mutiny Vertx 客户端解码 JSON？我意识到我可以调用 bodyAsJsonObject 并返回它，但我需要对从 API 调用返回的数据进行处理，因此我需要将其解码为表示数据形状/结构的类。

我正在 quarkus ( 1.8.3.Final) 中实现一个 gprc 服务器。我的服务是用反应式编写的（smallrye mutiny）这是我的服务类

现在写在里面的实际逻辑executeMyLogic有点阻塞，导致vertx阻塞事件循环警告（和一些其他错误）。因此，正如 quarkus grpc 服务器指南（https://quarkus.io/guides/grpc-getting-started）中所述，我用@Blocking (io.smallrye.common.annotation.Blocking).

在我添加这个注释之前，我在 sys.out 上得到了这个日志， grpc thread name vert.x-eventloop-thread-0 这表明这个逻辑正在一个似乎导致问题的 vertx 事件循环上运行。

现在根据我的理解，在添加此 @Blocking 注释后executeMyLogic应该在某个工作线程上运行。但它仍然在 vertx 事件循环上运行。

看起来这个注释没有被框架所尊重。

如果我的理解有误，请纠正我，否则请帮助我完成这项工作。

假设我们有：

• 一个 URL 列表，它是我们 Multi 的来源
• 作为第一步，我们使用 HTTP 客户端调用获取此页面的 HTML
• 然后我们尝试找到一些特定的标签并获取它的内容
• 然后我们将找到的东西存储到数据库中

现在我们在这里有 3 个步骤。有没有办法可以并行运行这些步骤？我的意思是一段时间后它应该：抓取 HTML 并同时处理 html + 获取标签内容，同时将数据从已处理的项目中保存到数据库中。（希望我的意思很明显）这样我们就可以进行并行处理。默认情况下，我可以看到，mutiny 以串行方式进行。

这是一个例子：

现在这会报告以下控制台输出：

可以看到它没有并行运行。我在这里错过了什么？

在 Quarkus 上，我尝试使用动态 baseUrl 构建反应式微配置文件 REST 客户端，但到目前为止，在所有实现变体中，REST 调用在订阅后都会阻塞。有趣的是，非响应式实现就像一个魅力。让我们看一些代码...

REST 客户端界面：

Quarkus 测试：

在运行 4 个测试时，只有使用非反应式 API 的 testRetrieve2 是成功的。所有其他人都挂在 AssertSubscriber.await() 中。

有任何想法吗？