调整KFKA数据推送测试。

jiazx0107@163.com

2024-01-01 d454e51e06dbf7beb50551588fc58a5810295d7c

调整KFKA数据推送测试。

 src/main/java/com/fzzy/async/fzzy40/repository/Fzzy40Sync1302Rep.java

@@ -35,6 +35,6 @@
     * @param end
     * @return
     */
    @Query("from Fz40Grain where depotId =:depotId and receiveDate >=:start and receiveDate <:end order by receiveDate ")
    @Query("from Fz40Grain where depotId =:depotId and receiveDate >=:start and receiveDate <:end order by receiveDate desc ")
    List<Fz40Grain> findByReceiveDate(@Param("depotId") String depotId, @Param("start") Date start, @Param("end") Date end);
}

 src/main/java/com/fzzy/gateway/controller/GatewayDataApi.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,397 @@
package com.fzzy.gateway.controller;

import com.fzzy.api.data.GatewayDeviceProtocol;
import com.fzzy.api.data.GatewayDeviceType;
import com.fzzy.gateway.GatewayUtils;
import com.fzzy.gateway.api.GatewayRemoteManager;
import com.fzzy.gateway.api.GatewaySyncLprService;
import com.fzzy.gateway.data.BaseReqData;
import com.fzzy.gateway.data.BaseResp;
import com.fzzy.gateway.entity.GateWayParam;
import com.fzzy.gateway.entity.GatewayDevice;
import com.fzzy.gateway.service.GatewayDeviceService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import javax.annotation.Resource;
import java.util.List;
import java.util.stream.Collectors;


/**
 * ä¸Žç½‘关相关的接口数据对接，由第三方对接
 */
@Slf4j
@Controller
@RequestMapping("/gateway/api/v1")
public class GatewayDataApi {

    @Resource
    private GatewayDeviceService gatewayDeviceService;

    @Resource
    private GatewayRemoteManager gatewayRemoteManager;


    /**
     * æµ‹è¯•å…¥å£
     *
     * @param param
     * @return
     */
    @PostMapping("/test/deviceTest")
    public @ResponseBody
    String deviceTest(@RequestBody GateWayParam param) throws Exception {

        if (StringUtils.isEmpty(param.getBizType())) {
            return "ERROR:没有获取到业务类型，执行失败";
        }

        if ("testGrain".equals(param.getBizType())) {
            return testGrain(param);
        }

        if ("ajaxTestKafkaGrain".equals(param.getBizType())) {
            return ajaxTestKafkaGrain(param);
        }

        if ("ajaxTestWeight".equals(param.getBizType())) {
            return ajaxTestWeight(param);
        }

        if ("ajaxTestLpr".equals(param.getBizType())) {
            return ajaxTestLpr(param);
        }

        return "SUCCESS";
    }


    /**
     * ä¸‹è¡Œåˆ†æœºæµ‹è¯•å…¥å£
     *
     * @param param
     * @return
     */
    @PostMapping("/test/IDE")
    public @ResponseBody
    String testIDE(@RequestBody GateWayParam param) throws Exception {

        if (StringUtils.isEmpty(param.getBizType())) {
            return "ERROR:没有获取到业务类型，执行失败";
        }

        //获取分机配置
        if ("syncConf".equals(param.getBizType())) {
            return syncConf(param);
        }

        if ("writeConf".equals(param.getBizType())) {
            return writeConf(param);
        }

        if ("initParam".equals(param.getBizType())) {
            return initCable(param);
        }

        if ("disconnect".equals(param.getBizType())) {
            return disconnect(param);
        }

        if ("transparent".equals(param.getBizType())) {
            return transparent(param);
        }

        return "SUCCESS";
    }


    private String transparent(GateWayParam param) {

        String deviceId = param.getDeviceId();

        GatewayDevice device = GatewayUtils.getCacheByDeviceId(deviceId);

        BaseReqData reqData = new BaseReqData();
        reqData.setDeviceId(device.getDeviceId());
        reqData.setProductId(device.getProductId());
        reqData.setDeviceName(device.getDeviceName());
        reqData.setDevice(device);
        reqData.setAutoReplay(false);

        BaseResp resp = gatewayRemoteManager.getSyncGrainService(device.getSyncProtocol()).transparent(reqData);
        if (BaseResp.CODE_200 != resp.getCode()) {
            return "ERROR：" + resp.getMsg();
        }
        return "SUCCESS";
    }

