package com.citygis.web.service.impl;

import com.citygis.web.mapper.KafkaMonitorMapper;
import com.citygis.web.service.KafkaConsumerServer;
import com.citygis.web.service.KafkaMonitorService;
import com.sun.management.OperatingSystemMXBean;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import javax.management.MBeanServerConnection;
import javax.management.ObjectName;
import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;
import java.io.File;
import java.lang.management.ManagementFactory;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @Package com.shcd.web.impl.kafka
 * @Author zyl
 * @Date 2024/5/29 0029 15:28
 * @description: kafka消费信息
 */
@Transactional(rollbackFor = Exception.class)
@Service
public class KafkaMonitorServiceImpl implements KafkaMonitorService {

    @Value("${kafka.ip}")
    private String ip;


    @Resource
    KafkaMonitorMapper kafkaMonitorMapper;

    @Resource
    KafkaConsumerServer kafkaConsumerServer;

    @Resource
    private AdminClient adminClient;

    private final static Logger log = LoggerFactory.getLogger(KafkaMonitorServiceImpl.class);

    @Value("${jmx.jmxServiceUrl}")
    private String jmxServiceUrl;

    private JMXConnector jmxConnector() throws Exception {
        JMXServiceURL serviceURL = new JMXServiceURL(jmxServiceUrl);
        return JMXConnectorFactory.connect(serviceURL);
    }

    private String getDate() {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        return simpleDateFormat.format(new Date());
    }

    public KafkaConsumer<String, String> getConsumer(Collection<String> topics, String group, String offset) {
        Properties props = this.getCommonProperties();
        props.setProperty("group.id", group);
        props.setProperty("auto.offset.reset", offset);
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(topics);
        return consumer;
    }

