191.8. 使用带有 Kafka 消费者的手动提交
从 Camel 2.21 开始提供
默认情况下,Kafka 使用者将使用自动提交,其中偏移将使用给定间隔在后台自动提交。
如果要强制手动提交,您可以使用 Camel Exchange 中的 KafkaManualCommit
API,存储在消息标头中。这需要将在 KafkaComponent
或端点上的选项 allowManualCommit
设置为 true
来打开手动提交,例如:
KafkaComponent kafka = new KafkaComponent(); kafka.setAllowManualCommit(true); ... camelContext.addComponent("kafka", kafka);
KafkaComponent kafka = new KafkaComponent();
kafka.setAllowManualCommit(true);
...
camelContext.addComponent("kafka", kafka);
然后,您可以使用 Java 代码中的 KafkaManualCommit
,如 Camel Processor
:
public void process(Exchange exchange) { KafkaManualCommit manual = exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class); manual.commitSync(); }
public void process(Exchange exchange) {
KafkaManualCommit manual =
exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
manual.commitSync();
}
这将强制进行同步提交,该提交将阻止在 Kafka 上确认提交,或者在抛出异常时失败。
如果要使用 KafkaManualCommit
的自定义实现,您可以在 KafkaComponent
上创建自定义实现实例上配置自定义 KafkaManualCommitFactory
。