blame | history | patch | commit | commitdiff | show whitespace
jinpengyong
2023-10-13 bb6856a82a588f05040b0410a9747ae8029bb3be 
 screen-manage/src/main/java/com/moral/api/kafka/consumer/DeviceConsumer.java


@@ -5,24 +5,25 @@


import com.moral.constant.KafkaConstants;


import com.moral.constant.RedisConstants;


import lombok.extern.slf4j.Slf4j;


import org.apache.kafka.clients.consumer.CommitFailedException;


import org.apache.kafka.clients.consumer.ConsumerRecord;


import org.springframework.beans.factory.annotation.Autowired;


import org.springframework.beans.factory.annotation.Value;


import org.springframework.data.redis.core.RedisTemplate;


import org.springframework.kafka.annotation.KafkaListener;


import org.springframework.kafka.support.Acknowledgment;


import org.springframework.stereotype.Component;


import org.springframework.util.ObjectUtils;




import java.util.HashMap;


import java.util.Iterator;


import java.util.Map;




import java.util.*;






@Component


@Slf4j


public class DeviceConsumer {




    @Value("${result.device.list}")


    private String deviceList;


    @Autowired


    private HistoryMinutelyService historyMinutelyService;




@@ -143,28 +144,38 @@


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


        String msg = record.value();


        try {




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


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


            Object time = data.get("DataTime");


            if (ObjectUtils.isEmpty(time) || ObjectUtils.isEmpty(mac)) {


                log.warn("some properties is null, param{}", msg);


                ack.acknowledge();


                return;


            }


            List<String> listDictDevice = Arrays.asList(deviceList.split(","));


            if(!listDictDevice.contains(mac.toString())){


                ack.acknowledge();


            }


            //数据过滤


            data.remove("time");


            data.remove("entryTime");




            //数据校准


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


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


            //存入redis


            data.put("DataTime", time);


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


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


            //判断并修改设备状态


            data.put("mac", mac);


            deviceService.judgeDeviceState(data);


            ack.acknowledge();


            deviceService.judgeDeviceState(data);




        }catch (CommitFailedException e){


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


        } catch (Exception e) {


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


            ack.acknowledge();


        }


    }