叛变-卡夫卡写顺序发生

Question

我是夸克的新手。我正在尝试使用quarkus reactive编写一个REST端点，该端点接收输入，进行一些验证，将输入转换为列表，然后向kafka写入消息。我的理解是将所有内容转换为Uni/Multi，将导致以异步方式在I/O线程上执行。在intelllij日志中，我可以看到代码在执行器线程中以顺序的方式执行。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

兵变和其他活性图书馆设计主要围绕数据流控制。

也就是说，在其核心，它将提供一组功能(通常通过一些操作符)来控制流的执行和调度。这意味着，除非您指示community对象异步执行，否则它们将简单地以顺序(旧)方式执行。

执行调度使用两个操作符控制:

• runSubscriptionOn:这将导致生成项目(通常指上游)的代码片段在指定的Executor
的线程上执行。
• emitOn:这将导致订阅代码(通常指下游)在指定的Executor
的线程上执行。

你可以这样更新你的代码，使紧缩变成异步的:

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();
});
}

Answer 2

Multi用于当您的源连续发出项，直到它发出一个完成事件，这不是您的情况。

From Mutiny docs:

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

您将很少创建Multi yourself的实例而是用响应式客户端，公开一个Mutiny API。

你正在寻找的是一个Uni<List<OutputSample>>，因为你的API你返回1，只有1项与完整的结果列表。

所以你需要的是将每条消息发送给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;
}