package com.yeejoin.amos.boot.module.jxiop.biz.kafka;

import cn.hutool.core.date.DateUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.core.toolkit.ObjectUtils;
import com.yeejoin.amos.boot.biz.common.utils.DateUtils;
import com.yeejoin.amos.boot.biz.common.utils.RedisUtils;
import com.yeejoin.amos.boot.module.jxiop.biz.entity.*;
import com.yeejoin.amos.boot.module.jxiop.biz.mapper2.IdxBizFanPointVarCentralValueMapper;
import com.yeejoin.amos.boot.module.jxiop.biz.mapper2.IdxBizPvPointVarCentralValueMapper;
import com.yeejoin.amos.boot.module.jxiop.biz.service.impl.*;
import com.yeejoin.amos.boot.module.jxiop.biz.tdmapper.IndicatorDataMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import tech.tablesaw.api.ColumnType;
import tech.tablesaw.api.DoubleColumn;
import tech.tablesaw.api.Table;
import tech.tablesaw.io.json.JsonReadOptions;
import tech.tablesaw.selection.Selection;

import java.util.*;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.stream.Collectors;

import static com.yeejoin.amos.boot.module.jxiop.biz.kafka.Constant.*;

/**
 * kafka 消费服务
 *
 * @author litw
 * @create 2022/11/1 10:06
 **/
@Slf4j
@Service
public class KafkaConsumerService {

    @Autowired
    private IndicatorDataMapper indicatorDataMapper;

    @Autowired
    RedisUtils redisUtils;

    @Autowired
    private IdxBizFanPointVarCorrelationServiceImpl idxBizFanPointVarCorrelationService;


    @Autowired
    private IdxBizPvPointVarCorrelationServiceImpl idxBizPvPointVarCorrelationService;

    @Autowired
    IdxBizFanPointProcessVariableClassificationServiceImpl idxBizFanPointProcessVariableClassificationService;


    @Autowired
    IdxBizPvPointProcessVariableClassificationServiceImpl idxBizPvPointProcessVariableClassificationService;

    @Autowired
    IdxBizFanPointVarCentralValueServiceImpl idxBizFanPointVarCentralValueService;

    @Autowired
    IdxBizFanPointVarCentralValueMapper idxBizFanPointVarCentralValueMapper;

    @Autowired
    IdxBizPvPointVarCentralValueServiceImpl idxBizPvPointVarCentralValueService;

    @Autowired
    IdxBizPvPointVarCentralValueMapper idxBizPvPointVarCentralValueMapper;

    // 相关性
    @Value("${base.url.XGX:http://139.9.171.247:8052/intelligent-analysis/correlation}")
    private String baseUrlXGX;

    // 工况划分
    @Value("${base.url.GKHF:http://139.9.171.247:8052/intelligent-analysis/working-condition-division}")
    private String baseUrlGKHF;

    @Value("${base.url.ZXZ:http://172.16.3.29:8052/intelligent-analysis/central-value}")
    private String zxzJsUrlFanBySF;

    @Value("${spring.kafka.consumer.max-poll-records:30}")
    private Integer threadNum = 30;


    @Value("${last.month.num:12}")
    private Integer lastMonthNum;

    ExecutorService service = Executors.newFixedThreadPool(threadNum);

    BlockingQueue<PointData> queue = new LinkedBlockingQueue<>();

