如何在消息被消费或处理后从Kafka那里得到确认。可能听起来很愚蠢，但是有没有办法知道接收到ack的消息的开始和结束偏移量？
wlp8pajw1#
Kafka并没有真正做到这一点。要了解原因，请查看此处的设计文档。为了提供准确的一次确认，您需要为应用程序创建一些外部跟踪系统，在其中显式地编写确认并对事务id实现锁，以确保只处理一次。实现这样的系统的计算成本非常高，这也是大型事务系统需要比较奇特的硬件并且具有比kafka这样的系统更低的可伸缩性的主要原因之一。如果您不需要强持久性语义，那么可以使用groups api粗略地跟踪最后一条消息的读取时间。这样可以确保每条消息至少读取一次。请注意，由于groups api不提供显式跟踪应用程序自己的处理逻辑的能力，因此在这种情况下，实际的处理保证相当薄弱。依赖幂等处理的方案在这种环境中很常见。或者，您可以使用名称很差的simpleconsumerapi（使用起来相当复杂），它使您能够显式地跟踪应用程序中的时间戳。这是通过本机kafkaapi可以实现的最高级别的处理保证，因为它使您能够跟踪应用程序自己对从队列中读取的数据的处理。
6qfn3psc2#
到目前为止，我发现在0.8中，他们引入了以下方法来从偏移量中选择读数。。kafka.api.offsetrequest.earliesttime（）在日志中查找数据的开头并从那里开始流式处理，kafka.api.offsetrequest.latesttime（）将只流式处理新消息。示例代码https://cwiki.apache.org/confluence/display/kafka/0.8.0+simpleconsumer+example还是不确定确认部分
2条答案
按热度按时间wlp8pajw1#
Kafka并没有真正做到这一点。要了解原因，请查看此处的设计文档。
为了提供准确的一次确认，您需要为应用程序创建一些外部跟踪系统，在其中显式地编写确认并对事务id实现锁，以确保只处理一次。实现这样的系统的计算成本非常高，这也是大型事务系统需要比较奇特的硬件并且具有比kafka这样的系统更低的可伸缩性的主要原因之一。
如果您不需要强持久性语义，那么可以使用groups api粗略地跟踪最后一条消息的读取时间。这样可以确保每条消息至少读取一次。请注意，由于groups api不提供显式跟踪应用程序自己的处理逻辑的能力，因此在这种情况下，实际的处理保证相当薄弱。依赖幂等处理的方案在这种环境中很常见。
或者，您可以使用名称很差的simpleconsumerapi（使用起来相当复杂），它使您能够显式地跟踪应用程序中的时间戳。这是通过本机kafkaapi可以实现的最高级别的处理保证，因为它使您能够跟踪应用程序自己对从队列中读取的数据的处理。
6qfn3psc2#
到目前为止，我发现在0.8中，他们引入了以下方法来从偏移量中选择读数。。
kafka.api.offsetrequest.earliesttime（）在日志中查找数据的开头并从那里开始流式处理，kafka.api.offsetrequest.latesttime（）将只流式处理新消息。
示例代码https://cwiki.apache.org/confluence/display/kafka/0.8.0+simpleconsumer+example
还是不确定确认部分