| | |
|---|
| | | public void listen(ConsumerRecord<String, String> record, Consumer consumer) throws Exception { |
|---|
| | | String messageStr = record.value(); |
|---|
| | | Map<String, Object> message = (Map<String, Object>) JSON.parse(messageStr); |
|---|
| | | //long serverStartTime = System.currentTimeMillis(); |
|---|
| | | CopyOnWriteArraySet<SingleDeviceServer> sockets = SingleDeviceServer.sockets; |
|---|
| | | for (SingleDeviceServer socket : sockets) { |
|---|
| | | //判断消息是否数据该socket |
|---|
| | |
|---|
| | | Map<String, Object> adjustFormula = socket.getAdjustFormula(); |
|---|
| | | Map<String, Object> regionAqi = socket.getRegionAqi(); |
|---|
| | | //补偿数据 |
|---|
| | | if(adjustFormula!=null&®ionAqi!=null) |
|---|
| | | if(adjustFormula!=null) |
|---|
| | | message = AdjustDataUtils.adjust(message,adjustFormula,regionAqi); |
|---|
| | | //创建最终消息对象 |
|---|
| | | Map<String, Object> resultMessgae = new HashMap<>(); |
|---|
| | |
|---|
| | | resultMessgae.put(sensor.getCode(), sourceData); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | //测试时间延迟使用的属性 |
|---|
| | | //resultMessgae.put("DataTime", message.get("DataTime")); |
|---|
| | | //resultMessgae.put("time", message.get("time")); |
|---|
| | | //resultMessgae.put("time1", message.get("time1")); |
|---|
| | | //resultMessgae.put("serverTime", System.currentTimeMillis()); |
|---|
| | | //resultMessgae.put("serverStartTime", serverStartTime); |
|---|
| | | socket.sendMessage(JSON.toJSONString(resultMessgae)); |
|---|
| | | } |
|---|
| | | } |
|---|
