// 通过Kafka的方式消费消息，如果将位移提交更改为手动位移提交。那么我们就需要主动调用consumer的方法完成位移提交
private static void consumerFromKafka() {

	// 创建属性集合，设置初始化参数
	Properties properties = new Properties();
	properties.put("key.deserializer",  "org.apache.kafka.common.serialization.StringDeserializer");
	properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
	properties.put("bootstrap.servers", "192.168.23.131:9092");
	properties.put("group.id", "itheima.demo");
	properties.put("enable.auto.commit" , "false") ;

	// 创建KafkaConsumer对象
	KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
	consumer.subscribe(Collections.singletonList("itheima"));

	while (true) {
		ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
		for (ConsumerRecord<String, String> record : records) {
			System.out.println(record.value());
		}
		// consumer.commitSync();          // 进行同步位移提交,会阻塞当前线程
		consumer.commitAsync(new OffsetCommitCallback(){

			@Override
			public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
				if (exception == null) {
					System.out.println(offsets);
				} else {
					LOGGER.error("fail to commit offsets {}", offsets, exception);
				}
			}

		});
	}

}
