两个Quarkus服务之间的非阻塞数据流（Java中带有Mutiny的Vert.x）

Question

使现代化

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

背景信息：

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

问题：

现在，我正在尝试使用叛变多与垂直。Quarkus下的x，但无法确定消费者服务如何在不阻塞的情况下接收此流。

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

问题：

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

下面是一个简化的示例

测试实体

java prettyprint-override">public class SearchResult implements Serializable {

    private String content;

    public SearchResult(String content) {
        this.content = content;
    }


    //.. toString, getters and setters
    
}

生产商1。简单无限流-

@GET
@Path("/search")
@Produces(MediaType.SERVER_SENT_EVENTS)
@SseElementType(MediaType.APPLICATION_JSON)
public Multi<SearchResult> getResults() {
        return Multi.createFrom().ticks().every(Duration.ofSeconds(2)              .onItem().transform(n -> new SearchResult(n.toString()));
}

生产商2。Vertx路径无限流-

@Route(path = "/routed", methods = HttpMethod.GET)
public Multi<SearchResult> getSrStreamRouted(RoutingContext context) {
        log.info("routed run");
        return ReactiveRoutes.asEventStream(Multi.createFrom().ticks().every(Duration.ofSeconds(2))
                .onItem().transform(n -> new SearchResult(n.toString()));
}

制片人3。简单有限流-

@GET
@Path("/search")
@Produces(MediaType.SERVER_SENT_EVENTS)
@SseElementType(MediaType.APPLICATION_JSON)
public Multi<SearchResult> getResults() {
    return Multi.createFrom().ticks().every(Duration.ofSeconds(2))
        .transform().byTakingFirstItems(5)
        .onItem().transform(n -> new SearchResult(n.toString()));
}

消费者：

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

WebClientOptions webClientOptions = new WebClientOptions().setDefaultHost("localhost").setDefaultPort(8182);
WebClient client = WebClient.create(vertx, webClientOptions);
        
client.get("/string")
                .send()
                .onFailure().invoke(resp -> log.error("error: " + resp))
                .onItem().invoke(resp -> log.info("result: " + resp.statusCode()))
                .toMulti()
                .subscribe().with(r -> log.info(String.format("Subscribe: code:%d body:%s",r.statusCode(), r.bodyAsString())));

Answer 1

Vert. x Web客户端不适用于SSE（无配置）。从https://vertx.io/docs/vertx-web-client/java/：

响应已完全缓冲，请使用BodyCodec。管道到管道对写入流的响应

它一直等到响应完成。您可以使用原始的Vert. x HTTP客户端或使用管道编解码器。示例在https://vertx.io/docs/vertx-web-client/java/#_decoding_responses.

或者，您可以使用SSE客户端，例如：https://github.com/quarkusio/quarkus-quickstarts/blob/master/kafka-quickstart/src/test/java/org/acme/kafka/PriceResourceTest.java#L27-L34级