叛变——Kafka写下了连续发生的事情

Question

我是Quarkus的新手。我正在尝试使用quarkus reactive编写一个RESTendpoint，它接收输入，进行一些验证，将输入转换为列表，然后将消息写入kafka。我的理解是，将所有内容转换为Uni/Multi，将导致以异步方式在I/O线程上执行。在intelliJ日志中，我可以看到代码在executor线程中以顺序方式执行。Kafka写入在其自己的网络线程中按顺序进行，这会增加延迟。

@POST
    @Consumes(MediaType.APPLICATION_JSON)
    @Produces(MediaType.APPLICATION_JSON)
    public Multi<OutputSample> send(InputSample inputSample) {
        ObjectMapper mapper = new ObjectMapper();

       //deflateMessage() converts input to a list of inputSample
        Multi<InputSample> keys = Multi.createFrom().item(inputSample)
                .onItem().transformToMulti(array -> Multi.createFrom().iterable(deflateMessage.deflateMessage(array)))
                .concatenate();

     
        return keys.onItem().transformToUniAndMerge(payload -> {
            try {
                return producer.writeToKafka(payload, mapper);
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
            return null;
        });
    } 

@Inject
@Channel("write")
    Emitter<String> emitter;

Uni<OutputSample> writeToKafka(InputSample kafkaPayload, ObjectMapper mapper) throws JsonProcessingException {
        
        String inputSampleJson = mapper.writeValueAsString(kafkaPayload);
        return Uni.createFrom().completionStage(emitter.send(inputSampleJson))
                .onItem().transform(ignored -> new OutputSample("id", 200, "OK"))
                .onFailure().recoverWithItem(new OutputSample("id", 500, "INTERNAL_SERVER_ERROR"));
    }

我已经用了几天了。不确定是否做错了什么。任何帮助都将不胜感激。谢谢

Answer 1

Multi用于源连续发射项目，直到发射完成事件，而这不是您的情况。

来自叛变文件：

Multi表示数据流。流可以发出0、1、n或无限多个项。

您很少自己创建多实例，而是使用公开Mutiny API的反应式客户端。

你要找的是一个Uni

所以，你需要的是将每条信息发送到Kafka，而不是立即等待它们的返回，而是收集生成的Uni，然后将其收集到单个Uni。

@POST
public Uni<List<OutputSample>> send(InputSample inputSample) {
    // This could be injected directly inside your producer
    ObjectMapper mapper = new ObjectMapper();

    // Send each item to Kafka and collect resulting Unis
    List<Uni<OutputSample>> uniList = deflateMessage(inputSample).stream()
            .map(input -> producer.writeToKafka(input, mapper))
            .collect(Collectors.toList());

    // Transform a list of Unis to a single Uni of a list
    @SuppressWarnings("unchecked") // Mutiny API fault...
    Uni<List<OutputSample>> result = Uni.combine().all().unis(uniList)
            .combinedWith(list -> (List<OutputSample>) list);

    return result;
}

Answer 2

与任何其他反应式库一样，Mutiny主要围绕数据流控制进行设计。

也就是说，它的核心将提供一组控制流执行和调度的功能（通常通过一些操作员）。这意味着，除非您指示社区对象异步，否则它们将以顺序（旧）方式执行。

执行调度使用两个运算符进行控制：

• runSubscriptionOn：这将导致生成项的代码段（通常指上游）在指定的执行器的线程上执行
• emitOn：这将导致订阅代码（通常指下游）在指定的执行器的线程上执行

然后，您可以按如下方式更新您的代码，使通货紧缩变得异步：

Multi<InputSample> keys = Multi.createFrom()
    .item(inputSample)
    .onItem()
    .transformToMulti(array -> Multi.createFrom()
            .iterable(deflateMessage.deflateMessage(array)))
    .runSubscriptionOn(Infrastructure.getDefaultExecutor()) // items will be transformed on a separate thread
    .concatenate();

为了在单独的线程上完成对Kafka队列的完整下游、转换和写入，可以使用emitOn操作符，如下所示：

@POST
@Consumes(MediaType.APPLICATION_JSON)
@Produces(MediaType.APPLICATION_JSON)
public Multi<OutputSample> send(InputSample inputSample) {
    ObjectMapper mapper = new ObjectMapper();
    return Uni.createFrom()
            .item(inputSample)
            .onItem()
            .transformToMulti(array -> Multi.createFrom().iterable(deflateMessage.deflateMessage(array)))
            .emitOn(Executors.newFixedThreadPool(5)) // items will be emitted on a separate thread after transformation
            .onItem()
            .transformToUniAndConcatenate(payload -> {
                try {
                    return producer.writeToKafka(payload, mapper);
                } catch (JsonProcessingException e) {
                    e.printStackTrace();
                }
                return Uni.createFrom().<OutputSample>nothing();
            });
}