public class KafkaConsumerAnalysis {
    
    private static final String brokerList = "";
    private static final String topic = "topic-demo";
    private static final String groupId = "group.demo";
    private static final AtomicBoolean isRunningg = new AtomicBoolean(true);

    public static Properties initConfig() {
        Properties props = new Properties();
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        props.put("bootstrap.servers", brokerList);
        props.put("group.id", groupId);
        props.put("client.id", "consumer.client.id.demo");
        return props;
    }

    public static void main(String[] args) {
        Properties props = initConfig();
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        /**
         * 订阅主题
         * 1。如果以集合形式订阅则每次调用会覆盖以前订阅的集合
         */
        consumer.subscribe(Collections.singleton(topic));
        /**
         * 2。如果以正则表达式订阅，则后续上线的新的主题也会被订阅，使用较多
         */
//        consumer.subscribe(compile("topic-.*"));

        /**
         * 订阅分区，不允许增量，会覆盖以前的
         */
//        consumer.assign(Collections.singletonList(new TopicPartition("topic-demo", 0)));
        //查询指定主题的元数据信息
        List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic);
        ArrayList<TopicPartition> topicPartitions = new ArrayList<>();

        //只获取指定分区集的消费信息
        partitionInfos.forEach(v -> topicPartitions.add(new TopicPartition(v.topic(), v.partition())));
//        consumer.assign(topicPartitions);
        /**
         * 取消订阅，可以订阅后取消后再订阅
         * 1。调用unsubscribe方法
         * 2。subscribe一个空的集合string
         * 3。assign一个空的集合TopicPartition
         */
//        consumer.unsubscribe();


        try {
            while (isRunningg.get()) {
                /**
                 * 设置拉取超时时间，如果为0则会立即返回，不管是否拉取到消息。
                 * 如果是单独开的线程，则可以将其设置为超长时间最大值
                 *
                 * poll涉及消费位移、消费者协调器、组协调器、消费者选举、分区分配的分发、再均衡、心跳
                 *
                 */
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));


                records.forEach(v -> {
                    System.out.println("topic = " + v.topic() + ", partition = " + v.partition() + ", offset= " + v.offset());
                    System.out.println("key = " + v.key() + ", value = " + v.value());

                    log.info("headers:{}",v.headers());
                    //如果序列化之后的值大小为为空则返回-1，否则返回序列化的大小
                    log.info("serializedKeySize:{}",v.serializedKeySize());
                    log.info("serializedValueSize:{}",v.serializedValueSize());
                    log.info("timestamp:{}",v.timestamp());
                    log.info("timestampType:{}",v.timestampType());
                    //TODO 处理消息
                });


                /**
                 * 按照分区纬度来处理消息
                 * 1.获取返回记录包含的分区set集合，如果未返回数据，则可能为空
                 * 2。获取指定分区的消息
                 */
                records.partitions().forEach(v->{
                    records.records(v).forEach(v1->{
                        //TODO 每个不同的分区处理
                    });
                });

                /**
                 * 按主题纬度来进行消费
                 * 根据订阅时的列表来处理
                 *
                 */
                ArrayList<String> topics = new ArrayList<>();
                partitionInfos.forEach(v->topics.add(v.topic()));
                topics.forEach(v->{
                    records.records(v).forEach(v1->{
                        //TODO 每个不同的主题topic处理
                    });
                });

            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            consumer.close();
        }


    }
}