ApacheKafka：0.10版中只有一次

Question

为了实现Kafka消费者对消息的一次处理，我一次提交一条消息，如下所示

public void commitOneRecordConsumer(long seconds) {
        KafkaConsumer<String, String> consumer = consumerConfigFactory.getConsumerConfig();

        try {

            while (running) {
                ConsumerRecords<String, String> records = consumer.poll(1000);
                try {
                    for (ConsumerRecord<String, String> record : records) {

                        processingService.process(record);

                        consumer.commitSync(Collections.singletonMap(new TopicPartition(record.topic(),record.partition()), new OffsetAndMetadata(record.offset() + 1)));

                        System.out.println("Committed Offset" + ": " + record.offset());

                    }
                } catch (CommitFailedException e) {
                    // application specific failure handling
                }
            }
        } finally {
            consumer.close();
        }
    }

上面的代码将消息的处理异步委托给下面的另一个类。

@Service
public class ProcessingService {

    @Async
    public void process(ConsumerRecord<String, String> record) throws InterruptedException {
        Thread.sleep(5000L);
        Map<String, Object> map = new HashMap<>();
        map.put("partition", record.partition());
        map.put("offset", record.offset());
        map.put("value", record.value());
        System.out.println("Processed" + ": " + map);
    }

}

但是，这仍然不能保证只发送一次，因为如果处理失败，它可能仍会提交其他消息，并且以前的消息将永远不会被处理和提交，我在这里的选项是什么？

Answer 1

我认为使用kafka 0.10. x本身可以实现精确的一次处理。但是有一些问题。我正在分享这本书的高级思想。相关内容可以在第4章中的“寻找和精确的一次处理”部分找到：Kafka消费者-从Kafka读取数据。您可以使用（免费）safaribooksonline帐户查看那本书的内容，或者在它出来后购买，或者可能从其他来源获得它，我们不会谈论它。

想法：

想想这个常见的场景：您的应用程序从Kafka中读取事件，处理数据，然后将结果存储在数据库中。假设我们真的不想丢失任何数据，也不想将相同的结果存储在数据库中两次。

如果有办法在一个原子动作中同时存储记录和偏移量，这是可行的。要么记录和偏移都已提交，要么两者都未提交。为了实现这一点，我们需要在一个事务中将记录和偏移量写入数据库。然后我们就知道要么我们处理完了记录，偏移量被提交了，要么我们没有，记录将被重新处理。

现在唯一的问题是：如果记录存储在数据库中而不是Kafka中，那么当分配给它一个分区时，我们的消费者如何知道从哪里开始读取？这正是seek（）。当使用者启动或分配新分区时，它可以在数据库中查找偏移量并seek（）到该位置。

书中的示例代码：

public class SaveOffsetsOnRebalance implements ConsumerRebalanceListener {
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        commitDBTransaction(); 
    }

    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        for(TopicPartition partition: partitions)
        consumer.seek(partition, getOffsetFromDB(partition)); 
    }
}

consumer.subscribe(topics, new SaveOffsetOnRebalance(consumer));
consumer.poll(0);

for (TopicPartition partition: consumer.assignment())
    consumer.seek(partition, getOffsetFromDB(partition));   

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records)
    {
        processRecord(record);
        storeRecordInDB(record);
        storeOffsetInDB(record.topic(), record.partition(), record.offset()); 
    }
    commitDBTransaction();
}

Answer 2

ApacheKafka 0.11.0.0刚刚发布，现在它只支持一次交付。

http://kafka.apache.org/documentation/#upgrade_11_exactly_once_semantics

https://cwiki.apache.org/confluence/display/KAFKA/KIP-98-精确一次交付和事务消息传递

Answer 3

0.10.2及更高版本的原始答案（0.11及更高版本请参阅答案）

目前，Kafka无法提供开箱即用的精确一次处理。如果您在成功处理消息后提交消息，您可以进行至少一次处理，或者如果您在开始处理之前直接在投票（）之后提交消息，您可以进行最多一次处理。

（另请参见本节“交货保证”一段。）http://docs.confluent.io/3.0.0/clients/consumer.html#synchronous-（提交）

但是，如果您的处理是幂等的，则至少一次保证“足够好”，即即使您处理两次记录，最终结果也将是相同的。幂等处理的示例是将消息添加到键值存储中。即使您两次添加相同的记录，第二次插入也只会替换第一个当前键值对，KV存储中仍然会有正确的数据。

在上面的示例代码中，您更新了一个HashMap，这将是一个幂等运算。即使在发生故障的情况下，如果在崩溃之前只执行了两个put调用，也可能会出现不一致的状态。然而，这种不一致的状态将被修复为再次重新处理同一记录。

对println（）的调用不是幂等的，因为这是一个具有“副作用”的操作。但我猜打印仅用于调试目的。

作为替代方案，您需要在用户代码中实现事务语义，在出现故障时需要“撤消”（部分执行）操作。总的来说，这是一个难题。

Apache Kafka 0.11的更新（关于0.11之前的版本，请参见上面的答案）

从0.11开始，Apache Kafka支持幂等生产者、事务性生产者和使用Kafka Streams的精确一次处理。它还向消费者添加了"read_committed"模式，以仅读取已提交的消息（并删除/过滤中止的消息）。

• https://kafka.apache.org/documentation/#semantics
• https://www.confluent.io/blog/exactly-once-semantics-are-possible-heres-how-apache-kafka-does-it/
• https://www.confluent.io/blog/transactions-apache-kafka/
• https://www.confluent.io/blog/enabling-exactly-kafka-streams/