主页
产品
Red Hat Fuse
7.7
Apache Camel 组件参考
189.7. 使用 Kafka consumer 的手动提交

189.7. 使用 Kafka consumer 的手动提交

可作为 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);

Copy to Clipboard

Toggle word wrap

然后，您可以使用来自 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();
}

Copy to Clipboard

Toggle word wrap

这将强制一个同步提交，该提交将阻止，直到在 Kafka 上确认提交被确认，或者是否引发异常。

如果要使用 KafkaManualCommit 的自定义实现，那么您可以在创建自定义实施的 KafkaComponent 上配置自定义 KafkaManualCommitFactory。

返回顶部

189.7. 使用 Kafka consumer 的手动提交

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links