| | |
| | | import com.moral.api.entity.Sensor; |
| | | import com.moral.api.entity.UnitConversion; |
| | | import com.moral.api.websocket.SingleDeviceServer; |
| | | import com.moral.util.UnitConvertUtils; |
| | | import org.apache.kafka.clients.consumer.Consumer; |
| | | import org.apache.kafka.clients.consumer.ConsumerRecord; |
| | | import org.apache.kafka.common.TopicPartition; |
| | |
| | | @Component |
| | | public class SecondDataConsumer implements ConsumerSeekAware { |
| | | |
| | | @KafkaListener(topics = "second_data",groupId = "SecondsDataGroup3") |
| | | public void listen(ConsumerRecord<String, String> record , Consumer consumer) throws Exception { |
| | | @KafkaListener(topics = "second_data", groupId = "SecondsDataGroup3") |
| | | public void listen(ConsumerRecord<String, String> record, Consumer consumer) throws Exception { |
| | | String messageStr = record.value(); |
| | | Map<String,String> message = (Map<String,String>)JSON.parse(messageStr); |
| | | Map<String, String> message = (Map<String, String>) JSON.parse(messageStr); |
| | | CopyOnWriteArraySet<SingleDeviceServer> sockets = SingleDeviceServer.sockets; |
| | | for (SingleDeviceServer socket : sockets) { |
| | | //判断消息书否数据该socket |
| | |
| | | Map<String, Device> devicesInfo = socket.getDevicesInfo(); |
| | | Device device = devicesInfo.get(mac); |
| | | //创建最终消息对象 |
| | | Map<String,String> resultMessgae = new HashMap<>(); |
| | | Map<String, String> resultMessgae = new HashMap<>(); |
| | | //拼接单位 |
| | | List<Sensor> sensors = device.getVersion().getSensors();//获取型号所有因子信息 |
| | | for (Sensor sensor : sensors) { |
| | |
| | | String unitKey = sensor.getUnitKey(); |
| | | //数据补偿 |
| | | //单位转换 |
| | | if(!unitKey.equals(showUnitKey)){ |
| | | if (!unitKey.equals(showUnitKey)) { |
| | | String formula = sensor.getFormula(); |
| | | if(ObjectUtils.isEmpty(formula)){ |
| | | //如果sensor中的公式为空则从缓存中获取公式 |
| | | if (ObjectUtils.isEmpty(formula)) { |
| | | List<UnitConversion> unitConversions = socket.getUnitConversions(); |
| | | System.out.println(unitConversions); |
| | | }else{ |
| | | |
| | | for (UnitConversion unitConversion : unitConversions) { |
| | | if (unitConversion.getOriginalUnitKey().equals(unitKey) && unitConversion.getTargetUnitKey().equals(showUnitKey)) |
| | | formula = unitConversion.getFormula(); |
| | | } |
| | | } |
| | | //单位转换 |
| | | UnitConvertUtils.calculate(message.get(code),formula); |
| | | } |
| | | //拼接单位 |
| | | String showUnit = sensor.getShowUnit(); |
| | | String value = message.get(code); |
| | | value+=showUnit; |
| | | resultMessgae.put(sensor.getCode(),value); |
| | | value += showUnit; |
| | | resultMessgae.put(sensor.getCode(), value); |
| | | } |
| | | resultMessgae.put("DataTime",message.get("DataTime")); |
| | | resultMessgae.put("DataTime", message.get("DataTime")); |
| | | socket.sendMessage(JSON.toJSONString(resultMessgae)); |
| | | } |
| | | } |