|  |  |  | 
|---|
|  |  |  | 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(); | 
|---|