主页
产品
Red Hat Integration
2023.Q1
Camel Spring Boot 参考
29.10. 在 Kafka 使用者中使用手动提交

29.10. 在 Kafka 使用者中使用手动提交

默认情况下，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.commit();
}

public void process(Exchange exchange) {
    KafkaManualCommit manual =
        exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
    manual.commit();
}

Copy to Clipboard

Toggle word wrap

这将强制同步提交，直到在 Kafka 上确认提交，或者是否抛出异常。您还可以使用异步提交，使用 'DefaultKafkaManualAsyncCommitFactory'implementation 配置 KafkaManualCommitFactory 'implementation。

然后，提交将使用 kafka 异步 commit api 在下一个消费者循环中进行。请注意，分区中的记录必须由唯一的线程处理和提交。如果没有，这可能会导致不一致的行为。这主要用于聚合的完成超时策略。

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

返回顶部

29.10. 在 Kafka 使用者中使用手动提交

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links