Quarkus Kafka消费者单实例处理4分区消息为何串行而非并行？

使用Quarkus处理一个包含4个分区的Kafka Topic消息时，发现单消费者实例下消息是逐个串行处理的。当前消息处理逻辑如下：

@Incoming("topic-1")
@Retry(delay = 3, maxRetries = 3, retryOn = { MyException.class })
@NonBlocking
public CompletionStage<Void> processEvent(KafkaRecordBatch<String, String> payload) {
    Multi.createFrom().iterable(payload.getRecords())
            .onItem().transform(EventProcessor1::newLogEvent).filter(l -> !l.isEmptyEvent())
            .onItem()
            .transform(this::processEvent).onItem().transformToUniAndConcatenate(this::process2).collect().with(Collectors.counting())
            .subscribe().with(c -> {
                log.info("completed {} events", c);
            });

   return payload.ack().whenCompleteAsync((s, e) -> {
        if (e != null) {
            throw new RuntimeException(e);
        }
        log.debug("batch processed & comitted successfully");
    });
} 

疑问：单消费者实例针对4分区Topic，是否应该并行处理消息？是否遗漏了配置或逻辑？

单消费者实例下，4个分区的消息应该可以实现多分区并行处理（每个分区的消息串行，但不同分区的消息并行），你的代码和配置存在以下问题导致串行：

1. 内部消息处理被强制串行 #

代码中使用的transformToUniAndConcatenate方法特性是：必须等待前一个Uni执行完成，才会启动下一个Uni，直接将批量内的消息处理变成了串行。要实现批量内消息并行处理，需替换为transformToUniAndMerge，它会同时启动所有Uni并合并结果：

// 替换原串行处理的代码行
.transform(this::processEvent).onItem().transformToUniAndMerge(this::process2)

2. 异步处理与ACK生命周期未绑定 #

当前代码中，内部Multi处理是独立调用.subscribe()，而返回的payload.ack()会立即执行，这意味着消息还未处理完成就提交了偏移量，一旦处理失败会直接丢失消息。必须将Multi的处理结果与payload.ack()绑定，确保处理完成后再执行ACK：
修正后的完整代码示例：

@Incoming("topic-1")
@Retry(delay = 3, maxRetries = 3, retryOn = { MyException.class })
@NonBlocking
public CompletionStage<Void> processEvent(KafkaRecordBatch<String, String> payload) {
    return Multi.createFrom().iterable(payload.getRecords())
            .onItem().transform(EventProcessor1::newLogEvent)
            .filter(l -> !l.isEmptyEvent())
            .onItem().transform(this::processEvent)
            .onItem().transformToUniAndMerge(this::process2)
            .collect().with(Collectors.counting())
            .subscribeAsCompletionStage() // 将Multi转换为CompletionStage，绑定生命周期
            .thenAccept(c -> log.info("completed {} events", c))
            .thenCompose(v -> payload.ack()) // 处理完成后再执行ACK
            .whenComplete((s, e) -> {
                if (e != null) {
                    log.error("Batch processing failed", e);
                    throw new RuntimeException(e);
                }
                log.debug("batch processed & committed successfully");
            });
}

3. 消费者并发配置缺失 #

需在application.properties中添加以下配置，开启多分区并行处理：

# 设置消费者并发数，建议等于分区数（4）
quarkus.kafka.consumer.max-concurrency=4
# 每次拉取的最大记录数，根据业务批量需求调整
quarkus.kafka.consumer.max-poll-records=500
# 启用批量消费（已使用KafkaRecordBatch，需确保此配置开启）
quarkus.kafka.consumer.batch=true

其中max-concurrency指定了消费者可并行处理的分区数量，设置为4后，4个分区的消息会被并行交付到@Incoming方法，实现真正的多分区并行处理。

补充说明 #

• 单消费者实例下，Kafka保证单个分区内的消息串行处理（维护分区消息顺序），但不同分区的消息可以并行处理；
• 如果业务允许忽略分区内消息顺序，可通过拆分更多分区提升并行度，但不建议在同一分区内强行并行处理，会破坏Kafka的顺序语义。

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