    private String disconnect(GateWayParam param) {
        String deviceId = param.getDeviceId();

        GatewayDevice device = GatewayUtils.getCacheByDeviceId(deviceId);

        BaseReqData reqData = new BaseReqData();
        reqData.setDeviceId(device.getDeviceId());
        reqData.setProductId(device.getProductId());
        reqData.setDeviceName(device.getDeviceName());
        reqData.setDevice(device);
        reqData.setAutoReplay(false);

        BaseResp resp = gatewayRemoteManager.getSyncGrainService(device.getSyncProtocol()).disconnect(reqData);
        if (BaseResp.CODE_200 != resp.getCode()) {
            return "ERROR：" + resp.getMsg();
        }
        return "SUCCESS";
    }

    private String initCable(GateWayParam param) {
        String deviceId = param.getDeviceId();

        GatewayDevice device = GatewayUtils.getCacheByDeviceId(deviceId);

        BaseReqData reqData = new BaseReqData();
        reqData.setDeviceId(device.getDeviceId());
        reqData.setProductId(device.getProductId());
        reqData.setDeviceName(device.getDeviceName());
        reqData.setDevice(device);
        reqData.setAutoReplay(false);

        BaseResp resp = gatewayRemoteManager.getSyncGrainService(device.getSyncProtocol()).initCable(reqData);
        if (BaseResp.CODE_200 != resp.getCode()) {
            return "ERROR：" + resp.getMsg();
        }
        return "SUCCESS";
    }

    private String syncConf(GateWayParam param) {

        String deviceId = param.getDeviceId();

        GatewayDevice device = GatewayUtils.getCacheByDeviceId(deviceId);

        BaseReqData reqData = new BaseReqData();
        reqData.setDeviceId(device.getDeviceId());
        reqData.setProductId(device.getProductId());
        reqData.setDeviceName(device.getDeviceName());
        reqData.setDevice(device);
        reqData.setAutoReplay(false);

        BaseResp resp = gatewayRemoteManager.getSyncGrainService(device.getSyncProtocol()).syncConf(reqData);
        if (BaseResp.CODE_200 != resp.getCode()) {
            return "ERROR：" + resp.getMsg();
        }
        return "SUCCESS";
    }

    private String writeConf(GateWayParam param) {
        String deviceId = param.getDeviceId();

        GatewayDevice device = GatewayUtils.getCacheByDeviceId(deviceId);

        BaseReqData reqData = new BaseReqData();
        reqData.setDeviceId(device.getDeviceId());
        reqData.setProductId(device.getProductId());
        reqData.setDeviceName(device.getDeviceName());
        reqData.setDevice(device);
        reqData.setAutoReplay(false);

        BaseResp resp = gatewayRemoteManager.getSyncGrainService(device.getSyncProtocol()).writeConf(reqData);
        if (BaseResp.CODE_200 != resp.getCode()) {
            return "ERROR：" + resp.getMsg();
        }
        return "SUCCESS";
    }

    /**
     * åˆå§‹åŒ–车牌识别
     *
     * @param param
     * @return
     */
    @PostMapping("/control/init-lpr")
    public @ResponseBody
    String initLpr(@RequestBody GateWayParam param) throws Exception {

        List<GatewayDevice> list = gatewayDeviceService.listAll();

        if (null == list || list.isEmpty()) {
            return "ERROR:没有获取到设备信息";
        }

        BaseReqData reqData;
        int i = 0;
        GatewaySyncLprService syncLprService;
        for (GatewayDevice device : list) {

            if (!GatewayDeviceType.TYPE_02.getCode().equals(device.getType())) {
                continue;
            }
            reqData = new BaseReqData(device);
            reqData.setIndex(i);
            syncLprService = gatewayRemoteManager.getSyncLprService(device.getSyncProtocol());
            if (null == syncLprService) continue;
            syncLprService.initLpr(reqData);
            i++;
        }

        return "SUCCESS";
    }