    public Properties getCommonProperties() {
        Properties prop = new Properties();
        prop.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, ip);
        prop.setProperty(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "2000");
        prop.setProperty(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "2000");
        prop.setProperty("enable.auto.commit", "true");
        prop.setProperty("auto.commit.interval.ms", "1000");
        prop.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        prop.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        return prop;
    }

    // @Scheduled(cron = "0 */5 * * * ?")
    @Override
    public String getPartitionSizeValue() {
        try {
            JMXConnector jmxConnector = this.jmxConnector();
            MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();

            Map<String, Integer> topicAndPartition = kafkaConsumerServer.getTopicAndPartition();

            for (String key : topicAndPartition.keySet()) {
                int totalUnderReplicated = 0;
                for (int i = 0; i < topicAndPartition.get(key); i++) {
                    ObjectName partitionSize = new ObjectName("kafka.log:type=Log,name=Size,topic=" + key + ",partition=" + i);
                    Object partitionSizeValue = mBeanServerConnection.getAttribute(partitionSize, "Value");
                    // 查询分区大小
                    log.info("分区大小：" + partitionSizeValue);

                    Integer value = Integer.parseInt(partitionSizeValue.toString());
                    totalUnderReplicated += value;
                }
                String uuId = UUID.randomUUID().toString();

                String date = this.getDate();

                kafkaMonitorMapper.addTopicSize(uuId, date, totalUnderReplicated, key);
            }

            jmxConnector.close();
            return null;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    // @Scheduled(cron = "0 */5 * * * ?")
    @Override
    public String getActiveControllerCount() {
        try {
            JMXConnector jmxConnector = this.jmxConnector();
            MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();
            ObjectName activeControllerCount = new ObjectName("kafka.controller:type=KafkaController,name=ActiveControllerCount");
            Object activeControllerCountValue = mBeanServerConnection.getAttribute(activeControllerCount, "Value");
            // 查询分区大小
            log.info("集群连接数量：" + activeControllerCountValue);
            String uuId = UUID.randomUUID().toString();

            String date = this.getDate();

            kafkaMonitorMapper.addConnectionNum(uuId, date, Integer.parseInt(activeControllerCountValue.toString()));
            jmxConnector.close();
            return activeControllerCountValue.toString();
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    @Override
    public String getConnectionCloseRate(String clientId) {
        try {
            JMXConnector jmxConnector = this.jmxConnector();
            MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();
            ObjectName connectionCloseRate = new ObjectName("kafka.consumer:type=consumer-metrics,client-id=" + clientId);
            Object connectionCloseRateValue = mBeanServerConnection.getAttribute(connectionCloseRate, "Value");
            // 查询分区大小
            log.info("连接关闭数量：" + connectionCloseRateValue);

            jmxConnector.close();
            return connectionCloseRateValue.toString();
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }


    // @Scheduled(cron = "0 */5 * * * ?")
    @Override
    public String getBytesOutPerSe()  {
        JMXConnector jmxConnector = null;
        try {
            jmxConnector = this.jmxConnector();

            MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();

            Map<String, Integer> topicAndPartition = kafkaConsumerServer.getTopicAndPartition();

            for (String key : topicAndPartition.keySet()) {
                try {
                    ObjectName bytesOutPerSe = new ObjectName("kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec,topic=" + key);
                    Object bytesOutPerSeValue = mBeanServerConnection.getAttribute(bytesOutPerSe, "OneMinuteRate");
                    // 查询分区大小
                    log.info("broker一分钟内平均每秒出流量：" + bytesOutPerSeValue);

                    String uuId = UUID.randomUUID().toString();

                    String date = this.getDate();

                    kafkaMonitorMapper.addPersec(uuId, date, (double) bytesOutPerSeValue, "流出", key);
                } catch (Exception e) {
                    e.printStackTrace();
                    // 可以添加异常处理逻辑，比如记录日志或进行其他操作
                }

            }

            jmxConnector.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    // @Scheduled(cron = "0 */5 * * * ?")
    @Override
    public String getBytesInPerSec(){
        try {
            JMXConnector jmxConnector = this.jmxConnector();
            MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();
            Map<String, Integer> topicAndPartition = kafkaConsumerServer.getTopicAndPartition();

            for (String key : topicAndPartition.keySet()) {
                try {
                    ObjectName bytesInPerSec = new ObjectName("kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec,topic=" + key);
                    Object bytesInPerSecValue = mBeanServerConnection.getAttribute(bytesInPerSec, "OneMinuteRate");
                    // 查询分区大小
                    log.info("broker一分钟内平均每入流量：" + bytesInPerSecValue);

                    String uuId = UUID.randomUUID().toString();

                    String date = this.getDate();

                    kafkaMonitorMapper.addPersec(uuId, date, (double) bytesInPerSecValue, "流入", key);
                } catch (Exception e) {
                    e.printStackTrace();
                    // 可以添加异常处理逻辑，比如记录日志或进行其他操作
                }
            }
            jmxConnector.close();

        } catch (Exception e) {
            e.printStackTrace();
            // 可以添加异常处理逻辑，比如记录日志或进行其他操作
        }
        return null;
    }

    // @Scheduled(cron = "0 */5 * * * ?")
    @Override
    public String getDiskUsage() {
        try {
            File file = new File("/");
            long totalSpace = file.getTotalSpace();
            long freeSpace = file.getFreeSpace();
            long usedSpace = totalSpace - freeSpace;

            Float diskUsage = (float) usedSpace / totalSpace * 100;

            log.info(diskUsage.toString());

            String uuId = UUID.randomUUID().toString();

            String date = this.getDate();

            kafkaMonitorMapper.addDiskUsage(uuId, date, diskUsage);

            return String.valueOf(diskUsage);
        } catch (Exception e) {
            throw e;
        }
    }

    // @Scheduled(cron = "0 */5 * * * ?")
    @Override
    public String getCpuUsage() {
        OperatingSystemMXBean osBean = (OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean();

        // 获取系统的CPU使用率
        // 返回值范围为[0,1]，乘以100转换为百分比
        Float cpuUsage = (float) osBean.getSystemCpuLoad() * 100;

        System.out.println("系统CPU使用率: " + cpuUsage + "%");

        String uuId = UUID.randomUUID().toString();

        String date = this.getDate();

        kafkaMonitorMapper.addCpuUsage(uuId, date, cpuUsage);

        return String.valueOf(cpuUsage);
    }

    // @Scheduled(cron = "0 */5 * * * ?")
    @Override
    public String getUnderReplicated() {
        try {
            JMXConnector jmxConnector = this.jmxConnector();
            MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();

            Map<String, Integer> topicAndPartition = kafkaConsumerServer.getTopicAndPartition();
            int totalUnderReplicated = 0;

            for (String key : topicAndPartition.keySet()) {

                for (int i = 0; i < topicAndPartition.get(key); i++) {
                    ObjectName underReplicated = new ObjectName("kafka.cluster:type=Partition,name=UnderReplicated,topic=" + key + ",partition=" + i);
                    Object underReplicatedValue = mBeanServerConnection.getAttribute(underReplicated, "Value");
                    // 将 Object 转换为整数
                    int value = Integer.parseInt(underReplicatedValue.toString());
                    log.info("分区缺失副本数：" + underReplicatedValue);

                    String uuId = UUID.randomUUID().toString();

                    String date = this.getDate();

                    int temp = 0;

                    temp += value;
                    kafkaMonitorMapper.addUnderreplicated(uuId, date, temp, key, i);
                }
            }

            jmxConnector.close();
            return null;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    // @Scheduled(cron = "0 */5 * * * ?")
    @Override
    public String queueDepth() {

//        Properties props = new Properties();
//        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "your-bootstrap-servers");
//
//        // 创建AdminClient
//        try (AdminClient adminClient = AdminClient.create(props)) {
//            // 指定要获取深度的主题和分区
//            String topic = "your-topic";
//            int partition = 0; // 你想要检查的分区号
//
//            // 构建TopicPartition对象
//            TopicPartition tp = new TopicPartition(topic, partition);
//
//            // 获取指定分区的偏移量信息
//            ListOffsetsResult offsetsResult = adminClient.listOffsets(Collections.singletonMap(tp, OffsetSpec.latest()));
//            KafkaFuture<ListOffsetsResult.ListOffsetsResultInfo> future = offsetsResult.values().get(tp);
//
//            // 获取偏移量信息并计算深度
//            ListOffsetsResult.ListOffsetsResultInfo resultInfo = future.get();
//            long latestOffset = resultInfo.offset();
//            long queueDepth = latestOffset + 1; // Kafka offset是从0开始的，所以要加1
//
//            System.out.println("Queue depth for topic " + topic + ", partition " + partition + ": " + queueDepth);
//        } catch (ExecutionException e) {
//            throw new RuntimeException(e);
//        } catch (InterruptedException e) {
//            throw new RuntimeException(e);
//        }
        return "";
    }

    // @Scheduled(cron = "0 */5 * * * ?")
    @Override
    public String getConsumerMessage() {
        try {
            JMXConnector jmxConnector = this.jmxConnector();
            MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();

            Map<String, Integer> topicAndPartition = kafkaConsumerServer.getTopicAndPartition();

            for (String key : topicAndPartition.keySet()) {
                int totalUnderReplicated = 0;
                for (int i = 0; i < topicAndPartition.get(key); i++) {
                    ObjectName consumerMessage = new ObjectName("kafka.log:type=Log,name=LogEndOffset,topic=" + key + ",partition=" + i);
                    Object consumerMessageValue = mBeanServerConnection.getAttribute(consumerMessage, "Value");
                    // 将 Object 转换为整数
                    int value = Integer.parseInt(consumerMessageValue.toString());
                    log.info(key + "-----partition" + i + "消费消息数：" + consumerMessageValue);

                    totalUnderReplicated += value;

                }
                String uuId = UUID.randomUUID().toString();

                String date = this.getDate();

                Integer temp = kafkaMonitorMapper.queryMaxNum(key);

                System.out.println(temp + "---" + totalUnderReplicated);

                int consumerNum = 0;
                if (temp != null) {
                    if (temp != totalUnderReplicated) {
                        consumerNum = totalUnderReplicated - temp;
                        float i1 = consumerNum / 5;
                        kafkaMonitorMapper.addConsumerMessage(uuId, date, consumerNum, key, i1);
                    }
                } else {
                    float i1 = totalUnderReplicated / 5;
                    kafkaMonitorMapper.addConsumerMessage(uuId, date, totalUnderReplicated, key, i1);
                }
            }

            jmxConnector.close();
            return null;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    private final MeterRegistry meterRegistry;

    @Autowired
    public KafkaMonitorServiceImpl(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
    }


    // @Scheduled(cron = "0 */5 * * * ?")
    @Override
    public String getProducerThrottledRate() {

        try {


            Map<String, Integer> topicAndPartition = kafkaConsumerServer.getTopicAndPartition();

            for (String key : topicAndPartition.keySet()) {
                Timer timer = meterRegistry.find("kafka.producer.request")
                        .tag("topic", key)
                        .timer();
                long l = timer != null ? timer.count() : 0;
                String uuId = UUID.randomUUID().toString();

                String date = this.getDate();
                kafkaMonitorMapper.addThrottledrate(uuId, date, l, "生产", key);
            }
        } catch (Exception e)
        {e.getStackTrace();}
        return null;
    }

    // @Scheduled(cron = "0 */5 * * * ?")
    @Override
    public String getConsumerThrottledRate() {
        try {
            Map<String, Integer> topicAndPartition = kafkaConsumerServer.getTopicAndPartition();

            for (Map.Entry<String, Integer> entry : topicAndPartition.entrySet()) {
                String key = entry.getKey();
                Timer timer = meterRegistry.find("kafka.consumer.records-consumed-total")
                        .tag("topic", key)
                        .timer();
                long l = timer != null ? timer.count() : 0;

                String uuid = UUID.randomUUID().toString();

                String date = getDate();
                kafkaMonitorMapper.addThrottledrate(uuid, date, l, "消费", key);
            }
        } catch (Exception e)
        {e.getStackTrace();}
        return null;
    }

}
