package com.itheima.kafka.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.Properties;

/**
 * 消费者启动类
 */
@SpringBootApplication
public class KafkaConsumerApplication {

    // 日志记录器对象
    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaConsumerApplication.class) ;

    public static void main(String[] args) {
        SpringApplication.run(KafkaConsumerApplication.class , args) ;
        // consumerFromKafka();
    }

    // 通过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);
                        }
                    }

                });

            }

        }

    }

}