    private String testGrain(GateWayParam param) {
        String deviceId = param.getDeviceId();

        GatewayDevice device = GatewayUtils.getCacheByDeviceId(deviceId);

        BaseReqData reqData = new BaseReqData();
        reqData.setDeviceId(device.getDeviceId());
        reqData.setProductId(device.getProductId());
        reqData.setDeviceName(device.getDeviceName());
        reqData.setDevice(device);
        reqData.setAutoReplay(true);

        if (!GatewayDeviceType.TYPE_07.getCode().equals(device.getType())) {
            return "ERROR：当前设备非粮情设备不支持当前操作";
        }

        if (StringUtils.isEmpty(device.getCableRule())) {
            return "ERROR：当前设备没有配置布线规则，无法执行";
        }

        BaseResp resp;
        if (GatewayDeviceProtocol.DEVICE_TEST.getCode().equals(device.getSyncProtocol())) {
            resp = gatewayRemoteManager.getGatewayTestService(device.getPushProtocol()).testGrain(reqData);
        } else {
            reqData.setAutoReplay(false);
            resp = gatewayRemoteManager.getSyncGrainService(device.getSyncProtocol()).syncGrain(reqData);
        }

        //自动推送
        if (BaseResp.CODE_200 == resp.getCode() && reqData.isAutoReplay()) {
            reqData.setData(resp.getData());
            gatewayRemoteManager.getDeviceReportService(device.getPushProtocol()).reportGrainData(reqData);
        }

        return "SUCCESS";
    }


    /**
     * ç²®æƒ…推送测试KAFKA方式
     *
     * @param param
     * @return
     */
    public String ajaxTestKafkaGrain(GateWayParam param) throws Exception {
        List<GatewayDevice> list = gatewayDeviceService.listAll();
        if (null == list || list.isEmpty()) {
            return "ERROR：为获取到系统中设备配置，取消执行";
        }

        BaseReqData reqData;
        BaseResp resp;
        for (GatewayDevice device : list) {
            reqData = new BaseReqData();
            reqData.setDeviceId(param.getDeviceId());
            reqData.setDayTime(param.getDayTime());
            reqData.setDevice(device);

            resp = gatewayRemoteManager.getGatewayTestService(device.getPushProtocol()).testGrainKafka(reqData);

            //自动推送
            if (BaseResp.CODE_200 == resp.getCode()) {
                reqData.setData(resp.getData());
                gatewayRemoteManager.getDeviceReportService(device.getPushProtocol()).reportGrainDataByKafka(reqData);
            }
        }

        return "SUCCESS";
    }

    /**
     * åœ°ç£…推送测试
     *
     * @param param
     * @return
     */
    public String ajaxTestWeight(GateWayParam param) throws Exception {

        double weight = param.getWeight();

        List<GatewayDevice> list = gatewayDeviceService.listAll();
        if (list == null || list.size() <= 0) {
            return "ERROR：没有配置设备信息，执行失败";
        }

        List<GatewayDevice> weights = list.stream().filter(s -> (GatewayDeviceType.TYPE_01.getCode().equals(s.getType()))).collect(Collectors.toList());
        if (weights == null || weights.size() <= 0) {
            return "ERROR：ERROR：没有配置设备信息，执行失败";
        }

        BaseReqData reqData;
        BaseResp resp;
        for (GatewayDevice device : weights) {
            reqData = new BaseReqData();
            reqData.setDeviceId(device.getDeviceId());
            reqData.setProductId(device.getProductId());
            reqData.setDeviceName(device.getDeviceName());
            reqData.setDevice(device);
            reqData.setAutoReplay(true);
            reqData.setWeight(weight);
            resp = gatewayRemoteManager.getGatewayTestService(device.getPushProtocol()).testWeight(reqData);

            //自动推送
            if (BaseResp.CODE_200 == resp.getCode() && reqData.isAutoReplay()) {
                reqData.setData(resp.getData());
                gatewayRemoteManager.getDeviceReportService(device.getPushProtocol()).reportWeightData(reqData);
            }
        }
        return "SUCCESS";
    }

