Parcourir la source

增加线程池处理

gaoyu il y a 3 ans
Parent
commit
b56ae7ba3c
1 fichiers modifiés avec 57 ajouts et 19 suppressions
  1. 57 19
      src/main/java/com/persagy/service/impl/AlarmHandleServiceImpl.java

+ 57 - 19
src/main/java/com/persagy/service/impl/AlarmHandleServiceImpl.java

@@ -10,8 +10,11 @@ import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.stream.Collectors;
 
 import org.apache.commons.lang3.StringUtils;
@@ -98,6 +101,10 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
 	 */
 	@Value("${system.id}")
 	public String systemId;
+	/**
+	 * 处理iot采集值消息线程池
+	 */
+	private ExecutorService fixedThreadPool = Executors.newFixedThreadPool(30);
 
 	/**
 	 * @description: 处理iot采集值消息
@@ -112,7 +119,7 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
 	 */
 	@RabbitHandler
 	@RabbitListener(queues = "alarm-data-queue")
-	public void handleIotMsg(Message message, Channel channel) throws IOException {
+	public void handleIotMsg(Message message, Channel channel) {
 		try {
 			JSONArray msgJsonArray = JSON.parseArray(new String(message.getBody()));
 			log.debug("接收到采集值:[{}]", msgJsonArray);
@@ -122,33 +129,65 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
 				log.error("采集值校验未通过");
 				return;
 			}
+			Map<String, JSONArray> map = new HashMap<>();
 			for (Object object : msgJsonArray) {
 				JSONObject jsonObject = (JSONObject) object;
-				// 采集时间
-				String dataTime = jsonObject.getString("dataTime");
 				// 报警类型
 				String itemCode = jsonObject.getString("itemCode");
 				// 设备id
 				String meterId = jsonObject.getString("meterId");
-				// 信息点id
-				String funcId = jsonObject.getString("funcId");
-				// 采集值
-				String value = jsonObject.getString("value");
-				synchronized (meterId.intern()) {
-					// 处理iot数据
-					HashMap<String, Object> paramMap = new HashMap<>();
-					JSONArray funcIdArray = JSON.parseArray(funcId);
-					JSONArray valuesArray = JSON.parseArray(value);
-					for (int i = 0; i < funcIdArray.size(); i++) {
-						paramMap.put((String) funcIdArray.getString(i), Double.parseDouble(valuesArray.getString(i)));
-					}
-					triggerOrEndAlarm(dataTime, meterId, itemCode, paramMap);
+				String key = meterId + "-" + itemCode;
+				JSONArray valueArray = map.get(key);
+				if (null != valueArray) {
+					valueArray.add(jsonObject);
+				} else {
+					valueArray = new JSONArray();
+					valueArray.add(jsonObject);
 				}
+			}
 
+			for (Entry<String, JSONArray> entry : map.entrySet()) {
+				fixedThreadPool.execute(() -> {
+					for (Object object : entry.getValue()) {
+						try {
+							JSONObject jsonObject = (JSONObject) object;
+							// 采集时间
+							String dataTime = jsonObject.getString("dataTime");
+							// 报警类型
+							String itemCode = jsonObject.getString("itemCode");
+							// 设备id
+							String meterId = jsonObject.getString("meterId");
+							// 信息点id
+							String funcId = jsonObject.getString("funcId");
+							// 采集值
+							String value = jsonObject.getString("value");
+							// 处理iot数据
+							HashMap<String, Object> paramMap = new HashMap<>();
+							JSONArray funcIdArray = JSON.parseArray(funcId);
+							JSONArray valuesArray = JSON.parseArray(value);
+							for (int i = 0; i < funcIdArray.size(); i++) {
+								paramMap.put((String) funcIdArray.getString(i),
+										Double.parseDouble(valuesArray.getString(i)));
+							}
+							triggerOrEndAlarm(dataTime, meterId, itemCode, paramMap);
+							alarmRedisCache.releaseLock(entry.getKey());
+							channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+						} catch (Exception e) {
+							try {
+								channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
+							} catch (IOException e1) {
+								e1.printStackTrace();
+							}
+						}
+					}
+				});
 			}
-			channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
 		} catch (Exception e) {
-			channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
+			try {
+				channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
+			} catch (IOException e1) {
+				e1.printStackTrace();
+			}
 		}
 
 	}
@@ -396,7 +435,6 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
 		}
 		// 更新报警定义缓存
 		alarmRedisCache.setAlarmState(defineId, alarmConditionState);
-		alarmRedisCache.releaseLock(alarmDefine.getObjId() + "-" + alarmDefine.getItemCode());
 	}
 
 	/**