| | |
| | | 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; |
| | |
| | | public void listenSecond(ConsumerRecord<String, String> record, Acknowledgment ack) { |
| | | String msg = record.value(); |
| | | try { |
| | | |
| | | ack.acknowledge(); |
| | | 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"); |
| | | //存入redis |
| | | data.put("DataTime", time); |
| | | redisTemplate.opsForHash().put(RedisConstants.DATA_SECOND, mac, data); |
| | | //判断并修改设备状态 |
| | | data.put("mac", mac); |
| | | deviceService.judgeDeviceState(data); |
| | | ack.acknowledge(); |
| | | List<String> listDictDevice = Arrays.asList(deviceList.split(",")); |
| | | if(listDictDevice.contains(mac.toString())){ |
| | | //判断并修改设备状态 |
| | | data.put("mac", mac); |
| | | deviceService.judgeDeviceState(data); |
| | | } |
| | | //数据校准 |
| | | //data = deviceService.adjustDeviceData(data,"0"); |
| | | //存入redis |
| | | //redisTemplate.opsForHash().put(RedisConstants.DATA_SECOND, mac, data); |
| | | }catch (CommitFailedException e){ |
| | | log.error("param{}" + msg); |
| | | } catch (Exception e) { |
| | | log.error("param{}" + msg); |
| | | ack.acknowledge(); |