Helidon 的 Reactive messaging (kafka) 如何实现批处理？
结论
可以将多条 Kafka 消息组批后写入数据库。通常的做法是接收 Message<T> 流，在应用中按数量或时间组批，执行数据库批量写入，等事务提交成功后再确认这批消息。
仅把消费方法的参数改成 List<Message<?>>，不会让 Kafka connector 自动组批。Helidon 4 支持注入 Publisher<Message<T>> 等消息流，可以通过这个入口在应用层实现批处理。Helidon Reactive Messaging 文档
问题未指定 Helidon 版本，这里以 Helidon MP 4、Reactor 和 JDBC 为例。使用其他版本或 Helidon SE 时，需要调整通道接入方式。
为什么需要自己组批
Kafka 一次 poll() 可以取回多条记录，但 connector 仍可能向下游逐条发送 Message<T>。Kafka 拉取的一批记录，不一定就是数据库写入的一批数据。
应用需要分别处理以下问题：
- 什么时候结束一批，例如达到 100 条，或等待 200 毫秒。
- 如何执行数据库批量写入。
- 写入成功或失败后，如何确认消息。
max.poll.records 控制一次 poll() 返回的记录数，不会让消费方法改为接收列表。Helidon 文档中的 batch-size 是 producer 的发送批量配置，也不能用来启用消费者批量回调。Kafka consumer 配置、Helidon Kafka connector 配置
实现步骤
1. 关闭 Kafka 自动提交
如果要求数据库写入成功后才能推进消费进度，应关闭自动提交：
mp.messaging.incoming.from-kafka.connector=helidon-kafka
mp.messaging.incoming.from-kafka.topic=events
mp.messaging.incoming.from-kafka.bootstrap.servers=localhost:9092
mp.messaging.incoming.from-kafka.group.id=db-batch-writer
mp.messaging.incoming.from-kafka.enable.auto.commit=false
mp.messaging.incoming.from-kafka.key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
mp.messaging.incoming.from-kafka.value.deserializer=org.apache.kafka.common.serialization.StringDeserializer其中的 topic、地址和 group ID 都是示例值。
根据 Helidon 的 KafkaPublisher.Builder 文档，关闭自动提交后，调用 Message.ack() 会触发提交。如果直接使用 builder 创建 publisher，还要让 autoCommit(false) 与 Kafka 的 enable.auto.commit=false 保持一致。KafkaPublisher.Builder API
2. 注入完整消息流
在 CDI bean 中接收带有确认回调的消息：
import jakarta.inject.Inject;
import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Message;
import org.reactivestreams.Publisher;
@Inject
@Channel("from-kafka")
Publisher<Message<String>> incoming;这里的 String 与配置中的 StringDeserializer 对应。实际项目可以换成业务对象，但反序列化结果必须与消费端类型一致。
同一个通道只接入这条消费流水线，不要再添加另一个 @Incoming("from-kafka") 消费者。
3. 组批、写入并确认
项目需要引入兼容的 io.projectreactor:reactor-core，且所用版本须提供 bufferTimeout(int, Duration, boolean) 重载。这里由 Reactor 完成组批，Helidon Kafka connector 没有对应的组批配置项。
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.time.Duration;
import java.util.List;
import java.util.function.Consumer;
import javax.sql.DataSource;
import org.eclipse.microprofile.reactive.messaging.Message;
import org.reactivestreams.Publisher;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
public final class BatchWriter {
private final DataSource dataSource;
public BatchWriter(DataSource dataSource) {
this.dataSource = dataSource;
}
public Disposable start(
Publisher<Message<String>> incoming,
Consumer<Throwable> onFailure) {
return Flux.from(incoming)
.bufferTimeout(100, Duration.ofMillis(200), true)
.filter(batch -> !batch.isEmpty())
.concatMap(this::writeAndAck, 1)
.subscribe(ignored -> { }, onFailure);
}
private Mono<Void> writeAndAck(List<Message<String>> batch) {
return Mono.fromCallable(() -> {
insertBatch(batch);
return Boolean.TRUE;
})
.subscribeOn(Schedulers.boundedElastic())
.thenMany(
Flux.fromIterable(batch)
.concatMap(
message -> Mono.fromCompletionStage(
message.ack()),
1))
.then();
}
private void insertBatch(List<Message<String>> batch)
throws SQLException {
try (Connection connection = dataSource.getConnection()) {
connection.setAutoCommit(false);
try {
try (PreparedStatement statement =
connection.prepareStatement(
"INSERT INTO events (payload) VALUES (?)")) {
for (Message<String> message : batch) {
statement.setString(1, message.getPayload());
statement.addBatch();
}
statement.executeBatch();
}
connection.commit();
} catch (SQLException | RuntimeException failure) {
try {
connection.rollback();
} catch (SQLException rollbackFailure) {
failure.addSuppressed(rollbackFailure);
}
throw failure;
}
}
}
}这个示例要求数据库支持事务和 JDBC batch，已存在 events(payload) 表，且 DataSource 提供的连接允许代码自行管理事务。如果连接由 JTA 管理，应使用相应的事务边界，不能直接照搬 commit() 和 rollback()。
应用启动时调用一次 start(incoming, onFailure)，并保存返回的 Disposable。onFailure 应接入日志、告警和消费恢复机制。生产环境关闭应用时，需要先停止接收新消息，等正在执行的批次结束后再释放资源。
代码按以下方式处理消息：
- 每批最多收集 100 条，超时后即使不足 100 条也会输出。
- 使用
concatMap串行处理批次，等上一批写入和确认结束后再处理下一批。 - 在
boundedElastic上执行阻塞式 JDBC 调用。 executeBatch()执行成功且数据库事务提交成功后，逐条调用ack()，并等待确认完成。
100 条和 200 毫秒只是示例值，应根据消息大小、数据库延迟和吞吐需求调整。bufferTimeout 的行为及背压重载见 Reactor Flux API。
数据库批处理还需要注意什么
executeBatch() 不等于一条 SQL
示例通过 addBatch() 收集参数，再调用一次 executeBatch()，使用的是 JDBC batch。驱动是否将其改写为多值 INSERT、会发出多少次网络请求，取决于数据库、驱动和配置。
如果要求只执行一条 SQL，需要使用数据库支持的多行插入语法，并检查参数数量等限制。JDBC PreparedStatement API
使用 Helidon DbClient 时，也应核对当前版本和数据库提供方的批量执行能力。在一个事务中循环调用单条 insert，只能说明这些操作共享事务，不能说明使用了数据库批量接口。
数据库提交和 Kafka 确认之间可能发生故障
这段代码中的数据库事务与 Kafka offset 提交仍是两个独立的操作，无法保证原子性。例如，数据库已经提交，但进程在 ack() 完成前退出，恢复消费时就可能再次收到这些消息。
生产环境应实现幂等写入，例如使用业务事件 ID，或将 Kafka 的 (topic, partition, offset) 保存为唯一键，并明确重复记录的处理逻辑。示例中的普通 INSERT 尚未实现幂等写入。
失败后不能直接跳过这一批
数据库操作失败时，示例会回滚事务并终止订阅，不会对失败批次执行成功确认。代码没有实现自动重试，需要另行安排消费恢复。
如果选择调用 nack()，还要核对 connector 的失败策略。Helidon Kafka connector 提供终止通道、DLQ 和仅记录日志等策略，调用 nack() 不一定会让消息稍后自动重试。其中，仅记录日志的策略会丢弃消息并提交 offset。Helidon Kafka connector 的 NACK 策略
批大小之外，还要限制在途消息
组批和流水线预取都会增加尚未确认的消息数量，因此还要考虑数据库连接池、内存上限以及 connector 的未提交消息限制。只按数量组批会让低流量下的消息长时间等待，持续消费通常还需要设置时间触发条件。
不要对长期运行的 Kafka 流直接调用 collectList() 来定期生成批次，因为这类收集操作需要等待流结束。
备注：内容仅供参考。