kafka-0.9消費者新API

## kafka-0.9消費者新API

注：如下僅限kafka版本0.9以上Consumer新版api

Consumer自動提交示例：

Properties props = new Properties();
props.put("bootstrap.servers", kafkaBrokerURL);
props.put("group.id", groupId);
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("auto.offset.reset", "earliest");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList(topic));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        LOGGER.info(record.key() + "," + record.value());
    }
}

在0.9版本消費者新API中，有3種ack策略

1.自動提交

就是上面示例中的，程序只要獲取到消息就會提交offset，無論你的方法有沒有執行完。同時消費速度是最快的

props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");

2.手動提交

手動提交就是在你的方法執行完後手動調用consumer.commitSync();手動同步ack。每處理完1條消息，commitSync 1次
速度慢，測試手動commit耗時在0.1秒左右

props.put("enable.auto.commit", "false");

// 執行完方法後調用
consumer.commitSync();

3. 手動異步提交

手動異步提交跟手動提交方式同樣，只是加了異步，速度會有所提高

props.put("enable.auto.commit", "false");

// 執行完方法後調用
consumer.commitAsync();