    /**
     * åœ°ç£…推送测试
     *
     * @return
     */
    public String ajaxTestLpr(GateWayParam param) throws Exception {

        String carNumber = param.getCarNumber();

        List<GatewayDevice> list = gatewayDeviceService.listAll();
        if (list == null || list.size() <= 0) {
            log.error("ERROR：没有配置设备信息，执行失败");
            return "ERROR：没有配置设备信息，执行失败";
        }
        List<GatewayDevice> weights = list.stream().filter(s -> (GatewayDeviceType.TYPE_02.getCode().equals(s.getType()))).collect(Collectors.toList());
        if (weights == null || weights.size() <= 0) {
            log.error("ERROR：没有配置设备信息，执行失败");
            return "ERROR：没有配置设备信息，执行失败";
        }

        BaseReqData reqData;
        BaseResp resp;
        for (GatewayDevice device : weights) {
            reqData = new BaseReqData();
            reqData.setDeviceId(device.getDeviceId());
            reqData.setProductId(device.getProductId());
            reqData.setDeviceName(device.getDeviceName());
            reqData.setDevice(device);
            reqData.setAutoReplay(true);
            reqData.setCarNumber(carNumber);
            resp = gatewayRemoteManager.getGatewayTestService(device.getPushProtocol()).testLpr(reqData);

            //自动推送
            if (BaseResp.CODE_200 == resp.getCode() && reqData.isAutoReplay()) {
                reqData.setData(resp.getData());
                gatewayRemoteManager.getDeviceReportService(device.getPushProtocol()).reportLprData(reqData);
            }
        }
        return "SUCCESS";
    }
}

 src/main/java/com/fzzy/gateway/controller/GatewayDeviceController.java

@@ -296,6 +296,11 @@
        BaseReqData reqData;
        BaseResp resp;
        for (GatewayDevice device : list) {

            //如果不是粮情分机跳过
            if (!GatewayDeviceType.TYPE_07.getCode().equals(device.getType())) {
                continue;
            }
            reqData = new BaseReqData();
            reqData.setDeviceId(param.getDeviceId());
            reqData.setDayTime(param.getDayTime());

 src/main/java/com/fzzy/gateway/hx2023/data/KafaGrainData.java

@@ -45,7 +45,5 @@
    //模块识别码-常量，固定传 apiTemperature
    private String apISource = "apiTemperature";


    private JSONObject params;

}

 src/main/java/com/fzzy/gateway/hx2023/data/TRHInfo.java

@@ -9,7 +9,5 @@
public class TRHInfo {

    private List<KafkaGrainDataDetail> temperature;


    private List<KafkaGrainTH> temperatureAndhumidity;
}

 src/main/java/com/fzzy/gateway/hx2023/kafka/KafkaDeviceReportService.java

