blame | history | patch | commit | commitdiff | ignore whitespace
jinpengyong
2023-08-25 5242d687d4edcd6c0f7e10fb82b0e5505dbb3da3 
 screen-manage/src/main/java/com/moral/api/kafka/consumer/DeviceConsumer.java


@@ -1,3 +1,4 @@


/*


package com.moral.api.kafka.consumer;




import com.moral.api.service.*;


@@ -10,6 +11,9 @@


import org.springframework.stereotype.Component;


import org.springframework.util.ObjectUtils;




import java.text.DecimalFormat;


import java.text.SimpleDateFormat;


import java.util.Date;


import java.util.HashMap;


import java.util.Iterator;


import java.util.Map;


@@ -18,9 +22,8 @@


import com.moral.constant.KafkaConstants;


import com.moral.constant.RedisConstants;




/*


 * 普通设备消费者


 * */






@Component


@Slf4j


public class DeviceConsumer {


@@ -36,9 +39,6 @@




    @Autowired


    private RedisTemplate redisTemplate;




    @Autowired


    private HistorySecondSpecialService historySecondSpecialService;




    @Autowired


    private HistorySecondCruiserService historySecondCruiserService;


@@ -61,7 +61,7 @@


            }




            //数据过滤


            data.remove("time");


//            data.remove("time");


            data.remove("entryTime");


            Iterator<Map.Entry<String, Object>> iterator = data.entrySet().iterator();


            Map<String, Object> newMap = new HashMap<>();


@@ -102,6 +102,17 @@


            //数据过滤


            data.remove("time");


            data.remove("entryTime");


            SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");


            String format = dateFormat.format(new Date().getTime());


            //获取年份


            String substring = format.substring(0, 4);




            String substring1 = time.toString().substring(0, 4);




            String replace = time.toString().replace(substring1, substring);




            data.put("DataTime",replace);




            Iterator<Map.Entry<String, Object>> iterator = data.entrySet().iterator();


            Map<String, Object> newMap = new HashMap<>();


            Map.Entry<String, Object> next;


@@ -126,7 +137,7 @@




    //秒数据，修改设备状态，缓存最新秒数据


    @KafkaListener(topics = KafkaConstants.TOPIC_SECOND, containerFactory = "stateListenerContainerFactory")


    public void listenSecond(ConsumerRecord<String, String> record) {


    public void listenSecond(ConsumerRecord<String, String> record, Acknowledgment ack) {


        String msg = record.value();


        try {


            Map<String, Object> data = JSON.parseObject(msg, Map.class);


@@ -141,13 +152,14 @@


            data.remove("entryTime");




            //数据校准


            data = deviceService.adjustDeviceData(data);


            data = deviceService.adjustDeviceData(data,"0");


            //存入redis


            data.put("DataTime", time);


            redisTemplate.opsForHash().put(RedisConstants.DATA_SECOND, mac, data);


            //判断并修改设备状态


            data.put("mac", mac);


            deviceService.judgeDeviceState(data);


            ack.acknowledge();


        } catch (Exception e) {


            log.error("param{}" + msg);


        }


@@ -157,7 +169,6 @@


    @KafkaListener(topics = KafkaConstants.UAV_TOPIC_SECOND, groupId = KafkaConstants.GROUP_INSERT, containerFactory = "insertListenerContainerFactory")


    public void listenSecondSpecial(ConsumerRecord<String, String> record, Acknowledgment ack) {


        String msg = record.value();


        System.out.println(msg);


        try {


            Map<String, Object> data = JSON.parseObject(msg, Map.class);


            Object mac = data.get("mac");


@@ -180,7 +191,7 @@


    }




    //走航车秒数据


    @KafkaListener(topics = KafkaConstants.CRUISER_TOPIC_SECOND, groupId = KafkaConstants.GROUP_INSERT, containerFactory = "insertListenerContainerFactory")


    @KafkaListener(topics = KafkaConstants.CRUISER_TOPIC_SECOND, containerFactory = "insertListenerContainerFactory")


    public void listenSecondCruiser(ConsumerRecord<String, String> record, Acknowledgment ack) {


        String msg = record.value();


        try {


@@ -200,7 +211,8 @@


            historySecondCruiserService.insertHistorySecond(data);


            ack.acknowledge();


        } catch (Exception e) {


            log.error("param{}" + msg);


            log.error("param{}" + e);


        }


    }


}


*/