| | |
|---|
| | | import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
|---|
| | | import org.springframework.beans.factory.annotation.Value;
|
|---|
| | | import org.springframework.core.io.ClassPathResource;
|
|---|
| | | import org.springframework.data.mongodb.core.MongoTemplate;
|
|---|
| | | import org.springframework.data.redis.core.RedisTemplate;
|
|---|
| | | import org.springframework.util.StringUtils;
|
|---|
| | |
|
|---|
| | |
|---|
| | | import com.moral.monitor.dao.TaskDao;
|
|---|
| | | import com.moral.monitor.entity.History;
|
|---|
| | | import com.moral.monitor.entity.Sensor;
|
|---|
| | | import com.moral.monitor.service.RedisService;
|
|---|
| | | import com.moral.monitor.util.RedisUtil;
|
|---|
| | |
|
|---|
| | | public class TaskListener implements MessageListener {
|
|---|
| | |
|---|
| | | @Resource
|
|---|
| | | RedisTemplate<String, String> redisTemplate;
|
|---|
| | |
|
|---|
| | | @Resource
|
|---|
| | | protected MongoTemplate mongoTemplate;
|
|---|
| | |
|
|---|
| | | private Logger logger = LoggerFactory.getLogger(TaskListener.class);
|
|---|
| | |
|
|---|
| | | @Resource
|
|---|
| | | RedisService redisService;
|
|---|
| | |
|
|---|
| | | private String levelKey = "alarm_level_config";
|
|---|
| | |
|
|---|
| | | @Override
|
|---|
| | | public void onMessage(Message msg) {
|
|---|
| | | try {
|
|---|
| | | String message = null;
|
|---|
| | | try {
|
|---|
| | | message = new String(msg.getBody(), "utf-8");
|
|---|
| | | } catch (UnsupportedEncodingException e) {
|
|---|
| | | logger.warn(e.getMessage());
|
|---|
| | | }
|
|---|
| | |
|
|---|
| | | String message = null;
|
|---|
| | | try {
|
|---|
| | | message = new String(msg.getBody(), "utf-8");
|
|---|
| | | } catch (UnsupportedEncodingException e) {
|
|---|
| | | logger.warn(e.getMessage());
|
|---|
| | | }
|
|---|
| | | Map<String, Object> msgData = JSON.parseObject(message, new TypeReference<Map<String, Object>>() {});
|
|---|
| | | Object macObj = msgData.get("mac");
|
|---|
| | | Object timeObj = msgData.get("time");
|
|---|
| | | Object verObj = msgData.get("ver");
|
|---|
| | | if(StringUtils.isEmpty(verObj) || StringUtils.isEmpty(timeObj) || StringUtils.isEmpty(macObj)) {
|
|---|
| | | logger.warn("some properties is null, param[0] message:"+JSON.toJSONString(message));
|
|---|
| | | return;
|
|---|
| | | }
|
|---|
| | |
|
|---|
| | | Map<String, String> msgData = JSON.parseObject(message, new TypeReference<Map<String, String>>() {});
|
|---|
| | | String mac = msgData.get("mac");
|
|---|
| | | String ver = msgData.get("ver");
|
|---|
| | | String mac = macObj.toString();
|
|---|
| | | String ver = verObj.toString();
|
|---|
| | |
|
|---|
| | | if(StringUtils.isEmpty(ver) || StringUtils.isEmpty(mac)) {
|
|---|
| | | return;
|
|---|
| | | }
|
|---|
| | |
|
|---|
| | | Map<String, Float> adjustMap;
|
|---|
| | | String adjust_key = "adjust_" + mac;
|
|---|
| | | if(RedisUtil.hasKey(redisTemplate, adjust_key)) {
|
|---|
| | | adjustMap = JSON.parseObject(RedisUtil.get(redisTemplate, adjust_key), new TypeReference<Map<String, Float>>() {});
|
|---|
| | | } else {
|
|---|
| | | adjustMap = getAdjustData(mac);
|
|---|
| | | RedisUtil.set(redisTemplate, adjust_key, JSON.toJSONString(adjustMap));
|
|---|
| | | }
|
|---|
| | |
|
|---|
| | | if(!adjustMap.isEmpty()) {
|
|---|
| | | for (Map.Entry<String, Float> entry : adjustMap.entrySet()) {
|
|---|
| | | String key = entry.getKey();
|
|---|
| | | if(msgData.containsKey(key)) {
|
|---|
| | | Float value = entry.getValue();
|
|---|
| | | Float dataValue = Float.valueOf(msgData.get(key)) ;
|
|---|
| | | msgData.put(key, String.valueOf(dataValue + value));
|
|---|
| | | //获取缓存中的校准值map |
|---|
| | | Map<String, Float> adjustMap = redisService.getAdjustsByMac(mac);
|
|---|
| | | if(adjustMap!=null&&!adjustMap.isEmpty()) {
|
|---|
| | | for (Map.Entry<String, Float> entry : adjustMap.entrySet()) {
|
|---|
| | | String key = entry.getKey();
|
|---|
| | | if(msgData.containsKey(key)) {
|
|---|
| | | Float value = entry.getValue();
|
|---|
| | | Float dataValue = Float.valueOf(msgData.get(key).toString()) ;
|
|---|
| | | msgData.put(key, dataValue + value);
|
|---|
| | | }
|
|---|
| | | }
|
|---|
| | | }
|
|---|
| | | }
|
|---|
| | | // System.out.println(message);
|
|---|
| | | // System.out.println(JSON.toJSONString(msgData));
|
|---|
| | |
|
|---|
| | | // System.out.println(message);
|
|---|
| | | // System.out.println(JSON.toJSONString(msgData));
|
|---|
| | | String new_message = JSON.toJSONString(msgData);
|
|---|
| | | rabbitTemplate.convertAndSend("monitors_data2", "", new_message.getBytes());
|
|---|
| | | // rabbitTemplate.send("monitors_data2", "", new Message(JSON.toJSONString(msgData).getBytes(), new MessageProperties()));
|
|---|
| | |
|
|---|
| | | rabbitTemplate.convertAndSend("monitors_data2", "", JSON.toJSONString(msgData).getBytes());
|
|---|
| | | // rabbitTemplate.send("monitors_data_3", "", new Message(JSON.toJSONString(msgData).getBytes(), new MessageProperties()));
|
|---|
| | | // int state = (new Random()).nextInt(4) % 5; //TODO
|
|---|
| | | int state = detEquState(msgData);
|
|---|
| | | // rabbitTemplate.convertAndSend("monitors_alarm", "", "{\"mac\": \"" + mac + "\", \"state\": " + state + "}");
|
|---|
| | |
|
|---|
| | | // int state = (new Random()).nextInt(4) % 5; //TODO
|
|---|
| | | int state = detEquState(msgData);
|
|---|
| | | // rabbitTemplate.convertAndSend("monitors_alarm", "", "{\"mac\": \"" + mac + "\", \"state\": " + state + "}");
|
|---|
| | | //更新设备状态
|
|---|
| | | jobDao.updateStateByMac(mac, state);
|
|---|
| | |
|
|---|
| | | //更新设备状态
|
|---|
| | | jobDao.updateStateByMac(mac, state);
|
|---|
| | | // //保存原始数据
|
|---|
| | | // taskDao.insertTologger(mac, message, getDate());
|
|---|
| | | mongoTemplate.insert(message, "logger");
|
|---|
| | |
|
|---|
| | | //保存原始数据
|
|---|
| | | taskDao.insertTologger(mac, message, getDate());
|
|---|
| | |
|
|---|
| | | //保存历史记录
|
|---|
| | | List<History> histories = new ArrayList<History>();
|
|---|
| | | List<Sensor> sensorList = taskDao.selectFromsensorByver(ver);
|
|---|
| | | for (Sensor sensor : sensorList) {
|
|---|
| | | String key = sensor.getMac_key();
|
|---|
| | | if(msgData.containsKey(key)) {
|
|---|
| | | //taskDao.insertTohistory(mac, sensor.getSensor(), key, msgData.get(key));
|
|---|
| | | History history = new History();
|
|---|
| | | history.setMac(mac);
|
|---|
| | | history.setMac_key(key);
|
|---|
| | | history.setSensor(sensor.getSensor());
|
|---|
| | | history.setMac_value(Double.valueOf(msgData.get(key)));
|
|---|
| | | histories.add(history);
|
|---|
| | | }
|
|---|
| | | }
|
|---|
| | | jobDao.batchInsertHistory(histories);
|
|---|
| | | }
|
|---|
| | | //暂停保存历史记录
|
|---|
| | | // //保存历史记录
|
|---|
| | | // List<History> histories = new ArrayList<History>();
|
|---|
| | | // List<Sensor> sensorList = taskDao.selectFromsensorByver(ver);
|
|---|
| | | // for (Sensor sensor : sensorList) {
|
|---|
| | | // String key = sensor.getMac_key();
|
|---|
| | | // if(msgData.containsKey(key)) {
|
|---|
| | | // //taskDao.insertTohistory(mac, sensor.getSensor(), key, msgData.get(key));
|
|---|
| | | // History history = new History();
|
|---|
| | | // history.setMac(mac);
|
|---|
| | | // history.setMac_key(key);
|
|---|
| | | // history.setTime(new Date(Long.parseLong(msgData.get("time").toString())));
|
|---|
| | | // history.setSensor(sensor.getSensor());
|
|---|
| | | // history.setMac_value(Double.valueOf(msgData.get(key).toString()));
|
|---|
| | | // histories.add(history);
|
|---|
| | | // }
|
|---|
| | | // }
|
|---|
| | | // jobDao.batchInsertHistory(histories);
|
|---|
| | |
|
|---|
| | | private Map<String, Float> getAdjustData(String mac) {
|
|---|
| | | List<Map<String, String>> adjusts = jobDao.findAdjustByMac(mac);
|
|---|
| | | Map<String, Float> dataMap = new HashMap<String, Float>();
|
|---|
| | | for (int i = 0; i < adjusts.size(); i++) {
|
|---|
| | | Map adjust = adjusts.get(i);
|
|---|
| | | if(adjust.get("value") != null) {
|
|---|
| | | String key = String.valueOf(adjust.get("key"));
|
|---|
| | | String data = String.valueOf(adjust.get("value"));
|
|---|
| | | BigDecimal value = new BigDecimal(data);
|
|---|
| | | value.setScale(3, BigDecimal.ROUND_HALF_UP);
|
|---|
| | | dataMap.put(key, value.floatValue());
|
|---|
| | | }
|
|---|
| | | }
|
|---|
| | | return dataMap;
|
|---|
| | | }
|
|---|
| | | //三级警报值阀值
|
|---|
| | | private static Map<String,Double[]> alarmLevles =new HashMap<String, Double[]>();
|
|---|
| | | @Value(value="alarmLevles.json")
|
|---|
| | | public void setAlarmLevles(String path) throws IOException {
|
|---|
| | | org.springframework.core.io.Resource resource = new ClassPathResource(path);
|
|---|
| | | InputStreamReader reader=new InputStreamReader(resource.getInputStream());
|
|---|
| | | JSONReader jsonReader = new JSONReader(reader);
|
|---|
| | | //三级警报值阀值
|
|---|
| | | Map<String,Map<String,Double>> alarmLevleMap= jsonReader.readObject(new TypeReference<Map<String,Map<String,Double>>>(){});
|
|---|
| | | //将map转换成数组
|
|---|
| | | for(String key:alarmLevleMap.keySet()) {
|
|---|
| | | Map<String,Double> levels = alarmLevleMap.get(key);
|
|---|
| | | double level1 = levels.get("level1");
|
|---|
| | | double level2 = levels.get("level2");
|
|---|
| | | double level3 = levels.get("level3");
|
|---|
| | | //三级警报数组
|
|---|
| | | Double arr[] = {0.0,level1,level2,level3};
|
|---|
| | | alarmLevles.put(key, arr);
|
|---|
| | |
|
|---|
| | | //mongoTemplate.insert(JSON.toJSONString(histories), "data");
|
|---|
| | | mongoTemplate.insert(new_message, "data");
|
|---|
| | | } catch (Exception e) {
|
|---|
| | | //打印错误
|
|---|
| | | logger.error(e.getMessage());
|
|---|
| | | e.printStackTrace();
|
|---|
| | | }
|
|---|
| | | }
|
|---|
| | | // //三级警报值阀值
|
|---|
| | | // private static Map<String,Double[]> alarmLevles =new HashMap<String, Double[]>();
|
|---|
| | | // /*
|
|---|
| | | // * 设置三级警报阀值,把map数据结构转换成数组结构
|
|---|
| | | // */
|
|---|
| | | // @Value(value= "system/alarmLevels.json")
|
|---|
| | | // public void setAlarmLevles(String path) throws IOException {
|
|---|
| | | // org.springframework.core.io.Resource resource = new ClassPathResource(path);
|
|---|
| | | // InputStreamReader reader=new InputStreamReader(resource.getInputStream());
|
|---|
| | | // JSONReader jsonReader = new JSONReader(reader);
|
|---|
| | | // //三级警报值阀值
|
|---|
| | | // Map<String,Map<String,Double>> alarmLevleMap= jsonReader.readObject(new TypeReference<Map<String,Map<String,Double>>>(){});
|
|---|
| | | // //将map转换成数组
|
|---|
| | | // for(String key:alarmLevleMap.keySet()) {
|
|---|
| | | // Map<String,Double> levels = alarmLevleMap.get(key);
|
|---|
| | | // double level1 = levels.get("level1");
|
|---|
| | | // double level2 = levels.get("level2");
|
|---|
| | | // double level3 = levels.get("level3");
|
|---|
| | | // //当三个值都为零时，不加入阀值
|
|---|
| | | // if(levels.get("enable")!=null&&levels.get("enable")==1) {
|
|---|
| | | // //三级警报数组
|
|---|
| | | // Double arr[] = {0.0,level1,level2,level3};
|
|---|
| | | // alarmLevles.put(key, arr);
|
|---|
| | | // }
|
|---|
| | | // }
|
|---|
| | | // }
|
|---|
| | |
|
|---|
| | | private Map<String,Double[]> getAlarmLevels() throws IOException {
|
|---|
| | |
|
|---|
| | | Map<String,Map<String,Double>> alarmLevleMap = null;
|
|---|
| | | Map<String,Double[]> alarmLevles = new HashMap<String, Double[]>();
|
|---|
| | | if(RedisUtil.hasKey(redisTemplate, levelKey)) {
|
|---|
| | | String levelConfigStr = RedisUtil.get(redisTemplate, levelKey);
|
|---|
| | | alarmLevleMap = JSON.parseObject(levelConfigStr, new TypeReference<Map<String, Map<String,Double>>>() {});
|
|---|
| | | } else {
|
|---|
| | | org.springframework.core.io.Resource resource = new ClassPathResource("system/alarmLevels.json");
|
|---|
| | | InputStreamReader reader = new InputStreamReader(resource.getInputStream());
|
|---|
| | | JSONReader jsonReader = new JSONReader(reader);
|
|---|
| | | //三级警报值阀值
|
|---|
| | | alarmLevleMap= jsonReader.readObject(new TypeReference<Map<String,Map<String,Double>>>(){});
|
|---|
| | | }
|
|---|
| | |
|
|---|
| | | //将map转换成数组
|
|---|
| | | for(String key:alarmLevleMap.keySet()) {
|
|---|
| | | Map<String,Double> levels = alarmLevleMap.get(key);
|
|---|
| | | double level1 = levels.get("level1");
|
|---|
| | | double level2 = levels.get("level2");
|
|---|
| | | double level3 = levels.get("level3");
|
|---|
| | | //当三个值都为零时，不加入阀值
|
|---|
| | | if(levels.get("enable")!=null&&levels.get("enable")==1) {
|
|---|
| | | //三级警报数组
|
|---|
| | | Double arr[] = {0.0,level1,level2,level3};
|
|---|
| | | alarmLevles.put(key, arr);
|
|---|
| | | }
|
|---|
| | | }
|
|---|
| | | return alarmLevles;
|
|---|
| | | }
|
|---|
| | |
|
|---|
| | | //根据三级警报阀值，确定设备状态
|
|---|
| | | private int detEquState(Map<String,String> data) {
|
|---|
| | | private int detEquState(Map<String,Object> data) throws IOException {
|
|---|
| | |
|
|---|
| | | Map<String,Double[]> alarmLevles = getAlarmLevels();
|
|---|
| | |
|
|---|
| | | int state = 0;
|
|---|
| | | Map<String, Object> equMap = new HashMap<String, Object>();
|
|---|
| | | String mac = data.get("mac").toString();
|
|---|
| | | equMap.putAll(data);
|
|---|
| | | if(data!=null) {
|
|---|
| | | for(String key:alarmLevles.keySet()) {
|
|---|
| | | String value = data.get(key);
|
|---|
| | | String value = data.get(key).toString();
|
|---|
| | | if(!StringUtils.isEmpty(value)) {
|
|---|
| | | double val = Double.parseDouble(value);
|
|---|
| | | //获取三级警报阀值
|
|---|
| | |
|---|
| | | if(val>=arr[index]) {
|
|---|
| | | //如果当前状态级别更高就取当前状态。否则保持之前状态
|
|---|
| | | state = (index>state)?index:state;
|
|---|
| | | break;
|
|---|
| | | String e_key = "level"+Integer.toString(index);
|
|---|
| | | Object e_States = equMap.get(e_key);
|
|---|
| | | if(e_States!=null&&e_States instanceof List) {
|
|---|
| | | ((List)e_States).add(key);
|
|---|
| | | }else {
|
|---|
| | | e_States = new ArrayList<String>();
|
|---|
| | | equMap.put(e_key, e_States);
|
|---|
| | | ((List)e_States).add(key);
|
|---|
| | | }
|
|---|
| | | equMap.put(e_key, e_States);
|
|---|
| | | break;
|
|---|
| | | }
|
|---|
| | | }
|
|---|
| | | //当前状态到达最大级别不再遍历，跳出循环
|
|---|
| | | if(state == (arr.length-1)) {
|
|---|
| | | break;
|
|---|
| | | }
|
|---|
| | | }
|
|---|
| | | }
|
|---|
| | | }
|
|---|
| | | equMap.put("state", state);//设备状态 |
|---|
| | | redisService.setEquState(mac,equMap);
|
|---|
| | | return state;
|
|---|
| | | }
|
|---|
| | | private String getDate() {
|
|---|
| | | Date d = new Date();
|
|---|
| | | SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
|
|---|
| | | return formatter.format(d);
|
|---|
| | | }
|
|---|
| | | }
|
|---|