@@ -22,6 +22,9 @@


    public String publishWithTopic(String sendData, String topic) {



        //推送信息
        kafkaTemplate.send(topic, sendData).addCallback(success -> {
            // æ¶ˆæ¯å‘送到的topic

 src/main/java/com/fzzy/gateway/hx2023/service/DeviceReportServiceImpl.java

@@ -44,7 +44,7 @@
        topic = topic.replace("${productId}", reqData.getProductId()).replace("${deviceId}", reqData.getDeviceId());

        //如果是测试模式不执行推送
        if(configData.getActive().indexOf("dev")>=0){
        if (configData.getActive().indexOf("dev") >= 0) {

            log.info("----------------------------推送MQTT粮情信息，注：调试模式不推送---------------------------");
            log.info("-----TOPIC-----{}", topic);
@@ -146,6 +146,16 @@
    @Override
    public BaseResp reportGrainDataByKafka(BaseReqData reqData) {
        String topic = ScConstant.TOPIC_MESSAGE_REPORT;
        //如果是测试模式不执行推送
        if (configData.getActive().indexOf("dev") >= 0) {

            log.info("----------------------------推送KAFKA粮情信息，注：调试模式不推送---------------------------");
            log.info("-----TOPIC-----{}", topic);
            log.info("-----Message-----{}", reqData.getData());

            return new BaseResp();
        }

        kafkaDeviceReportService.publishWithTopic(reqData.getData(), topic);
        return new BaseResp();
    }

 src/main/java/com/fzzy/gateway/hx2023/service/ScGatewayTestServiceImpl.java

@@ -22,9 +22,7 @@
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.*;

@Slf4j
@Component
@@ -63,8 +61,6 @@
    }

    private BaseResp pushByV40(BaseReqData reqData, Date start, Date end) {


        GatewayDevice device = reqData.getDevice();
        String depotIdSys = device.getDepotIdSys();

@@ -75,12 +71,12 @@

        List<Fz40Grain> listGrain = fzzy40CommonService.listGrain(depotIdSys, start, end);
        if (null == listGrain || listGrain.isEmpty()) {
            log.error("---------设备---{}--未同步到粮情信息，请确认当前条件下是否有数据", device.getDeviceName());
            log.error("---------设备-{}-系统仓库编码-{}-未同步到粮情信息，请确认当前条件下是否有数据", device.getDeviceName(), device.getDepotIdSys());
            return new BaseResp(BaseResp.CODE_50, "未同步到粮情信息，请确认当前条件下是否有数据");
        }

        //获取最后一条粮情作为当前数据
        Fz40Grain lastData = listGrain.get(listGrain.size() - 1);
        //获取第一条数据执行推送
        Fz40Grain lastData = listGrain.get(0);
        KafaGrainData pushData = this.lastData2PushData(lastData, device);

        return new BaseResp(JSONObject.toJSONString(pushData));
@@ -89,26 +85,24 @@
    /**
     * å°†ç²®æƒ…数据转换为推送需要的数据格式
     *
     * @param lastData
     * @param grainData
     * @return
     */
    private KafaGrainData lastData2PushData(Fz40Grain lastData, GatewayDevice device) {
    private KafaGrainData lastData2PushData(Fz40Grain grainData, GatewayDevice device) {
        KafaGrainData result = new KafaGrainData();

        result.setMessageId(ScConstant.getMessageId());
        result.setMessgeId(result.getMessageId());
        result.setDeviceID(device.getDeviceId());
        result.setAvgTemperature(lastData.getTempAve() + "");
        result.setMinTemperature(lastData.getTempMin() + "");
        result.setMaxTemperature(lastData.getTempMax() + "");
        result.setCollectTime(DateFormatUtils.format(lastData.getReceiveDate(), "yyyy-MM-dd HH:mm:ss"));
        result.setAvgTemperature(grainData.getTempAve() + "");
        result.setMinTemperature(grainData.getTempMin() + "");
        result.setMaxTemperature(grainData.getTempMax() + "");
        result.setCollectTime(DateFormatUtils.format(grainData.getReceiveDate(), "yyyy-MM-dd HH:mm:ss"));

        //层-行-列
        String[] attrCable = lastData.getCable().split("-");


        if (StringUtils.isNotEmpty(lastData.getCableCir())) {
            return lastData2PushData2(lastData, device);
        String[] attrCable = grainData.getCable().split("-");
        if (StringUtils.isNotEmpty(grainData.getCableCir())) {
            return lastData2PushData2(grainData, device);
        }

        //层行列
@@ -117,53 +111,51 @@
        int cableX = Integer.valueOf(attrCable[2]);

        //温度集合
        String[] attr = lastData.getPoints().split(",");
        String[] attr = grainData.getPoints().split(",");

        //根号
        int cableNum = 1, position = 0;

        String curTemp;
        List<KafkaGrainDataDetail1> temperature = new ArrayList<>();

        int x = 0, y = 0, z = 0;
        for (int i = 0; i < attr.length; i++) {

            position = i;

            z = i % cableZ + 1;
            x = i / (cableZ * cableY);
            y = x * (cableZ * cableY);
            y = (i - y) / cableZ;

            // å€’转Xè½´
            x = cableX - 1 - x;

            //根号
            cableNum = (i / cableZ) + 1;

            curTemp = attr[i];


            //判断最大
            if (curTemp.equals(result.getMaxTemperature())) {
                result.setMaxX(x + "");
                result.setMaxY(y + "");
                result.setMaxZ(position + "");
            }
            //如果是异常值,执行调整数据 TODO
            if (Double.valueOf(curTemp) < -99.9) {

                curTemp = grainData.getTempAve() + "";

            //判断最小
            if (curTemp.equals(result.getMinTemperature())) {
                result.setMinX(x + "");
                result.setMinY(y + "");
                result.setMinZ(position + "");
            } else {
                //判断最大
                if (curTemp.equals(result.getMaxTemperature())) {
                    result.setMaxX(x + "");
                    result.setMaxY(y + "");
                    result.setMaxZ(position + "");
                }
                //判断最小
                if (curTemp.equals(result.getMinTemperature())) {
                    result.setMinX(x + "");
                    result.setMinY(y + "");
                    result.setMinZ(position + "");
                }
            }

            temperature.add(new KafkaGrainDataDetail1(cableNum + "", z + "", curTemp, position + "", x + "", y + ""));
        }

        //粮温信息

        JSONObject trhInfo = new JSONObject();
        // TRHInfo trhInfo = new TRHInfo();
        trhInfo.put("temperature", temperature);
@@ -171,10 +163,10 @@

        //仓温度信息
        KafkaGrainTH grainTH = new KafkaGrainTH();
        grainTH.setHumidity(lastData.getHumidityIn() + "");
        grainTH.setTemperature(lastData.getTempIn() + "");
        grainTH.setAirHumidity(lastData.getHumidityOut() + "");
        grainTH.setAirTemperature(lastData.getTempOut() + "");
        grainTH.setHumidity(grainData.getHumidityIn() + "");
        grainTH.setTemperature(grainData.getTempIn() + "");
        grainTH.setAirHumidity(grainData.getHumidityOut() + "");
        grainTH.setAirTemperature(grainData.getTempOut() + "");

        List<KafkaGrainTH> temperatureAndhumidity = new ArrayList<>();
        temperatureAndhumidity.add(grainTH);
@@ -313,19 +305,19 @@
        for (int i = 0; i < sumNum; i++) {
            log.info("i=：" + i);
            if (i % cableZ == 0) {
                randomNumber = RandomUtils.nextDouble(tMIn, tMax+1.5);
                randomNumber = RandomUtils.nextDouble(tMIn, tMax + 1.5);
                log.info("第1层温度：" + randomNumber);
            } else if (i % cableZ == 1)  {
                randomNumber = RandomUtils.nextDouble(tMIn-1, tMax-0);
            } else if (i % cableZ == 1) {
                randomNumber = RandomUtils.nextDouble(tMIn - 1, tMax - 0);
                log.info("第2层温度：" + randomNumber);
            } else if (i % cableZ == 2)  {
                randomNumber = RandomUtils.nextDouble(tMIn-2, tMax-1.5);
            } else if (i % cableZ == 2) {
                randomNumber = RandomUtils.nextDouble(tMIn - 2, tMax - 1.5);
                log.info("第3层温度：" + randomNumber);
            } else if (i % cableZ == 3)  {
                randomNumber = RandomUtils.nextDouble(tMIn-3, tMax-3);
            } else if (i % cableZ == 3) {
                randomNumber = RandomUtils.nextDouble(tMIn - 3, tMax - 3);
                log.info("第4层温度：" + randomNumber);
            } else if (i % cableZ == 4)  {
                randomNumber = RandomUtils.nextDouble(tMIn-4, tMax-4.5);
            } else if (i % cableZ == 4) {
                randomNumber = RandomUtils.nextDouble(tMIn - 4, tMax - 4.5);
                log.info("第5层温度：" + randomNumber);
            }
            // randomNumber = Math.random() * (tMax - tMIn + 1) + tMIn;
@@ -345,7 +337,7 @@
        outPut.setTemperature(temperature);
        List<GrainTH> ths = new ArrayList<>();

        ths.add(new GrainTH(weather.getTem()!=null?weather.getTem()+"":"",weather.getHumidity()!=null?weather.getHumidity()+"":"","1"));
        ths.add(new GrainTH(weather.getTem() != null ? weather.getTem() + "" : "", weather.getHumidity() != null ? weather.getHumidity() + "" : "", "1"));
        outPut.setTemperatureAndhumidity(ths);

        grain.setOutput(JSONObject.toJSONString(outPut));