| | |
|---|
| | | @KafkaListener(topics = "second_data", groupId = "SecondsDataGroup998") |
|---|
| | | 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, Object> message = (Map<String, Object>) 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, Object> resultMessgae = new HashMap<>(); |
|---|
| | | //拼接单位 |
|---|
| | | List<Sensor> sensors = device.getVersion().getSensors();//获取型号所有因子信息 |
|---|
| | | for (Sensor sensor : sensors) { |
|---|
| | |
|---|
| | | String showUnitKey = sensor.getShowUnitKey(); |
|---|
| | | String unitKey = sensor.getUnitKey(); |
|---|
| | | String unit = sensor.getUnit(); |
|---|
| | | String sourceData = message.get(code); |
|---|
| | | String sourceData = String.valueOf(message.get(code)); |
|---|
| | | //数据补偿 |
|---|
| | | //单位转换 |
|---|
| | | if (!unitKey.equals(showUnitKey)) {//如果源单位和显示单位不同，则进行单位转换 |
|---|
| | |
|---|
| | | } |
|---|
| | | } |
|---|
| | | //单位转换 |
|---|
| | | String resultData = UnitConvertUtils.calculate(message.get(code), formula); |
|---|
| | | String resultData = UnitConvertUtils.calculate(sourceData, formula); |
|---|
| | | if (resultData != null) { |
|---|
| | | resultData+=showUnit; |
|---|
| | | }else{//如果转换出的数据为null，则代表缓存中也没有公式，依然使用源单位。 |
|---|
| | |
|---|
| | | resultMessgae.put(sensor.getCode(), sourceData); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | resultMessgae.put("DataTime", message.get("DataTime")); |
|---|
| | | resultMessgae.put("serverTime",System.currentTimeMillis()); |
|---|
| | | socket.sendMessage(JSON.toJSONString(resultMessgae)); |
|---|
| | | } |
|---|
| | | } |
|---|
| | |
|---|
| | | public void onIdleContainer(Map<TopicPartition, Long> map, ConsumerSeekCallback consumerSeekCallback) { |
|---|
| | | |
|---|
| | | } |
|---|
| | | |
|---|
| | | } |
|---|
| | | |
|---|
