Kafka消费者接收相同消息

Question

我是Kafka的新手，我有一个使用Java Apache Camel库实现的Kafka消费者。我发现的问题是-消费者花了很长的时间(>15分钟）来处理很少的消息-这对于我们的用例来说是很好的。

需要一些配置帮助，因为相同的消息会在15分钟后重新发送，如果在15分钟内没有处理（我相信线程控制不会返回）。我想这可能是默认间隔，不确定这是哪一个属性。

那么，我必须在哪里修复配置

• 生产者级别，以便它不重新发送
• 或者生产者不参与重新发送，而是代理-Kafka服务器，因此消费者必须在处理之前确认消息。

我的制作人具有以下属性：

<entry key="bootstrap.servers" value="${kafka.bootstrap.servers}" />
       <entry key="key.serializer" value="org.apache.kafka.common.serialization.StringSerializer" />
       <entry key="value.serializer" value="org.apache.kafka.common.serialization.StringSerializer" />

我的消费者配置看起来像

<endpoint id="apolloKafkaJanitorEventListenerURI"
            uri="kafka:${kafka.bootstrap.servers}?topic=${apollo.janitor.event.topic}&amp;
                                                        groupId=${apollo.janitor.event.group.id}&amp;
                                                        consumersCount=${apollo.janitor.event.consumer.count}&amp;
                                                        consumerRequestTimeoutMs=${eventConsumerRequestTimeoutMs}&amp;
                                                        sessionTimeoutMs=${eventConsumerSessionTimeoutMs}&amp;
                                                        maxPartitionFetchBytes=${eventConsumerMaxPartitionFetchBytes}" />

我已经谷歌没有找到任何相关的问题。在生产者和使用者上找到“acks=0”属性如下。我还没有测试，但我想先看看我是否在正确的轨道上

KafkaManualCommit manual =
       exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
   manual.commitSync();

https://github.com/apache/camel/blob/master/components/camel-kafka/src/main/docs/kafka-component.adoc

Answer 1

问题可能出在生产者端。您可能不需要检查生产者是否正在重新发送消息。您可以使用日志记录语句。也可以对kafka Producer使用一次语义。您只需要为相同的添加一个额外的属性。

另一种可能是您的消费者没有提交抵消。在这一端你可能也需要做一些头脑Storm