package com.muyu.carData.consumer;

import com.alibaba.fastjson.JSONObject;
import com.muyu.carData.event.EventPublisher;
import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**卡夫卡消费者
 * @Author：张腾
 * @Package：com.muyu.carData.consumer
 * @Project：cloud-server-8
 * @name：MyKafkaConsumer
 * @Date：2024/9/26 15:42
 */
@Component
@Log4j2
public class MyKafkaConsumer implements ApplicationRunner, ApplicationListener<ContextClosedEvent> {

    @Autowired
    private KafkaConsumer kafkaConsumer;

    @Autowired
    private EventPublisher eventPublisher;

    private final String topicName = "carJsons";

    private final ExecutorService executorService = Executors.newFixedThreadPool(10);

    @Override
    public void run(ApplicationArguments args) throws Exception {
        log.info("开始监听kafka-topic:{}",topicName);
        List<String> topics = Collections.singletonList(topicName);
        kafkaConsumer.subscribe(topics);

        while (true){
            ConsumerRecords<String,String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(100));
            consumerRecords.forEach(record ->{
                 executorService.submit(() -> handleRecord(record));
                 log.info("数据为:{},消费成功!",record);
            });
        }
    }

    private void handleRecord(ConsumerRecord<String,String> record) {
        String value = record.value();
        JSONObject jsonObject = JSONObject.parseObject(value);
        eventPublisher.publishEvent(jsonObject);
    }

    @Override
    public void onApplicationEvent(ContextClosedEvent event) {
         log.info("关闭kafka和线程");
         kafkaConsumer.close();
         executorService.shutdown();
    }
}
