Quarkus Reactive基于Mutiny开发时Kafka顺序写入导致延迟升高如何解决

现有代码核心问题 #

• 每次请求新建ObjectMapper实例：ObjectMapper是线程安全的，重复创建会带来不必要的内存和性能开销，Quarkus已经内置了预配置的全局实例可直接注入使用。
• 流处理未指定并发度和调度器：上游Multi使用concatenate按顺序发射元素，transformToUniAndMerge默认并发度为1，导致payload只能逐个处理，天然就是顺序执行。
• 异常处理逻辑错误：在transformToUniAndMerge的回调中手动捕获异常后返回null，会触发响应流的空指针异常，破坏流处理正常逻辑。
• 序列化操作放在回调层直接执行：writeValueAsString操作直接在流的转换回调中执行，异常无法被响应流的异常处理链自动捕获，需要手动处理增加冗余代码。
• Kafka生产者默认配置限制并发：默认配置下max.in.flight.requests.per.connection为1，同一时间只能有一个请求在发送，自然会顺序执行写入操作，拉高延迟。

具体优化方案 #

1. 复用全局ObjectMapper #

删除接口中新建ObjectMapper的逻辑，直接注入Quarkus提供的全局实例：

@Inject
ObjectMapper mapper;

2. 调整流处理逻辑，提升并发度 #

去掉concatenate操作，调整transformToUniAndMerge的并发度，同时指定调度器避免阻塞事件循环线程：

@POST
@Consumes(MediaType.APPLICATION_JSON)
@Produces(MediaType.APPLICATION_JSON)
public Multi<OutputSample> send(InputSample inputSample) {
    // 直接从deflate结果创建Multi，无需先包一层item再转
    return Multi.createFrom().iterable(deflateMessage.deflateMessage(inputSample))
            // 指定订阅运行在worker线程池，避免阻塞事件循环
            .runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
            // 调整并发度为你需要的并行处理数，比如10，可同时处理10个payload的Kafka写入
            .onItem().transformToUniAndMerge(10, payload -> producer.writeToKafka(payload, mapper));
}

3. 修复writeToKafka的异常和执行逻辑 #

将序列化操作放到Uni的供应逻辑中，让异常自动被流处理链捕获，避免手动try-catch：

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

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

4. 调整Kafka生产者配置 #

在application.properties中添加以下配置，提升写入并发、降低延迟：

# 允许同一连接最多同时处理5个发送请求，解决顺序写入问题
kafka.producer.max.in.flight.requests.per.connection=5
# 开启批量发送，最多等待5ms攒批，大幅降低高频写入的延迟
kafka.producer.linger.ms=5
# 批量大小设置为16KB，可根据实际消息大小调整
kafka.producer.batch.size=16384
# 消息确认等级，不需要强一致可设为1，比all性能高很多
kafka.producer.acks=1
# 开启LZ4压缩，减少网络传输耗时
kafka.producer.compression.type=lz4

5. 可选终极优化（不需要返回每个消息写入结果时使用） #

如果业务不需要给客户端返回每个消息的写入结果，直接接收请求后返回成功，后台异步处理Kafka写入，延迟可以降到最低：

@POST
@Consumes(MediaType.APPLICATION_JSON)
@Produces(MediaType.APPLICATION_JSON)
public Uni<Response> send(InputSample inputSample) {
    // 异步触发Kafka写入逻辑，不等待结果
    Multi.createFrom().iterable(deflateMessage.deflateMessage(inputSample))
            .onItem().transformToUniAndMerge(10, payload -> producer.writeToKafka(payload, mapper))
            .subscribe().with(
                result -> {/* 可加成功日志 */},
                fail -> {/* 可加失败告警 */}
            );
    // 直接返回响应
    return Uni.createFrom().item(Response.accepted().build());
}

内容的提问来源于stack exchange，提问作者rightjoin