【kafka实践】11｜消费位移提交

本文介绍: 在实践生产使用中关于消费位移提交的选择，完全取决于业务场景，本季内容将介绍自动提交，手动提交以及同步、异步提交

消费者位移这一节介绍了消费者位移的基本概念和消息格式，本节我们来聊聊消费位移的提交。

Con sum e r 需要向 Kaf k a 汇报自己的位移数据，这个汇报过程被称为提交位移（Co m mitt in g Off sets）。因为 Con sum e r 能够同时消费多个分区的数据，所以位移的提交实际上是在分区粒度上进行的，即 Con sum e r 需要为分配给它的每个分区提交各自的位移数据。

提交位移主要是为了记录Con sum e r 的消费进度，这样当 Con su m e r 发生重启之后，就能够从 Kafka 中读取之前提交的位移，从而继续消费，避免以避免重复消费，或消息丢失等。换句话说，位移提交是 Kafka 提供给你的一个工具或语义保障，你负责维持这个语义保障，即如果你提交了位移 X，那么 Kafka 会认为所有位移值小于 X 的消息你都已经成功消费了。

因为位移提交非常灵活，你完全可以提交任何位移值。假设你的 Con sum e r 消费了 10 条消息，你提交的位移值却是 20，那么从理论上讲就丢失了10条数据；相反地，如果你提交的位移值是 5，那么就重复消费5条数据。所以你对位移提交的管理直接影响了你的 Consum e r 所能提供的消息语义保障。

从使用角度来说位移提交分为自动提交和手动提交；从 Consum er 的角度来说，位移提交分为同步提交和异步提交。

默认情况下就是自动提交，你根本无需关心位移提交的事情，Consum er 端有个参数 enable.auto.commit 默认值是 true，即 Consumer 默认自动提交位移的。还有个参数 auto.commit.interval.ms，默认值是 5 秒，即每 5 秒会为你自动提交一次位移。

Properties props = new Properties();
     props.put("bootstrap.servers", "localhost:9092");
     props.put("group.id", "kafka_test");
     // 自动提交
     props.put("enable.auto.commit", "true");
     // 间隔2秒  
     props.put("auto.commit.interval.ms", "2000");
     props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     KafkaConsumer<String, String&gt; consumer = new KafkaConsumer<&gt;(props);
     consumer.subscribe(Arrays.asList("topic"));
     while (true) {
         ConsumerRecords<String, String> records = consumer.poll(100);
         for (ConsumerRecord<String, String> record : records) {
            // process
         }
             
     }

// props.put("enable.auto.commit", "false");
while (true) {
            ConsumerRecords<String, String> records =
                        consumer.poll(Duration.ofSeconds(1));
            // 处理消息
            process(records); 
            try {
                        // 同步提交
                        consumer.commitSync();
            } catch (CommitFailedException e) {
                        handle(e); // 处理提交失败异常
            }
}

   try {
           while(true) {
                        ConsumerRecords<String, String> records = 
                                    consumer.poll(Duration.ofSeconds(1));
                        process(records); 
                        // 异步提交规避阻塞
                        commitAysnc(); 
            }
} catch(Exception e) {
            
} finally {
            try {
                        // 使用同步阻塞式提交兜底
                        consumer.commitSync(); 
  } finally {
       consumer.close();
}
}