    public KafkaConsumerService() {
        for (int i = 0 ; i < threadNum; i++) {
            service.execute(new Runnable() {
                @Override
                public void run() {
                    while(true) {
                        try {
                            PointData pointsData = queue.take();
                            List<ConsumerRecord<String, String>> consumerRecords = pointsData.getConsumerRecords();
                            Table table = pointsData.getTable();

                            if ("xgxFanConsumer".equals(pointsData.getOperatorType())){
                                execFanCorrelation(consumerRecords, table);
                            } else if ("xgxPvConsumer".equals(pointsData.getOperatorType())) {
                                execPvCorrelation(consumerRecords, table);
                            } else if ("GKHFFanConsumer".equals(pointsData.getOperatorType())) {
                                consumerRecordsGKFXFan(consumerRecords, table);
                            } else if ("GKHFPvConsumer".equals(pointsData.getOperatorType())) {
                                consumerRecordsGKFXPv(consumerRecords, table);
                            }  else if ("ZXZFanConsumer".equals(pointsData.getOperatorType())) {
                                consumerRecordsZXZFan(consumerRecords, pointsData);
                            } else if ("ZXZPvConsumer".equals(pointsData.getOperatorType())) {
                                consumerRecordsZXZPv(consumerRecords, pointsData);
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }

            });
        }
    }


    private void consumerRecordsZXZFan(List<ConsumerRecord<String, String>> consumerRecords, PointData pointsData ) {
        Table table = pointsData.getTable();

        Map<String, List<IdxBizFanPointProcessVariableClassification>> zxzIds = pointsData.getZxzIds();
        for (String id : zxzIds.keySet()) {
            List<IdxBizFanPointProcessVariableClassification> variableClassificationList = zxzIds.get(id);
            String  analysisVariableId = id;
            List<IdxBizFanPointProcessVariableClassification> processVariableList = variableClassificationList.stream().filter(v -> !id.equals(v.getSequenceNbr().toString())).collect(Collectors.toList());
            IdxBizFanPointProcessVariableClassification analysisVariable = variableClassificationList.stream().filter(v -> id.equals(v.getSequenceNbr().toString())).findFirst().get();

//            if (analysisVariable.getEquipmentName().equals("W2500#14风机") && analysisVariable.getPointName().equals("桨距角")) {
//                log.info("抓包数据：{}", JSON.toJSONString(analysisVariable));
//            }

            Map<String, Object> data1 = new HashMap<>();
            Map<String, Object> data2 = new HashMap<>();
            int index = 1;
            Table dataTable = Table.create();
            int minRow = 0;
            for (IdxBizFanPointProcessVariableClassification processVariable : processVariableList) {
                Selection selection = table.stringColumn("id").isEqualTo(processVariable.getIndexAddress() + "_" + processVariable.getGatewayId());
                DoubleColumn values = table.where(selection).doubleColumn("value");
                // 获取最小数据长度
                if (index == 1) {
                    minRow = values.size();
                } else {
                    minRow = minRow > values.size() ? values.size() : minRow;
                }
                values.setName("processVariable" + index);
                dataTable = getDataTable(dataTable, values);

                data1.put("processVariable" + index + "Id", processVariable.getSequenceNbr());

                // 构建工况区间数组
                List<Object> IntervalValues = new ArrayList<>();
                IntervalValues.add(processVariable.getIntervalValue1());
                IntervalValues.add(processVariable.getIntervalValue2());
                IntervalValues.add(processVariable.getIntervalValue3());
                IntervalValues.add(processVariable.getIntervalValue4());
                IntervalValues.add(processVariable.getIntervalValue5());
                data2.put("processVariable" + index, IntervalValues);

                index++;
            }
            Selection selection = table.stringColumn("id").isEqualTo(analysisVariable.getIndexAddress() + "_" + analysisVariable.getGatewayId());
            DoubleColumn values = table.where(selection).doubleColumn("value");
            values.setName("analysisVariable");
            dataTable = getDataTable(dataTable, values);
            data1.put("analysisVariableId", analysisVariable.getSequenceNbr());

            // 获取相同长度的数据
//            dataTable = dataTable.inRange(0, minRow);
            List<String> list = dataTable.columnNames();
            for (String column : list) {
                data1.put(column, dataTable.doubleColumn(column).asDoubleArray());
            }

            Map<String,Object> requestMap = new HashMap<>();
            requestMap.put("data1", data1);
            requestMap.put("data2", data2);
            String response = HttpUtil.createPost(zxzJsUrlFanBySF).body(JSON.toJSONString(requestMap)).execute().body();

            if (response.contains("stdDev")) {
                idxBizFanPointVarCentralValueMapper.delete(new QueryWrapper<IdxBizFanPointVarCentralValue>().eq("ANALYSIS_POINT_ID", analysisVariableId));

                JSONObject jsonObject = JSON.parseObject(response);
                int length = jsonObject.getJSONArray("stdDev").size() > 64 ? 64 : jsonObject.getJSONArray("stdDev").size();
                List<IdxBizFanPointVarCentralValue> insertList = new ArrayList<>();
                for (int i = 0; i < length; i++) {
                    IdxBizFanPointVarCentralValue idxBizFanPointVarCentralValue = new IdxBizFanPointVarCentralValue();
                    idxBizFanPointVarCentralValue.setProcess1Min(ObjectUtils.isNull(jsonObject.getJSONArray("process1Min").get(i)) ? null : jsonObject.getJSONArray("process1Min").getDoubleValue(i));
                    idxBizFanPointVarCentralValue.setProcess2Min(ObjectUtils.isNull(jsonObject.getJSONArray("process2Min").get(i)) ? null : jsonObject.getJSONArray("process2Min").getDoubleValue(i));
                    idxBizFanPointVarCentralValue.setProcess3Min(ObjectUtils.isNull(jsonObject.getJSONArray("process3Min").get(i)) ? null : jsonObject.getJSONArray("process3Min").getDoubleValue(i));
                    idxBizFanPointVarCentralValue.setProcess1Max(ObjectUtils.isNull(jsonObject.getJSONArray("process1Max").get(i)) ? null : jsonObject.getJSONArray("process1Max").getDoubleValue(i));
                    idxBizFanPointVarCentralValue.setPorcess2Max(ObjectUtils.isNull(jsonObject.getJSONArray("process2Max").get(i)) ? null : jsonObject.getJSONArray("process2Max").getDoubleValue(i));
                    idxBizFanPointVarCentralValue.setProcess3Max(ObjectUtils.isNull(jsonObject.getJSONArray("process3Max").get(i)) ? null : jsonObject.getJSONArray("process3Max").getDoubleValue(i));
                    idxBizFanPointVarCentralValue.setAnalysisPointId(analysisVariableId);
                    idxBizFanPointVarCentralValue.setAnalysisPointName(analysisVariable.getPointName());
                    idxBizFanPointVarCentralValue.setProcessPoint1Id(data1.get("processVariable1Id").toString());
                    idxBizFanPointVarCentralValue.setProcessPoint1Name(processVariableList.get(0).getPointName());
                    idxBizFanPointVarCentralValue.setProcessPoint2Id(data1.get("processVariable2Id").toString());
                    idxBizFanPointVarCentralValue.setProcessPoint2Name(processVariableList.get(1).getPointName());
                    idxBizFanPointVarCentralValue.setProcessPoint3Id(data1.get("processVariable3Id").toString());
                    idxBizFanPointVarCentralValue.setProcessPoint3Name(processVariableList.get(2).getPointName());
                    idxBizFanPointVarCentralValue.setAnalysisStdDev(ObjectUtils.isNull(jsonObject.getJSONArray("stdDev").get(i)) ? null : jsonObject.getJSONArray("stdDev").getDoubleValue(i));
                    idxBizFanPointVarCentralValue.setAnalysisCenterValue(ObjectUtils.isNull(jsonObject.getJSONArray("centerValue").get(i)) ? null : jsonObject.getJSONArray("centerValue").getDoubleValue(i));
                    idxBizFanPointVarCentralValue.setArae(analysisVariable.getArae());
                    idxBizFanPointVarCentralValue.setStation(analysisVariable.getStation());
                    idxBizFanPointVarCentralValue.setSubSystem(analysisVariable.getSubSystem());
                    idxBizFanPointVarCentralValue.setNumber(analysisVariable.getNumber());
                    idxBizFanPointVarCentralValue.setEquipmentName(analysisVariable.getEquipmentName());
                    insertList.add(idxBizFanPointVarCentralValue);
                }
                if (CollectionUtils.isNotEmpty(insertList)) {
                    idxBizFanPointVarCentralValueService.saveBatch(insertList);
                }
            }
            redisUtils.expire(kafkaTopicConsumerZXZFan, 600);
        }
    }


    private Table getDataTable(Table dataTable, DoubleColumn values) {
        if (!dataTable.isEmpty() && dataTable.rowCount() < values.size()) {
            dataTable.addColumns(values.inRange(0, dataTable.rowCount()));
        } else if (!dataTable.isEmpty() && dataTable.rowCount() > values.size()){
            dataTable = dataTable.inRange(0, values.size());
            dataTable.addColumns(values);
        } else {
            dataTable.addColumns(values);
        }
        return dataTable;
    }

    private void execPvCorrelation(List<ConsumerRecord<String, String>> consumerRecords, Table table) {
        consumerRecords.parallelStream().forEach(record -> {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                IdxBizPvPointVarCorrelation pvPointVarCorrelation = JSON.parseObject(kafkaMessage.get().toString(), IdxBizPvPointVarCorrelation.class);
                Selection selection = table.stringColumn("id").isEqualTo(pvPointVarCorrelation.getAnalysisIndexAddress() + "_" + pvPointVarCorrelation.getAnalysisGatewayId());
                double[] data1 = table.where(selection).doubleColumn("value").asDoubleArray();
                selection = table.stringColumn("id").isEqualTo(pvPointVarCorrelation.getProcessIndexAddress() + "_" + pvPointVarCorrelation.getProcessGatewayId());
                double[] data2 = table.where(selection).doubleColumn("value").asDoubleArray();

                int shortestLength = Math.min(data1.length, data2.length);
                data1 = subset(data1, shortestLength);
                data2 = subset(data2, shortestLength);

                HashMap<String, Object> resultMap = new HashMap<>();
                resultMap.put("data1", data1);
                resultMap.put("data2", data2);
                String response = HttpUtil.createPost(baseUrlXGX).body(JSON.toJSONString(resultMap)).execute().body();
                if (response.contains("correlation") && !response.contains("warning")) {
                    com.alibaba.fastjson.JSONObject jsonObject = JSON.parseObject(response);
                    pvPointVarCorrelation.setCorrelationCoefficient(jsonObject.getDoubleValue("correlation"));
                    log.info("------------------------------------------光伏相关性：：计算成功，待更新表数据----------------------------------------");
                } else {
                    pvPointVarCorrelation.setCorrelationCoefficient(0.0);
                }
                pvPointVarCorrelation.setRecDate(new Date());

                idxBizPvPointVarCorrelationService.saveOrUpdate(pvPointVarCorrelation);
            }
        });
        redisUtils.expire(kafkaTopicConsumerPv, 600);
    }

    private void execFanCorrelation(List<ConsumerRecord<String, String>> consumerRecords, Table table) {
        consumerRecords.parallelStream().forEach(record -> {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                IdxBizFanPointVarCorrelation fanPointVarCorrelation = JSON.parseObject(kafkaMessage.get().toString(), IdxBizFanPointVarCorrelation.class);
                Selection selection = table.stringColumn("id").isEqualTo(fanPointVarCorrelation.getAnalysisIndexAddress() + "_" + fanPointVarCorrelation.getAnalysisGatewayId());
                double[] data1 = table.where(selection).doubleColumn("value").asDoubleArray();
                selection = table.stringColumn("id").isEqualTo(fanPointVarCorrelation.getProcessIndexAddress() + "_" + fanPointVarCorrelation.getProcessGatewayId());
                double[] data2 = table.where(selection).doubleColumn("value").asDoubleArray();

                int shortestLength = Math.min(data1.length, data2.length);
                data1 = subset(data1, shortestLength);
                data2 = subset(data2, shortestLength);

                HashMap<String, Object> resultMap = new HashMap<>();
                resultMap.put("data1", data1);
                resultMap.put("data2", data2);
                String response = HttpUtil.createPost(baseUrlXGX).body(JSON.toJSONString(resultMap)).execute().body();
                if (response.contains("correlation") && !response.contains("warning")) {
                    com.alibaba.fastjson.JSONObject jsonObject = JSON.parseObject(response);
                    fanPointVarCorrelation.setCorrelationCoefficient(jsonObject.getDoubleValue("correlation"));
                    log.info("------------------------------------------风机相关性：：计算成功，待更新表数据----------------------------------------");
                } else {
                    fanPointVarCorrelation.setCorrelationCoefficient(0.0);
                }
                fanPointVarCorrelation.setRecDate(new Date());

                idxBizFanPointVarCorrelationService.saveOrUpdate(fanPointVarCorrelation);
            }
        });
        redisUtils.expire(kafkaTopicConsumer, 600);
    }

    public static double[] subset(double[] array, int length) {
        if (length >= array.length) {
            return array;
        } else {
            double[] subset = new double[length];
            System.arraycopy(array, 0, subset, 0, length);
            return subset;
        }
    }

    class PointData {
        private List<ConsumerRecord<String, String>> consumerRecords;
        private Table table;
        private String operatorType;

        private Map<String, List<IdxBizFanPointProcessVariableClassification>> zxzIds;

        private Map<String, List<IdxBizPvPointProcessVariableClassification>> zxzPvIds;


        public PointData(List<ConsumerRecord<String, String>> consumerRecords, Table table, String operatorType) {
            this.consumerRecords = consumerRecords;
            this.table = table;
            this.operatorType = operatorType;
        }

        public PointData(List<ConsumerRecord<String, String>> consumerRecords, Table table, String operatorType, Map<String, List<IdxBizFanPointProcessVariableClassification>> zxzIds) {
            this.consumerRecords = consumerRecords;
            this.table = table;
            this.operatorType = operatorType;
            this.zxzIds = zxzIds;
        }

        public PointData(List<ConsumerRecord<String, String>> consumerRecords, Table table, String operatorType, Map<String, List<IdxBizPvPointProcessVariableClassification>> zxzPvIds, String notString) {
            this.consumerRecords = consumerRecords;
            this.table = table;
            this.operatorType = operatorType;
            this.zxzPvIds = zxzPvIds;
        }
        public Map<String, List<IdxBizPvPointProcessVariableClassification>> getZxzIdsPv() {
            return zxzPvIds;
        }

        public Map<String, List<IdxBizFanPointProcessVariableClassification>> getZxzIds() {
            return zxzIds;
        }

        public String getOperatorType() {
            return operatorType;
        }

        public List<ConsumerRecord<String, String>> getConsumerRecords() {
            return consumerRecords;
        }

        public Table getTable() {
            return table;
        }
    }
    /**
     * 批量消费kafka消息         【风电站 相关性】
     *
     * @param consumerRecords messages
     * @param ack             ack
     */
    @KafkaListener(id = "xgxFanConsumer", groupId = "consumerGroup", topics = kafkaTopicConsumer)
    public void listenXGXFan(List<ConsumerRecord<String, String>> consumerRecords, Acknowledgment ack) {
        try {
            ack.acknowledge();
            Map<String, Set<String>> gatewayPoints = new HashMap<>();
            consumerRecords.stream().forEach(record -> {
                Optional<?> kafkaMessage = Optional.ofNullable(record.value());
                if (kafkaMessage.isPresent()) {
                    IdxBizFanPointVarCorrelation fanPointVarCorrelation = JSON.parseObject(kafkaMessage.get().toString(), IdxBizFanPointVarCorrelation.class);
                    Set<String> idSet = null;
                    if (gatewayPoints.containsKey(fanPointVarCorrelation.getAnalysisGatewayId())) {
                        idSet = gatewayPoints.get(fanPointVarCorrelation.getAnalysisGatewayId());
                    } else {
                        idSet = new HashSet<>();
                    }
                    idSet.add(fanPointVarCorrelation.getAnalysisIndexAddress());
                    idSet.add(fanPointVarCorrelation.getProcessIndexAddress());
                    gatewayPoints.put(fanPointVarCorrelation.getAnalysisGatewayId(), idSet);
                }
            });
            buildExecData(consumerRecords, gatewayPoints, "xgxFanConsumer");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 批量消费kafka消息        【光伏相关性 】
     *
     * @param consumerRecords messages
     * @param ack             ack
     */
    @KafkaListener(id = "xgxPvConsumer", groupId = "consumerGroup", topics = kafkaTopicConsumerPv)
    public void listenXGXPv(List<ConsumerRecord<String, String>> consumerRecords, Acknowledgment ack) {

        try {
            ack.acknowledge();
            Map<String, Set<String>> gatewayPoints = new HashMap<>();
            consumerRecords.stream().forEach(record -> {
                Optional<?> kafkaMessage = Optional.ofNullable(record.value());
                if (kafkaMessage.isPresent()) {
                    IdxBizPvPointVarCorrelation pvPointVarCorrelation = JSON.parseObject(kafkaMessage.get().toString(), IdxBizPvPointVarCorrelation.class);
                    Set<String> idSet = null;
                    if (gatewayPoints.containsKey(pvPointVarCorrelation.getAnalysisGatewayId())) {
                        idSet = gatewayPoints.get(pvPointVarCorrelation.getAnalysisGatewayId());
                    } else {
                        idSet = new HashSet<>();
                    }
                    idSet.add(pvPointVarCorrelation.getAnalysisIndexAddress().toString());
                    idSet.add(pvPointVarCorrelation.getProcessIndexAddress());
                    gatewayPoints.put(pvPointVarCorrelation.getAnalysisGatewayId(), idSet);
                }
            });
            buildExecData(consumerRecords, gatewayPoints, "xgxPvConsumer");
        } finally {

        }
    }


    private void buildZXZExecData(List<ConsumerRecord<String, String>> consumerRecords, Map<String, Set<String>> gatewayPoints, Map<String, List<IdxBizFanPointProcessVariableClassification>> zxzIds, String xgxPvConsumer) {
        for (String gatewayId : gatewayPoints.keySet()) {
            String join = String.join(",", gatewayPoints.get(gatewayId));
            List<IndicatorData> indicatorData = indicatorDataMapper.selectByAddresses(join, gatewayId);

            JsonReadOptions options = JsonReadOptions.builderFromString(JSON.toJSONString(indicatorData))
                    .columnTypes(new Function<String, ColumnType>() {
                        @Override
                        public ColumnType apply(String t) {
                            if (t.equals("value")) {
                                return ColumnType.DOUBLE;
                            }
                            return ColumnType.STRING;
                        }
                    }).build();
            Table table = Table.read().usingOptions(options);
            queue.add(new PointData(consumerRecords, table, xgxPvConsumer, zxzIds));
        }
    }

    private void buildExecData(List<ConsumerRecord<String, String>> consumerRecords, Map<String, Set<String>> gatewayPoints, String xgxPvConsumer) {
        for (String gatewayId : gatewayPoints.keySet()) {
            String join = String.join(",", gatewayPoints.get(gatewayId));
            List<IndicatorData> indicatorData = indicatorDataMapper.selectByAddresses(join, gatewayId);

            JsonReadOptions options = JsonReadOptions.builderFromString(JSON.toJSONString(indicatorData))
                    .columnTypes(new Function<String, ColumnType>() {
                        @Override
                        public ColumnType apply(String t) {
                            if (t.equals("value")) {
                                return ColumnType.DOUBLE;
                            }
                            return ColumnType.STRING;
                        }
                    }).build();
            Table table = Table.read().usingOptions(options);
            queue.add(new PointData(consumerRecords, table, xgxPvConsumer));
        }
    }


    /**
     * 批量消费kafka消息        【风电   工况划分 】
     *
     * @param consumerRecords messages
     * @param ack             ack
     */
    @KafkaListener(id = "GKHFFanConsumer", groupId = "consumerGroup", topics = kafkaTopicConsumerGKHFFan)
    public void listenGKHFFan(List<ConsumerRecord<String, String>> consumerRecords, Acknowledgment ack) {
        try {
            ack.acknowledge();
            Map<String, Set<String>> gatewayPoints = new HashMap<>();
            consumerRecords.stream().forEach(record -> {
                Optional<?> kafkaMessage = Optional.ofNullable(record.value());
                if (kafkaMessage.isPresent()) {
                    IdxBizFanPointProcessVariableClassification fanPointProcessVariable = JSON.parseObject(kafkaMessage.get().toString(), IdxBizFanPointProcessVariableClassification.class);
                    Set<String> idSet = null;
                    if (gatewayPoints.containsKey(fanPointProcessVariable.getGatewayId())) {
                        idSet = gatewayPoints.get(fanPointProcessVariable.getGatewayId());
                    } else {
                        idSet = new HashSet<>();
                    }
                    idSet.add(fanPointProcessVariable.getIndexAddress());
                    gatewayPoints.put(fanPointProcessVariable.getGatewayId(), idSet);
                }
            });
            buildExecData(consumerRecords, gatewayPoints, "GKHFFanConsumer");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 风电   工况划分 处理
     * @param consumerRecords
     * @param table
     * @return
     */
    boolean consumerRecordsGKFXFan(List<ConsumerRecord<String, String>> consumerRecords, Table table) {
        try {
            consumerRecords.parallelStream().forEach(record -> {
                Optional<?> kafkaMessage = Optional.ofNullable(record.value());
                if (kafkaMessage.isPresent()) {
                    IdxBizFanPointProcessVariableClassification fanPointVarCorrelation = JSON.parseObject(kafkaMessage.get().toString(), IdxBizFanPointProcessVariableClassification.class);
                    HashMap<String, Object> resultMap = new HashMap<>();
                    Selection selection = table.stringColumn("id").isEqualTo(fanPointVarCorrelation.getIndexAddress() + "_" + fanPointVarCorrelation.getGatewayId());
                    double[] data1 = table.where(selection).doubleColumn("value").asDoubleArray();
                    resultMap.put("processVariable", data1);
                    resultMap.put("processVariableId", fanPointVarCorrelation.getSequenceNbr());

                    String response = HttpUtil.createPost(baseUrlGKHF).body(JSON.toJSONString(resultMap)).execute().body();

                    if (response.contains("intervalValue1") && response.contains("processVariableId")) {
                        com.alibaba.fastjson.JSONObject jsonObject = JSON.parseObject(response);

                        fanPointVarCorrelation.setIntervalValue5(jsonObject.getDoubleValue("intervalValue5"));
                        fanPointVarCorrelation.setIntervalValue4(jsonObject.getDoubleValue("intervalValue4"));
                        fanPointVarCorrelation.setIntervalValue3(jsonObject.getDoubleValue("intervalValue3"));
                        fanPointVarCorrelation.setIntervalValue2(jsonObject.getDoubleValue("intervalValue2"));
                        fanPointVarCorrelation.setIntervalValue1(jsonObject.getDoubleValue("intervalValue1"));
                        log.info("------------------------------------------光伏相关性：：计算成功，待更新表数据----------------------------------------");
                    } else {
                        fanPointVarCorrelation.setIntervalValue5(0.0);
                        fanPointVarCorrelation.setIntervalValue4(0.0);
                        fanPointVarCorrelation.setIntervalValue3(0.0);
                        fanPointVarCorrelation.setIntervalValue2(0.0);
                        fanPointVarCorrelation.setIntervalValue1(0.0);
                    }
                    fanPointVarCorrelation.setRecDate(new Date());
                    idxBizFanPointProcessVariableClassificationService.saveOrUpdate(fanPointVarCorrelation);
                    log.info("表数据已更新");
                }
            });
        } catch (Exception e) {
            log.error("kafka失败，当前失败的批次");
        } finally {
            redisUtils.expire(kafkaTopicConsumerGKHFFan, 600);
        }
        return true;
    }



    /**
     * 批量消费kafka消息        【光伏  工况划分 】
     *
     * @param consumerRecords messages
     * @param ack             ack
     */
    @KafkaListener(id = "GKHFPvConsumer", groupId = "consumerGroup", topics = kafkaTopicConsumerGKHFPv)
    public void listenGKHFPv(List<ConsumerRecord<String, String>> consumerRecords, Acknowledgment ack) {

        try {
            ack.acknowledge();
            Map<String, Set<String>> gatewayPoints = new HashMap<>();
            consumerRecords.stream().forEach(record -> {
                Optional<?> kafkaMessage = Optional.ofNullable(record.value());
                if (kafkaMessage.isPresent()) {
                    IdxBizPvPointProcessVariableClassification pvPointProcessVariable = JSON.parseObject(kafkaMessage.get().toString(), IdxBizPvPointProcessVariableClassification.class);
                    Set<String> idSet = null;
                    if (gatewayPoints.containsKey(pvPointProcessVariable.getGatewayId())) {
                        idSet = gatewayPoints.get(pvPointProcessVariable.getGatewayId());
                    } else {
                        idSet = new HashSet<>();
                    }
                    idSet.add(pvPointProcessVariable.getIndexAddress());
                    gatewayPoints.put(pvPointProcessVariable.getGatewayId(), idSet);
                }
            });
            buildExecData(consumerRecords, gatewayPoints, "GKHFPvConsumer");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }



    /**
     * 风电   工况划分 处理
     * @param consumerRecords
     * @return
     */
    boolean consumerRecordsGKFXPv(List<ConsumerRecord<String, String>> consumerRecords, Table table) {
        try {
            consumerRecords.parallelStream().forEach(record -> {
                Optional<?> kafkaMessage = Optional.ofNullable(record.value());
                if (kafkaMessage.isPresent()) {
                    IdxBizPvPointProcessVariableClassification pvPointVarCorrelation = JSON.parseObject(kafkaMessage.get().toString(), IdxBizPvPointProcessVariableClassification.class);
                    HashMap<String, Object> resultMap = new HashMap<>();
                    Selection selection = table.stringColumn("id").isEqualTo(pvPointVarCorrelation.getIndexAddress() + "_" + pvPointVarCorrelation.getGatewayId());
                    double[] data1 = table.where(selection).doubleColumn("value").asDoubleArray();
                    resultMap.put("processVariable", data1);
                    resultMap.put("processVariableId", pvPointVarCorrelation.getSequenceNbr());

                    String response = HttpUtil.createPost(baseUrlGKHF).body(JSON.toJSONString(resultMap)).execute().body();

                    if (response.contains("intervalValue1") && response.contains("processVariableId")) {
                        com.alibaba.fastjson.JSONObject jsonObject = JSON.parseObject(response);

                        pvPointVarCorrelation.setIntervalValue5(jsonObject.getDoubleValue("intervalValue5"));
                        pvPointVarCorrelation.setIntervalValue4(jsonObject.getDoubleValue("intervalValue4"));
                        pvPointVarCorrelation.setIntervalValue3(jsonObject.getDoubleValue("intervalValue3"));
                        pvPointVarCorrelation.setIntervalValue2(jsonObject.getDoubleValue("intervalValue2"));
                        pvPointVarCorrelation.setIntervalValue1(jsonObject.getDoubleValue("intervalValue1"));
                        log.info("------------------------------------------光伏相关性：：计算成功，待更新表数据----------------------------------------");
                    } else {
                        pvPointVarCorrelation.setIntervalValue5(0.0);
                        pvPointVarCorrelation.setIntervalValue4(0.0);
                        pvPointVarCorrelation.setIntervalValue3(0.0);
                        pvPointVarCorrelation.setIntervalValue2(0.0);
                        pvPointVarCorrelation.setIntervalValue1(0.0);
                    }
                    pvPointVarCorrelation.setRecDate(new Date());
                    idxBizPvPointProcessVariableClassificationService.saveOrUpdate(pvPointVarCorrelation);
                    log.info("表数据已更新");
                }
            });
        } catch (Exception e) {
            log.error("kafka失败，当前失败的批次");
        } finally {
            redisUtils.expire(kafkaTopicConsumerGKHFPv, 600);
        }
        return true;
    }









    /**
     * 批量消费kafka消息        【风电   中心值 】
     *
     * @param consumerRecords messages
     * @param ack             ack
     */
    @KafkaListener(id = "ZXZFanConsumer", groupId = "consumerGroup", topics = kafkaTopicConsumerZXZFan)
    public void listenZXZFan(List<ConsumerRecord<String, String>> consumerRecords, Acknowledgment ack) {
        try {
            ack.acknowledge();
            Map<String, Set<String>> gatewayPoints = new HashMap<>();
            Map<String,  List<IdxBizFanPointProcessVariableClassification>> zxzIds = new HashMap<>();
            consumerRecords.stream().forEach(record -> {
                Optional<?> kafkaMessage = Optional.ofNullable(record.value());
                if (kafkaMessage.isPresent()) {
                    IdxBizFanPointProcessVariableClassification fanPointProcessVariable = JSON.parseObject(kafkaMessage.get().toString(), IdxBizFanPointProcessVariableClassification.class);

                    List<IdxBizFanPointVarCorrelation> gongkuangList = idxBizFanPointVarCorrelationService.list(new QueryWrapper<IdxBizFanPointVarCorrelation>().eq("ANALYSIS_GATEWAY_ID", fanPointProcessVariable.getGatewayId()).eq("ANALYSIS_POINT_ID", fanPointProcessVariable.getSequenceNbr()).orderByDesc("CORRELATION_COEFFICIENT").last("limit 3"));
                    List<String> processPointIds = gongkuangList.stream().map(idxBizFanPointVarCorrelation -> idxBizFanPointVarCorrelation.getProcessPointId().toString()).collect(Collectors.toList());
                    List<IdxBizFanPointProcessVariableClassification> idxBizFanPointProcessVariableClassificationList = idxBizFanPointProcessVariableClassificationService.list(new QueryWrapper<IdxBizFanPointProcessVariableClassification>().in("SEQUENCE_NBR", processPointIds));

                    idxBizFanPointProcessVariableClassificationList.add(fanPointProcessVariable);

                    zxzIds.put(fanPointProcessVariable.getSequenceNbr(), idxBizFanPointProcessVariableClassificationList);

                    idxBizFanPointProcessVariableClassificationList.forEach(item -> {
                        Set<String> idSet = null;
                        if (gatewayPoints.containsKey(fanPointProcessVariable.getGatewayId())) {
                            idSet = gatewayPoints.get(fanPointProcessVariable.getGatewayId());
                        } else {
                            idSet = new HashSet<>();
                        }
                        idSet.add(item.getIndexAddress());
                        gatewayPoints.put(fanPointProcessVariable.getGatewayId(), idSet);
                    });

                }
            });
            buildZXZExecData(consumerRecords, gatewayPoints, zxzIds, "ZXZFanConsumer");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }








    /**
     * 批量消费kafka消息        【光伏   中心值 】
     *
     * @param consumerRecords messages
     * @param ack             ack
     */
    @KafkaListener(id = "ZXZPvConsumer", groupId = "consumerGroup", topics = kafkaTopicConsumerZXZPv)
    public void listenZXZPv(List<ConsumerRecord<String, String>> consumerRecords, Acknowledgment ack) {
        try {
            ack.acknowledge();
            Map<String, Set<String>> gatewayPoints = new HashMap<>();
            Map<String,  List<IdxBizPvPointProcessVariableClassification>> zxzIds = new HashMap<>();
            consumerRecords.stream().forEach(record -> {
                Optional<?> kafkaMessage = Optional.ofNullable(record.value());
                if (kafkaMessage.isPresent()) {
                    IdxBizPvPointProcessVariableClassification pvPointProcessVariable = JSON.parseObject(kafkaMessage.get().toString(), IdxBizPvPointProcessVariableClassification.class);

                    List<IdxBizPvPointVarCorrelation> gongkuangList = idxBizPvPointVarCorrelationService.list(new QueryWrapper<IdxBizPvPointVarCorrelation>().eq("ANALYSIS_GATEWAY_ID", pvPointProcessVariable.getGatewayId()).eq("ANALYSIS_POINT_ID", pvPointProcessVariable.getSequenceNbr()).orderByDesc("CORRELATION_COEFFICIENT").last("limit 3"));
                    List<String> processPointIds = gongkuangList.stream().map(idxBizFanPointVarCorrelation -> idxBizFanPointVarCorrelation.getProcessPointId().toString()).collect(Collectors.toList());
                    List<IdxBizPvPointProcessVariableClassification> idxBizPvPointProcessVariableClassificationList = idxBizPvPointProcessVariableClassificationService.list(new QueryWrapper<IdxBizPvPointProcessVariableClassification>().in("SEQUENCE_NBR", processPointIds));

                    idxBizPvPointProcessVariableClassificationList.add(pvPointProcessVariable);

                    zxzIds.put(pvPointProcessVariable.getSequenceNbr(), idxBizPvPointProcessVariableClassificationList);

                    idxBizPvPointProcessVariableClassificationList.forEach(item -> {
                        Set<String> idSet = null;
                        if (gatewayPoints.containsKey(pvPointProcessVariable.getGatewayId())) {
                            idSet = gatewayPoints.get(pvPointProcessVariable.getGatewayId());
                        } else {
                            idSet = new HashSet<>();
                        }
                        idSet.add(item.getIndexAddress());
                        gatewayPoints.put(pvPointProcessVariable.getGatewayId(), idSet);
                    });

                }
            });
            buildZXZPvExecData(consumerRecords, gatewayPoints, zxzIds, "ZXZPvConsumer");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    private void buildZXZPvExecData(List<ConsumerRecord<String, String>> consumerRecords, Map<String, Set<String>> gatewayPoints, Map<String, List<IdxBizPvPointProcessVariableClassification>> zxzIds, String xgxPvConsumer) {
        for (String gatewayId : gatewayPoints.keySet()) {
            String join = String.join(",", gatewayPoints.get(gatewayId));
            List<IndicatorData> indicatorData = indicatorDataMapper.selectByAddresses(join, gatewayId);

            JsonReadOptions options = JsonReadOptions.builderFromString(JSON.toJSONString(indicatorData))
                    .columnTypes(new Function<String, ColumnType>() {
                        @Override
                        public ColumnType apply(String t) {
                            if (t.equals("value")) {
                                return ColumnType.DOUBLE;
                            }
                            return ColumnType.STRING;
                        }
                    }).build();
            Table table = Table.read().usingOptions(options);
            queue.add(new PointData(consumerRecords, table, xgxPvConsumer, zxzIds, ""));
        }
    }


    /**
     * 中心值 - 光伏
     * @param consumerRecords
     * @param pointsData
     */
    private void consumerRecordsZXZPv(List<ConsumerRecord<String, String>> consumerRecords, PointData pointsData ) {
        Table table = pointsData.getTable();

        Map<String, List<IdxBizPvPointProcessVariableClassification>> zxzIds = pointsData.getZxzIdsPv();
        for (String id : zxzIds.keySet()) {
            List<IdxBizPvPointProcessVariableClassification> variableClassificationList = zxzIds.get(id);
            String  analysisVariableId = id;
            List<IdxBizPvPointProcessVariableClassification> processVariableList = variableClassificationList.stream().filter(v -> !id.equals(v.getSequenceNbr().toString())).collect(Collectors.toList());
            IdxBizPvPointProcessVariableClassification analysisVariable = variableClassificationList.stream().filter(v -> id.equals(v.getSequenceNbr().toString())).findFirst().get();

            Map<String, Object> data1 = new HashMap<>();
            Map<String, Object> data2 = new HashMap<>();
            int index = 1;
            Table dataTable = Table.create();
            int minRow = 0;
            for (IdxBizPvPointProcessVariableClassification processVariable : processVariableList) {
                Selection selection = table.stringColumn("id").isEqualTo(processVariable.getIndexAddress() + "_" + processVariable.getGatewayId());
                DoubleColumn values = table.where(selection).doubleColumn("value");
                // 获取最小数据长度
                if (index == 1) {
                    minRow = values.size();
                } else {
                    minRow = minRow > values.size() ? values.size() : minRow;
                }
                values.setName("processVariable" + index);
                dataTable = getDataTable(dataTable, values);
                data1.put("processVariable" + index + "Id", processVariable.getSequenceNbr());

                // 构建工况区间数组
                List<Object> IntervalValues = new ArrayList<>();
                IntervalValues.add(processVariable.getIntervalValue1());
                IntervalValues.add(processVariable.getIntervalValue2());
                IntervalValues.add(processVariable.getIntervalValue3());
                IntervalValues.add(processVariable.getIntervalValue4());
                IntervalValues.add(processVariable.getIntervalValue5());
                data2.put("processVariable" + index, IntervalValues);

                index++;
            }
            Selection selection = table.stringColumn("id").isEqualTo(analysisVariable.getIndexAddress() + "_" + analysisVariable.getGatewayId());
            DoubleColumn values = table.where(selection).doubleColumn("value");
            values.setName("analysisVariable");
            dataTable = getDataTable(dataTable, values);

            data1.put("analysisVariableId", analysisVariable.getSequenceNbr());

            // 获取相同长度的数据
//            dataTable = dataTable.inRange(0, minRow);
            List<String> list = dataTable.columnNames();
            for (String column : list) {
                data1.put(column, dataTable.doubleColumn(column).asDoubleArray());
            }

            Map<String,Object> requestMap = new HashMap<>();
            requestMap.put("data1", data1);
            requestMap.put("data2", data2);
            String response = HttpUtil.createPost(zxzJsUrlFanBySF).body(JSON.toJSONString(requestMap)).execute().body();

            if (response.contains("stdDev")) {
                idxBizPvPointVarCentralValueMapper.delete(new QueryWrapper<IdxBizPvPointVarCentralValue>().eq("ANALYSIS_POINT_ID", analysisVariable.getSequenceNbr()));

                JSONObject jsonObject = JSON.parseObject(response);
                int length = jsonObject.getJSONArray("stdDev").size() > 64 ? 64 : jsonObject.getJSONArray("stdDev").size();
                List<IdxBizPvPointVarCentralValue> insertList = new ArrayList<>();
                for (int i = 0; i < length; i++) {
                    IdxBizPvPointVarCentralValue idxBizPvPointVarCentralValue = new IdxBizPvPointVarCentralValue();
                    idxBizPvPointVarCentralValue.setProcess1Min(ObjectUtils.isNull(jsonObject.getJSONArray("process1Min").get(i)) ? null : jsonObject.getJSONArray("process1Min").getDoubleValue(i));
                    idxBizPvPointVarCentralValue.setProcess2Min(ObjectUtils.isNull(jsonObject.getJSONArray("process2Min").get(i)) ? null : jsonObject.getJSONArray("process2Min").getDoubleValue(i));
                    idxBizPvPointVarCentralValue.setProcess3Min(ObjectUtils.isNull(jsonObject.getJSONArray("process3Min").get(i)) ? null : jsonObject.getJSONArray("process3Min").getDoubleValue(i));
                    idxBizPvPointVarCentralValue.setProcess1Max(ObjectUtils.isNull(jsonObject.getJSONArray("process1Max").get(i)) ? null : jsonObject.getJSONArray("process1Max").getDoubleValue(i));
                    idxBizPvPointVarCentralValue.setProcess2Max(ObjectUtils.isNull(jsonObject.getJSONArray("process2Max").get(i)) ? null : jsonObject.getJSONArray("process2Max").getDoubleValue(i));
                    idxBizPvPointVarCentralValue.setProcess3Max(ObjectUtils.isNull(jsonObject.getJSONArray("process3Max").get(i)) ? null : jsonObject.getJSONArray("process3Max").getDoubleValue(i));
                    idxBizPvPointVarCentralValue.setAnalysisPointId(jsonObject.getString("analysisVariableId"));
                    idxBizPvPointVarCentralValue.setAnalysisPointIdName(analysisVariable.getPointName());
                    idxBizPvPointVarCentralValue.setProcessPoint1Id(jsonObject.getString("processVariable1Id"));
                    idxBizPvPointVarCentralValue.setProcessPoint1IdName(processVariableList.get(0).getPointName());
                    idxBizPvPointVarCentralValue.setProcessPoint2Id(jsonObject.getString("processVariable2Id"));
                    idxBizPvPointVarCentralValue.setProcessPoint2IdName(processVariableList.get(1).getPointName());
                    idxBizPvPointVarCentralValue.setProcessPoint3Id(jsonObject.getString("processVariable3Id"));
                    idxBizPvPointVarCentralValue.setProcessPoint3IdName(processVariableList.get(2).getPointName());
                    idxBizPvPointVarCentralValue.setAnalysisStdDev(ObjectUtils.isNull(jsonObject.getJSONArray("stdDev").get(i)) ? null : jsonObject.getJSONArray("stdDev").getDoubleValue(i));
                    idxBizPvPointVarCentralValue.setAnalysisCenterValue(ObjectUtils.isNull(jsonObject.getJSONArray("centerValue").get(i)) ? null : jsonObject.getJSONArray("centerValue").getDoubleValue(i));
                    idxBizPvPointVarCentralValue.setArae(analysisVariable.getArae());
                    idxBizPvPointVarCentralValue.setStation(analysisVariable.getStation());
                    idxBizPvPointVarCentralValue.setSubarray(analysisVariable.getSubarray());
                    idxBizPvPointVarCentralValue.setManufacturer(analysisVariable.getManufacturer());
                    idxBizPvPointVarCentralValue.setEquipmentName(analysisVariable.getEquipmentName());
                    insertList.add(idxBizPvPointVarCentralValue);
                }
                if (CollectionUtils.isNotEmpty(insertList)) {
                    idxBizPvPointVarCentralValueService.saveBatch(insertList);
                }
            }

        }
        redisUtils.expire(kafkaTopicConsumerZXZPv, 600);
    }






















    /**
     * 风电处理消息   -  弃用
     * @param consumerRecord
     * @return
     */
    boolean consumerRecords(ConsumerRecord<String, String> consumerRecord) {

        try {
            Optional<?> kafkaMessage = Optional.ofNullable(consumerRecord.value());
            if (kafkaMessage.isPresent()) {
                IdxBizFanPointVarCorrelation fanPointVarCorrelation = JSON.parseObject(kafkaMessage.get().toString(), IdxBizFanPointVarCorrelation.class);

                List<IndicatorData> tdengineData1 = indicatorDataMapper.selectDataById(fanPointVarCorrelation.getAnalysisIndexAddress() + "_" + fanPointVarCorrelation.getAnalysisGatewayId());
//                List<IndicatorData> tdengineData1 = indicatorDataMapper.selectDataByAddressAndtimeNotDate(fanPointVarCorrelation.getAnalysisIndexAddress(), fanPointVarCorrelation.getAnalysisGatewayId());
                List<IndicatorData> tdengineData2 = indicatorDataMapper.selectDataById(fanPointVarCorrelation.getProcessIndexAddress() + "_" + fanPointVarCorrelation.getProcessGatewayId());

//        List<IndicatorData> tdengineData2 = indicatorDataMapper.selectDataByAddressAndtimeNotDate(fanPointVarCorrelation.getProcessIndexAddress(), fanPointVarCorrelation.getProcessGatewayId());
                List<Double> data1 = tdengineData1.stream().map(t -> Double.parseDouble(t.getValue())).collect(Collectors.toList());
                List<Double> data2 = tdengineData2.stream().map(t -> Double.parseDouble(t.getValue())).collect(Collectors.toList());

//        List<Double> data1 = new ArrayList<>();
//                List<Double> data2 = new ArrayList<>();
//                tdengineData1.forEach(item -> {
//                    if (item.getAddress().equals(fanPointVarCorrelation.getAnalysisIndexAddress()) && item.getGatewayId().equals(fanPointVarCorrelation.getAnalysisGatewayId())) {
//                        data1.add(Double.parseDouble(item.getValue()));
//                    } else {
//                        data2.add(Double.parseDouble(item.getValue()));
//                    }
//                });
                if (data1.size() < data2.size()) {
                    Integer a = data2.size() - data1.size();
                    for (int i = 0; i < a; i++) {
                        data2.remove(0);
                    }
                } else if (data2.size() < data1.size()) {
                    Integer a = data1.size() - data2.size();
                    for (int i = 0; i < a; i++) {
                        data1.remove(0);
                    }
                }
                HashMap<String, Object> resultMap = new HashMap<>();
                resultMap.put("data1", data1);
                resultMap.put("data2", data2);
                String response = HttpUtil.createPost(baseUrlXGX).body(JSON.toJSONString(resultMap)).execute().body();
                if (response.contains("correlation") && !response.contains("warning")) {
                    com.alibaba.fastjson.JSONObject jsonObject = JSON.parseObject(response);
                    fanPointVarCorrelation.setCorrelationCoefficient(jsonObject.getDoubleValue("correlation"));
                    log.info("------------------------------------------风机相关性：：计算成功，待更新表数据----------------------------------------");
                } else {
                    fanPointVarCorrelation.setCorrelationCoefficient(0.0);
                }
                fanPointVarCorrelation.setRecDate(new Date());

                idxBizFanPointVarCorrelationService.saveOrUpdate(fanPointVarCorrelation);
                log.info("表数据已更新");
                log.info("----------------------------风机相关性--------------分析变量与工况变量相关性分析算法结束----------------------------------------");
                log.info("kafka消费zhTestGroup消息{}", consumerRecord);
            }
        } catch (Exception e) {
            log.error("kafka失败，当前失败的批次: data:{}, {}", consumerRecord, e);
        } finally {
            redisUtils.expire(kafkaTopicConsumer, 600);
        }
        return true;
    }

    /**
     * 风电   工况划分 处理 -弃用
     * @param consumerRecord
     * @return
     */
    boolean consumerRecordsGKFXFan(ConsumerRecord<String, String> consumerRecord) {
        try {
            String startTime = DateUtils.convertDateToString(DateUtil.offsetMonth(new Date(), -lastMonthNum), DateUtils.DATE_TIME_PATTERN);
            String endTime = DateUtils.convertDateToString(DateUtils.getCurrentDayEndTime(new Date()), DateUtils.DATE_TIME_PATTERN);

            Optional<?> kafkaMessage = Optional.ofNullable(consumerRecord.value());
            if (kafkaMessage.isPresent()) {
                IdxBizFanPointProcessVariableClassification fanPointVarCorrelation = JSON.parseObject(kafkaMessage.get().toString(), IdxBizFanPointProcessVariableClassification.class);
                List<IndicatorData> tdengineData1 = indicatorDataMapper.selectDataByAddressAndtime(fanPointVarCorrelation.getIndexAddress(), startTime, endTime, fanPointVarCorrelation.getGatewayId());

                HashMap<String, Object> resultMap = new HashMap<>();
                resultMap.put("processVariable", tdengineData1.stream().map(t -> Double.parseDouble(t.getValue())).collect(Collectors.toList()));
                resultMap.put("processVariableId", fanPointVarCorrelation.getSequenceNbr());
                String response = HttpUtil.createPost(baseUrlGKHF).body(JSON.toJSONString(resultMap)).execute().body();

                if (response.contains("intervalValue1") && response.contains("processVariableId")) {
                    com.alibaba.fastjson.JSONObject jsonObject = JSON.parseObject(response);

                    fanPointVarCorrelation.setIntervalValue5(jsonObject.getDoubleValue("intervalValue5"));
                    fanPointVarCorrelation.setIntervalValue4(jsonObject.getDoubleValue("intervalValue4"));
                    fanPointVarCorrelation.setIntervalValue3(jsonObject.getDoubleValue("intervalValue3"));
                    fanPointVarCorrelation.setIntervalValue2(jsonObject.getDoubleValue("intervalValue2"));
                    fanPointVarCorrelation.setIntervalValue1(jsonObject.getDoubleValue("intervalValue1"));
                    log.info("------------------------------------------光伏相关性：：计算成功，待更新表数据----------------------------------------");
                } else {
                    fanPointVarCorrelation.setIntervalValue5(0.0);
                    fanPointVarCorrelation.setIntervalValue4(0.0);
                    fanPointVarCorrelation.setIntervalValue3(0.0);
                    fanPointVarCorrelation.setIntervalValue2(0.0);
                    fanPointVarCorrelation.setIntervalValue1(0.0);
                }
                fanPointVarCorrelation.setRecDate(new Date());
                idxBizFanPointProcessVariableClassificationService.saveOrUpdate(fanPointVarCorrelation);
                log.info("表数据已更新");
            }
        } catch (Exception e) {
            log.error("kafka失败，当前失败的批次: data:{}, {}", consumerRecord, e);
        } finally {
            redisUtils.expire(kafkaTopicConsumerGKHFFan, 600);
        }
        return true;
    }


    /**
     * 光伏处理消息 - 弃用
     * @param consumerRecord
     * @return
     */
    boolean consumerRecordsPv(ConsumerRecord<String, String> consumerRecord) {

        String startTime = DateUtils.convertDateToString(DateUtil.offsetMonth(new Date(), -lastMonthNum), DateUtils.DATE_TIME_PATTERN);
        String endTime = DateUtils.convertDateToString(DateUtils.getCurrentDayEndTime(new Date()), DateUtils.DATE_TIME_PATTERN);
        try {
            Optional<?> kafkaMessage = Optional.ofNullable(consumerRecord.value());
            if (kafkaMessage.isPresent()) {
                IdxBizPvPointVarCorrelation pvPointVarCorrelation = JSON.parseObject(kafkaMessage.get().toString(), IdxBizPvPointVarCorrelation.class);
                List<IndicatorData> tdengineData1 = indicatorDataMapper.selectDataByAddressAndtimeNew(pvPointVarCorrelation.getAnalysisIndexAddress().toString(), startTime, endTime, pvPointVarCorrelation.getAnalysisGatewayId(), pvPointVarCorrelation.getProcessGatewayId(), pvPointVarCorrelation.getProcessIndexAddress());
                List<Double> data1 = new ArrayList<>();
                List<Double> data2 = new ArrayList<>();
                tdengineData1.forEach(item -> {
                    if (item.getAddress().equals(pvPointVarCorrelation.getAnalysisIndexAddress()) && item.getGatewayId().equals(pvPointVarCorrelation.getAnalysisGatewayId())) {
                        data1.add(Double.parseDouble(item.getValue()));
                    } else {
                        data2.add(Double.parseDouble(item.getValue()));
                    }
                });
                if (data1.size() < data2.size()) {
                    Integer a = data2.size() - data1.size();
                    for (int i = 0; i < a; i++) {
                        data2.remove(0);
                    }
                } else if (data2.size() < data1.size()) {
                    Integer a = data1.size() - data2.size();
                    for (int i = 0; i < a; i++) {
                        data1.remove(0);
                    }
                }
                HashMap<String, Object> resultMap = new HashMap<>();
                resultMap.put("data1", data1);
                resultMap.put("data2", data2);
                String response = HttpUtil.createPost(baseUrlXGX).body(JSON.toJSONString(resultMap)).execute().body();
                if (response.contains("correlation") && !response.contains("warning")) {
                    com.alibaba.fastjson.JSONObject jsonObject = JSON.parseObject(response);
                    pvPointVarCorrelation.setCorrelationCoefficient(jsonObject.getDoubleValue("correlation"));
                    log.info("------------------------------------------光伏相关性：：计算成功，待更新表数据----------------------------------------");
                } else {
                    pvPointVarCorrelation.setCorrelationCoefficient(0.0);
                }
                pvPointVarCorrelation.setRecDate(new Date());

                idxBizPvPointVarCorrelationService.saveOrUpdate(pvPointVarCorrelation);
                log.info("表数据已更新");
                log.info("kafka消费zhTestGroup消息{}", consumerRecord);
            }
        } catch (Exception e) {
            log.error("kafka失败，当前失败的批次: data:{}, {}", consumerRecord, e);
        } finally {
            redisUtils.expire(kafkaTopicConsumerPv, 600);
        }
        return true;
    }


    /**
     * 光伏   工况划分 处理 - 弃用
     * @param consumerRecord
     * @return
     */
    boolean consumerRecordsGKFXPv(ConsumerRecord<String, String> consumerRecord) {
        try {String startTime = DateUtils.convertDateToString(DateUtil.offsetMonth(new Date(), -lastMonthNum), DateUtils.DATE_TIME_PATTERN);
            String endTime = DateUtils.convertDateToString(DateUtils.getCurrentDayEndTime(new Date()), DateUtils.DATE_TIME_PATTERN);

            Optional<?> kafkaMessage = Optional.ofNullable(consumerRecord.value());
            if (kafkaMessage.isPresent()) {
                IdxBizPvPointProcessVariableClassification pvPointVarCorrelation = JSON.parseObject(kafkaMessage.get().toString(), IdxBizPvPointProcessVariableClassification.class);
                List<IndicatorData> tdengineData1 = indicatorDataMapper.selectDataByAddressAndtime(pvPointVarCorrelation.getIndexAddress(), startTime, endTime, pvPointVarCorrelation.getGatewayId());

                HashMap<String, Object> resultMap = new HashMap<>();
                resultMap.put("processVariable", tdengineData1.stream().map(t -> Double.parseDouble(t.getValue())).collect(Collectors.toList()));
                resultMap.put("processVariableId", pvPointVarCorrelation.getSequenceNbr());
                String response = HttpUtil.createPost(baseUrlGKHF).body(JSON.toJSONString(resultMap)).execute().body();

                if (response.contains("intervalValue1") && response.contains("processVariableId")) {
                    com.alibaba.fastjson.JSONObject jsonObject = JSON.parseObject(response);

                    pvPointVarCorrelation.setIntervalValue5(jsonObject.getDoubleValue("intervalValue5"));
                    pvPointVarCorrelation.setIntervalValue4(jsonObject.getDoubleValue("intervalValue4"));
                    pvPointVarCorrelation.setIntervalValue3(jsonObject.getDoubleValue("intervalValue3"));
                    pvPointVarCorrelation.setIntervalValue2(jsonObject.getDoubleValue("intervalValue2"));
                    pvPointVarCorrelation.setIntervalValue1(jsonObject.getDoubleValue("intervalValue1"));
                    log.info("------------------------------------------光伏相关性：：计算成功，待更新表数据----------------------------------------");
                } else {
                    pvPointVarCorrelation.setIntervalValue5(0.0);
                    pvPointVarCorrelation.setIntervalValue4(0.0);
                    pvPointVarCorrelation.setIntervalValue3(0.0);
                    pvPointVarCorrelation.setIntervalValue2(0.0);
                    pvPointVarCorrelation.setIntervalValue1(0.0);
                }
                pvPointVarCorrelation.setRecDate(new Date());
                idxBizPvPointProcessVariableClassificationService.saveOrUpdate(pvPointVarCorrelation);
                log.info("表数据已更新");
            }
        } catch (Exception e) {
            log.error("kafka失败，当前失败的批次: data:{}, {}", consumerRecord, e);
        } finally {
            redisUtils.expire(kafkaTopicConsumerGKHFPv, 600);
        }
        return true;
    }


}


