瀏覽代碼

sdk增加websocket

menglu 3 年之前
父節點
當前提交
1c47cc4ecc

+ 13 - 15
ibms-data-sdk/pom.xml

@@ -149,11 +149,6 @@
 		</dependency>
 
 		<dependency>
-			<groupId>cn.hutool</groupId>
-			<artifactId>hutool-all</artifactId>
-			<version>5.5.4</version>
-		</dependency>
-		<dependency>
 			<groupId>org.apache.poi</groupId>
 			<artifactId>poi</artifactId>
 			<version>4.1.2</version>
@@ -165,10 +160,23 @@
 		</dependency>
 
 		<dependency>
+			<groupId>cn.hutool</groupId>
+			<artifactId>hutool-all</artifactId>
+			<version>5.5.4</version>
+			<scope>compile</scope>
+		</dependency>
+		<!--lombok -->
+		<dependency>
 			<groupId>org.projectlombok</groupId>
 			<artifactId>lombok</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<!-- netty 实现websocket -->
+		<dependency>
+			<groupId>io.netty</groupId>
+			<artifactId>netty-all</artifactId>
+			<version>4.1.62.Final</version>
+		</dependency>
 
 		<dependency>
 			<groupId>org.apache.mina</groupId>
@@ -180,16 +188,6 @@
 			<artifactId>ems-com</artifactId>
 			<version>1.7.0</version>
 		</dependency>
-		<dependency>
-			<groupId>com.persagy</groupId>
-			<artifactId>ems-websocket-manager</artifactId>
-			<version>1.0.0</version>
-		</dependency>
-		<dependency>
-			<groupId>org.projectlombok</groupId>
-			<artifactId>lombok</artifactId>
-			<version>1.18.0</version>
-		</dependency>
 
 		<!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-api -->
 		<dependency>

+ 3 - 3
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/service/rest/RestUtil.java

@@ -404,7 +404,7 @@ public class RestUtil {
         JSONObject parseObject = JSON.parseObject(param);
         String objId = parseObject.getString("objId");
         String code = parseObject.getString("code");
-        String pointString = Repository.info2point.get(objId).get(code);
+        String pointString = Repository.object2info2point.get(objId).get(code);
         int index_ = pointString.lastIndexOf('-');
         String meter = pointString.substring(0, index_);
         int funcid = Integer.parseInt(pointString.substring(index_ + 1));
@@ -417,7 +417,7 @@ public class RestUtil {
         JSONObject parseObject = JSON.parseObject(param);
         String objId = parseObject.getString("objId");
         String code = parseObject.getString("code");
-        String pointString = Repository.info2point.get(objId).get(code);
+        String pointString = Repository.object2info2point.get(objId).get(code);
         int index_ = pointString.lastIndexOf('-');
         String meter = pointString.substring(0, index_);
         int funcid = Integer.parseInt(pointString.substring(index_ + 1));
@@ -469,7 +469,7 @@ public class RestUtil {
             JSONObject parseObject = parseArray.getJSONObject(i);
             String objId = parseObject.getString("objId");
             String code = parseObject.getString("code");
-            String pointString = Repository.info2point.get(objId).get(code);
+            String pointString = Repository.object2info2point.get(objId).get(code);
             int index_ = pointString.lastIndexOf('-');
             String meter = pointString.substring(0, index_);
             int funcid = Integer.parseInt(pointString.substring(index_ + 1));

+ 57 - 0
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/service/websocket/server/NettyServer.java

@@ -0,0 +1,57 @@
+package com.persagy.ibms.data.sdk.service.websocket.server;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.HttpServerCodec;
+import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
+import io.netty.handler.stream.ChunkedWriteHandler;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class NettyServer {
+
+	private final int port;
+
+	public NettyServer(int port) {
+		this.port = port;
+	}
+
+	public void start() throws Exception {
+		EventLoopGroup bossGroup = new NioEventLoopGroup();
+
+		EventLoopGroup group = new NioEventLoopGroup();
+		try {
+			ServerBootstrap sb = new ServerBootstrap();
+			sb.option(ChannelOption.SO_BACKLOG, 1024);
+			sb.group(group, bossGroup) // 绑定线程池
+					.channel(NioServerSocketChannel.class) // 指定使用的channel
+					.localAddress(this.port)// 绑定监听端口
+					.childHandler(new ChannelInitializer<SocketChannel>() { // 绑定客户端连接时候触发操作
+						@Override
+						protected void initChannel(SocketChannel ch) throws Exception {
+							// websocket协议本身是基于http协议的,所以这边也要使用http解编码器
+							ch.pipeline().addLast(new HttpServerCodec());
+							// 以块的方式来写的处理器
+							ch.pipeline().addLast(new ChunkedWriteHandler());
+							ch.pipeline().addLast(new HttpObjectAggregator(655360));
+							ch.pipeline().addLast(new WebSocketHandler());
+							ch.pipeline()
+									.addLast(new WebSocketServerProtocolHandler("/websocket", null, true, 65536 * 10));
+						}
+					});
+			ChannelFuture cf = sb.bind().sync(); // 服务器异步创建绑定
+			log.info("启动正在监听: {}", cf.channel().localAddress());
+			cf.channel().closeFuture().sync(); // 关闭服务器通道
+		} finally {
+			group.shutdownGracefully().sync(); // 释放线程池资源
+			bossGroup.shutdownGracefully().sync();
+		}
+	}
+}

+ 31 - 0
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/service/websocket/server/WebSocketChannelPool.java

@@ -0,0 +1,31 @@
+package com.persagy.ibms.data.sdk.service.websocket.server;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import io.netty.channel.Channel;
+import io.netty.channel.group.ChannelGroup;
+import io.netty.channel.group.DefaultChannelGroup;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
+import io.netty.util.concurrent.GlobalEventExecutor;
+
+public class WebSocketChannelPool {
+	public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
+	// 收到信息后,群发消息
+	// ChannelHandlerPool.channelGroup.writeAndFlush(new TextWebSocketFrame(message));
+
+	public static Map<String, Channel> id2Channel = new ConcurrentHashMap<String, Channel>();
+	public static Map<String, Object> id2ContentJSON = new ConcurrentHashMap<String, Object>();
+
+	public static void Send(String id, Object content) {
+		try {
+			Channel channel = id2Channel.get(id);
+			if (channel != null) {
+				String sendString = content.toString();
+				channel.writeAndFlush(new TextWebSocketFrame(sendString));
+			}
+		} catch (Exception e) {
+			e.printStackTrace();
+		}
+	}
+}

+ 108 - 0
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/service/websocket/server/WebSocketHandler.java

@@ -0,0 +1,108 @@
+package com.persagy.ibms.data.sdk.service.websocket.server;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.alibaba.fastjson.JSON;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
+
+	private static final AtomicInteger onlineCount = new AtomicInteger(0);
+
+	private static Map<String, String> getUrlParams(String url) {
+		Map<String, String> map = new HashMap<String, String>();
+		url = url.replace("?", ";");
+		if (!url.contains(";")) {
+			return map;
+		}
+		if (url.split(";").length > 0) {
+			String[] arr = url.split(";")[1].split("&");
+			for (String s : arr) {
+				String key = s.split("=")[0];
+				String value = s.split("=")[1];
+				map.put(key, value);
+			}
+			return map;
+
+		} else {
+			return map;
+		}
+	}
+
+	@Override
+	public void channelActive(ChannelHandlerContext ctx) throws Exception {
+		log.info("与客户端{}建立连接,当前连接数量{}!", ctx.channel().remoteAddress(), onlineCount.incrementAndGet());
+		// 添加到channelGroup通道组
+		WebSocketChannelPool.channelGroup.add(ctx.channel());
+	}
+
+	@Override
+	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+		log.info("与客户端{}断开连接,通道关闭!当前连接数量{}!", ctx.channel().remoteAddress(), onlineCount.decrementAndGet());
+		// 添加到channelGroup 通道组
+		WebSocketChannelPool.channelGroup.remove(ctx.channel());
+		handlerRemove(ctx.channel());
+	}
+
+	@Override
+	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+		// 首次连接是FullHttpRequest,处理参数
+		if (null != msg && msg instanceof FullHttpRequest) {
+			FullHttpRequest request = (FullHttpRequest) msg;
+			String uri = request.uri();
+			// 如果url包含参数,需要记录
+			if (uri.contains("?")) {
+				Map<String, String> paramMap = getUrlParams(uri);
+				log.info("接收到的参数是:{}", JSON.toJSONString(paramMap));
+				Object ContentJSON = JSON.toJSON(paramMap);
+				handlerType(ctx.channel(), ContentJSON);
+				// url需要处理
+				String newUri = uri.substring(0, uri.indexOf("?"));
+				log.info(newUri);
+				request.setUri(newUri);
+			}
+		} else if (msg instanceof TextWebSocketFrame) {
+			// 正常的TEXT消息类型
+			TextWebSocketFrame frame = (TextWebSocketFrame) msg;
+			String remoteAddress = ctx.channel().remoteAddress().toString();
+			log.info("收到客户端{}数据:{}", remoteAddress, frame.text());
+			Object ContentJSON = JSON.parse(frame.text());
+			handlerType(ctx.channel(), ContentJSON);
+		}
+		super.channelRead(ctx, msg);
+	}
+
+	@Override
+	protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame)
+			throws Exception {
+
+	}
+
+	private synchronized void handlerType(Channel channel, Object ContentJSON) {
+		String id = channel.id().asLongText();
+		WebSocketChannelPool.id2Channel.put(id, channel);
+		WebSocketChannelPool.id2ContentJSON.put(id, ContentJSON);
+		WebSocketUtil.ProcessConnected(id, ContentJSON);
+	}
+
+	private synchronized void handlerRemove(Channel channel) {
+		try {
+			String id = channel.id().asLongText();
+			WebSocketChannelPool.id2Channel.remove(id);
+			WebSocketChannelPool.id2ContentJSON.remove(id);
+			WebSocketUtil.ProcessDisconnected(id);
+		} catch (Exception e) {
+			log.error(e.getMessage(), e);
+		}
+	}
+
+}

+ 54 - 0
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/service/websocket/server/WebSocketSendThread.java

@@ -0,0 +1,54 @@
+package com.persagy.ibms.data.sdk.service.websocket.server;
+
+import java.text.SimpleDateFormat;
+
+import com.alibaba.fastjson.JSONObject;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class WebSocketSendThread extends Thread {
+	SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
+	private volatile boolean stop = false;
+
+	public WebSocketSendThread() {
+	}
+
+	public void requestStop() {
+		stop = true;
+		try {
+			this.join();
+		} catch (InterruptedException e) {
+			log.error("发生异常", e);
+		}
+	}
+
+	@Override
+	public void run() {
+		while (!stop) {
+			try {
+				Thread.sleep(1000L);
+			} catch (InterruptedException e1) {
+				e1.printStackTrace();
+			}
+			try {
+				this.Process();
+			} catch (Exception e) {
+				log.error("发生异常", e);
+			}
+		}
+	}
+
+	private void Process() throws Exception {
+		for (String id : WebSocketUtil.idMap.keySet()) {
+			Object content = new JSONObject();
+			WebSocketUtil.executor.execute(() -> {
+				try {
+					WebSocketChannelPool.Send(id, content);
+				} catch (Exception e) {
+					e.printStackTrace();
+				}
+			});
+		}
+	}
+}

+ 35 - 0
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/service/websocket/server/WebSocketStarter.java

@@ -0,0 +1,35 @@
+package com.persagy.ibms.data.sdk.service.websocket.server;
+
+import javax.servlet.ServletContextEvent;
+import javax.servlet.ServletContextListener;
+import javax.servlet.annotation.WebListener;
+
+import org.springframework.beans.factory.annotation.Value;
+
+import cn.hutool.core.thread.ThreadUtil;
+import lombok.extern.slf4j.Slf4j;
+
+@WebListener
+@Slf4j
+public class WebSocketStarter implements ServletContextListener {
+	public static WebSocketSendThread thread = new WebSocketSendThread();
+
+	@Value("${websocket.port}")
+	private int websocketPort;
+
+	@Override
+	public void contextInitialized(ServletContextEvent sce) {
+
+		// 异步消费消息
+		ThreadUtil.execAsync(() -> {
+			try {
+				new NettyServer(websocketPort).start();
+			} catch (Exception e) {
+				log.error("netty websocker", e);
+			}
+		}, true);
+		log.info("----------contextInitialize end--------------");
+
+		// thread.start();
+	}
+}

+ 261 - 0
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/service/websocket/server/WebSocketUtil.java

@@ -0,0 +1,261 @@
+package com.persagy.ibms.data.sdk.service.websocket.server;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import com.persagy.ibms.core.data.SceneDataObject;
+import com.persagy.ibms.core.data.SceneDataSet;
+import com.persagy.ibms.core.data.SceneDataValue;
+import com.persagy.ibms.core.util.RWDUtil;
+import com.persagy.ibms.data.sdk.util.ObjectInfo;
+import com.persagy.ibms.data.sdk.util.RepositoryContainer;
+import com.persagy.ibms.data.sdk.util.RepositoryImpl;
+
+import cn.hutool.core.thread.ExecutorBuilder;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class WebSocketUtil {
+	public static ExecutorService executor = ExecutorBuilder.create().setCorePoolSize(4).setMaxPoolSize(10)
+			.setWorkQueue(new LinkedBlockingQueue<>(102400)).setHandler(new ThreadPoolExecutor.AbortPolicy()).build();
+	public static Map<String, Object> idMap = new ConcurrentHashMap<String, Object>();
+
+	public static Map<String, Map<String, Boolean>> objId2idList = new ConcurrentHashMap<String, Map<String, Boolean>>();
+	public static Map<String, Map<String, Boolean>> id2objIdList = new ConcurrentHashMap<String, Map<String, Boolean>>();
+
+	public static Map<String, Map<String, Boolean>> objInfoId2idList = new ConcurrentHashMap<String, Map<String, Boolean>>();
+	public static Map<String, Map<String, Boolean>> id2objInfoIdList = new ConcurrentHashMap<String, Map<String, Boolean>>();
+
+	public static synchronized void ProcessConnected(String id, Object ContentJSON) {
+		idMap.put(id, ContentJSON);
+		Map<String, Boolean> objIdList = new ConcurrentHashMap<String, Boolean>();
+		id2objIdList.put(id, objIdList);
+		Map<String, Boolean> objInfoIdList = new ConcurrentHashMap<String, Boolean>();
+		id2objInfoIdList.put(id, objInfoIdList);
+		JSONArray objArray = (JSONArray) ContentJSON;
+		for (int i = 0; i < objArray.size(); i++) {
+			JSONObject objJSON = objArray.getJSONObject(i);
+			String objId = (String) objJSON.get("objId");
+			if (objJSON.containsKey("infoCodeArray")) {
+				JSONArray infoCodeArray = (JSONArray) objJSON.get("infoCodeArray");
+				for (int ii = 0; ii < infoCodeArray.size(); ii++) {
+					String infoCode = (String) infoCodeArray.get(ii);
+					String objInfoId = objId + "-" + infoCode;
+					if (!objInfoId2idList.containsKey(objInfoId)) {
+						objInfoId2idList.put(objInfoId, new ConcurrentHashMap<String, Boolean>());
+					}
+					objInfoId2idList.get(objInfoId).put(id, true);
+					objInfoIdList.put(objInfoId, true);
+				}
+			} else {
+				if (!objId2idList.containsKey(objId)) {
+					objId2idList.put(objId, new ConcurrentHashMap<String, Boolean>());
+				}
+				objId2idList.get(objId).put(id, true);
+				objIdList.put(objId, true);
+			}
+		}
+
+		ProcessFirstSend(id);
+	}
+
+	public static synchronized void ProcessDisconnected(String id) {
+		for (String key : id2objIdList.keySet()) {
+			Map<String, Boolean> objIdList = id2objIdList.get(key);
+			for (String objId : objIdList.keySet()) {
+				Map<String, Boolean> idList = objId2idList.get(objId);
+				if (idList.containsKey(id)) {
+					idList.remove(id);
+				}
+			}
+		}
+		id2objIdList.remove(id);
+
+		for (String key : id2objInfoIdList.keySet()) {
+			Map<String, Boolean> objInfoIdList = id2objInfoIdList.get(key);
+			for (String objInfoId : objInfoIdList.keySet()) {
+				Map<String, Boolean> idList = objInfoId2idList.get(objInfoId);
+				if (idList.containsKey(id)) {
+					idList.remove(id);
+				}
+			}
+		}
+		id2objInfoIdList.remove(id);
+
+		idMap.remove(id);
+	}
+
+	private static void ProcessFirstSend(String id) {
+		try {
+			Runnable runnable = new Runnable() {
+				@Override
+				public void run() {
+					RepositoryImpl Repository = RepositoryContainer.instance;
+
+					JSONArray sendArray = new JSONArray();
+					Map<String, JSONObject> sendObj = new HashMap<String, JSONObject>();
+					Map<String, Boolean> objIdList = id2objIdList.get(id);
+					Map<String, Boolean> objInfoIdList = id2objInfoIdList.get(id);
+					for (String objId : objIdList.keySet()) {
+						if (Repository.id2sdv.containsKey(objId)) {
+							SceneDataObject sdo = Repository.id2sdv.get(objId);
+							String classCode = (String) sdo.get("classCode").value_prim.value;
+							SceneDataSet infoArray = Repository.infoArrayDic.get(classCode);
+							String[] infoCodes = sdo.keySet().toArray(new String[0]);
+							for (String infoCode : infoCodes) {
+								SceneDataValue infoValue = sdo.get(infoCode);
+								if ((RWDUtil.isRunParam(infoArray.set, infoCode) || RWDUtil.isSetParam(infoArray.set, infoCode))
+										&& infoValue.value_prim != null && infoValue.value_prim.value != null) {
+									if (!sendObj.containsKey(objId)) {
+										sendObj.put(objId, new JSONObject());
+									}
+									JSONObject sendItem = sendObj.get(objId);
+									sendItem.put(infoCode, infoValue.value_prim.value);
+								}
+							}
+						}
+					}
+					for (String objInfoId : objInfoIdList.keySet()) {
+						int index_ = objInfoId.indexOf('-');
+						String objId = objInfoId.substring(0, index_);
+						String infoCode = objInfoId.substring(index_ + 1);
+						if (Repository.id2sdv.containsKey(objId)) {
+							SceneDataObject sdo = Repository.id2sdv.get(objId);
+							String classCode = (String) sdo.get("classCode").value_prim.value;
+							SceneDataSet infoArray = Repository.infoArrayDic.get(classCode);
+							{
+								SceneDataValue infoValue = sdo.get(infoCode);
+								if ((RWDUtil.isRunParam(infoArray.set, infoCode) || RWDUtil.isSetParam(infoArray.set, infoCode))
+										&& infoValue.value_prim != null && infoValue.value_prim.value != null) {
+									if (!sendObj.containsKey(objId)) {
+										sendObj.put(objId, new JSONObject());
+									}
+									JSONObject sendItem = sendObj.get(objId);
+									sendItem.put(infoCode, infoValue.value_prim.value);
+								}
+							}
+						}
+					}
+					for (String objId : sendObj.keySet()) {
+						JSONObject sendItem = sendObj.get(objId);
+						sendItem.put("id", objId);
+						sendArray.add(sendItem);
+					}
+					if (sendArray.size() > 0) {
+						WebSocketChannelPool.Send(id, sendArray);
+					}
+				}
+			};
+			WebSocketUtil.executor.execute(new Thread(runnable));
+		} catch (Exception e) {
+			log.error("handlerSubMessage", e);
+		}
+	}
+
+	public static void ProcessIOTReceived(JSONObject json) {
+		try {
+			Runnable runnable = new Runnable() {
+				@Override
+				public void run() {
+					RepositoryImpl Repository = RepositoryContainer.instance;
+					String type = json.getString("type");
+					if (type.equals("iot") || type.equals("text")) {
+						String message = json.getString("data");
+						String[] splits = message.split(";");
+						for (String id : idMap.keySet()) {
+							try {
+								Map<String, Boolean> objIdList = id2objIdList.get(id);
+								Map<String, Boolean> objInfoIdList = id2objInfoIdList.get(id);
+								JSONArray sendArray = new JSONArray();
+								Map<String, JSONObject> sendObj = new HashMap<String, JSONObject>();
+								for (int i = 0; i < splits.length; i += 4) {
+									// String time = splits[i + 0];
+									String meter = splits[i + 1];
+									String funcid = splits[i + 2];
+									String value = splits[i + 3];
+									String point = meter + "-" + funcid;
+									if (!Repository.point2ObjectInfoList.containsKey(point)) {
+										continue;
+									}
+									List<ObjectInfo> ObjectInfoList = Repository.point2ObjectInfoList.get(point);
+									for (ObjectInfo ObjectInfo : ObjectInfoList) {
+										if (objIdList.containsKey(ObjectInfo.objId)
+												|| objInfoIdList.containsKey(ObjectInfo.objId + "" + ObjectInfo.infoCode)) {
+											if (!sendObj.containsKey(ObjectInfo.objId)) {
+												sendObj.put(ObjectInfo.objId, new JSONObject());
+											}
+											JSONObject sendItem = sendObj.get(ObjectInfo.objId);
+											sendItem.put(ObjectInfo.infoCode, value);
+										}
+									}
+								}
+								for (String objId : sendObj.keySet()) {
+									JSONObject sendItem = sendObj.get(objId);
+									sendItem.put("id", objId);
+									sendArray.add(sendItem);
+								}
+								if (sendArray.size() > 0) {
+									WebSocketChannelPool.Send(id, sendArray);
+								}
+							} catch (Exception e) {
+								e.printStackTrace();
+							}
+						}
+					} else if (type.equals("pointset")) {
+						String message = json.getString("data");
+						String[] splits = message.split(";");
+						for (String id : idMap.keySet()) {
+							try {
+								Map<String, Boolean> objIdList = id2objIdList.get(id);
+								Map<String, Boolean> objInfoIdList = id2objInfoIdList.get(id);
+								JSONArray sendArray = new JSONArray();
+								Map<String, JSONObject> sendObj = new HashMap<String, JSONObject>();
+								for (int i = 0; i < splits.length; i += 4) {
+									// String time = splits[i + 0];
+									String meter = splits[i + 1];
+									String funcid = splits[i + 2];
+									String value = splits[i + 3];
+									String point = meter + "-" + funcid;
+									if (!Repository.set2ObjectInfoList.containsKey(point)) {
+										continue;
+									}
+									List<ObjectInfo> ObjectInfoList = Repository.set2ObjectInfoList.get(point);
+									for (ObjectInfo ObjectInfo : ObjectInfoList) {
+										if (objIdList.containsKey(ObjectInfo.objId)
+												|| objInfoIdList.containsKey(ObjectInfo.objId + "" + ObjectInfo.infoCode)) {
+											if (!sendObj.containsKey(ObjectInfo.objId)) {
+												sendObj.put(ObjectInfo.objId, new JSONObject());
+											}
+											JSONObject sendItem = sendObj.get(ObjectInfo.objId);
+											sendItem.put(ObjectInfo.infoCode, value);
+										}
+									}
+								}
+								for (String objId : sendObj.keySet()) {
+									JSONObject sendItem = sendObj.get(objId);
+									sendItem.put("id", objId);
+									sendArray.add(sendItem);
+								}
+								if (sendArray.size() > 0) {
+									WebSocketChannelPool.Send(id, sendArray);
+								}
+							} catch (Exception e) {
+								e.printStackTrace();
+							}
+						}
+					}
+				}
+			};
+			WebSocketUtil.executor.execute(new Thread(runnable));
+		} catch (Exception e) {
+			log.error("handlerSubMessage", e);
+		}
+	}
+}

+ 11 - 0
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/util/ObjectInfo.java

@@ -0,0 +1,11 @@
+package com.persagy.ibms.data.sdk.util;
+
+public class ObjectInfo {
+	public String objId;
+	public String infoCode;
+
+	public ObjectInfo(String id, String info) {
+		this.objId = id;
+		this.infoCode = info;
+	}
+}

+ 13 - 2
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/util/RWDLoadUtil.java

@@ -9,6 +9,7 @@ import java.util.Date;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
 
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONArray;
@@ -262,9 +263,14 @@ public class RWDLoadUtil {
 						String point = (String) value_primitive;
 						if (point != null) {
 							if (RWDUtil.infoValue_is_point(point)) {
-								Repository.info2point.putIfAbsent(obj_id, new ConcurrentHashMap<String, String>());
-								ConcurrentHashMap<String, String> pointMap = Repository.info2point.get(obj_id);
+								Repository.object2info2point.putIfAbsent(obj_id, new ConcurrentHashMap<String, String>());
+								ConcurrentHashMap<String, String> pointMap = Repository.object2info2point.get(obj_id);
 								pointMap.putIfAbsent(Key, point);
+								if (!Repository.point2ObjectInfoList.containsKey(point)) {
+									Repository.point2ObjectInfoList.put(point, new CopyOnWriteArrayList<ObjectInfo>());
+								}
+								List<ObjectInfo> ObjectInfoList = Repository.point2ObjectInfoList.get(point);
+								ObjectInfoList.add(new ObjectInfo(obj_id, Key));
 								// 表号功能号改为null
 								{
 									SceneDataValue sdv = new SceneDataValue(null, null, null, null);
@@ -297,6 +303,11 @@ public class RWDLoadUtil {
 						String point = (String) value_primitive;
 						if (point != null) {
 							if (RWDUtil.infoValue_is_point(point)) {
+								if (!Repository.set2ObjectInfoList.containsKey(point)) {
+									Repository.set2ObjectInfoList.put(point, new CopyOnWriteArrayList<ObjectInfo>());
+								}
+								List<ObjectInfo> ObjectInfoList = Repository.set2ObjectInfoList.get(point);
+								ObjectInfoList.add(new ObjectInfo(obj_id, Key));
 								// 表号功能号改为null
 								{
 									SceneDataValue sdv = new SceneDataValue(null, null, null, null);

+ 1 - 1
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/util/RWDRepositoryUtil.java

@@ -57,7 +57,7 @@ public class RWDRepositoryUtil {
 			Repository.infoArrayDic = RepositoryContainer.instance.infoArrayDic;
 			Repository.objectArrayDic = RepositoryContainer.instance.objectArrayDic;
 			Repository.objectArrayAll = RepositoryContainer.instance.objectArrayAll;
-			Repository.info2point = RepositoryContainer.instance.info2point;
+			Repository.object2info2point = RepositoryContainer.instance.object2info2point;
 			Repository.objType2id2Value = RepositoryContainer.instance.objType2id2Value;
 			Repository.relationArrayDic = RepositoryContainer.instance.relationArrayDic;
 			Repository.graphCodeDic = RepositoryContainer.instance.graphCodeDic;

+ 4 - 1
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/util/RepositoryImpl.java

@@ -1,5 +1,6 @@
 package com.persagy.ibms.data.sdk.util;
 
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -32,7 +33,9 @@ public class RepositoryImpl extends RepositoryBase {
 	public Map<String, SceneDataSet> objectArrayDic = new ConcurrentHashMap<String, SceneDataSet>();
 	// 所有对象实例清单
 	public SceneDataSet objectArrayAll = new SceneDataSet(false);
-	public ConcurrentHashMap<String, ConcurrentHashMap<String, String>> info2point = new ConcurrentHashMap<String, ConcurrentHashMap<String, String>>();
+	public ConcurrentHashMap<String, ConcurrentHashMap<String, String>> object2info2point = new ConcurrentHashMap<String, ConcurrentHashMap<String, String>>();
+	public ConcurrentHashMap<String, List<ObjectInfo>> point2ObjectInfoList = new ConcurrentHashMap<String, List<ObjectInfo>>();
+	public ConcurrentHashMap<String, List<ObjectInfo>> set2ObjectInfoList = new ConcurrentHashMap<String, List<ObjectInfo>>();
 	// 对象类型到map,map里面是id到对象值。只有关联表里面的对象才有
 	public Map<String, Map<String, SceneDataObject>> objType2id2Value = new ConcurrentHashMap<String, Map<String, SceneDataObject>>();
 	// 图类型到边类型到变实例清单

+ 15 - 1
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/websocket/IOTWebSocketClient.java

@@ -1,6 +1,7 @@
 package com.persagy.ibms.data.sdk.websocket;
 
 import java.net.URI;
+import java.util.Date;
 
 import org.java_websocket.client.WebSocketClient;
 import org.java_websocket.handshake.ServerHandshake;
@@ -10,6 +11,7 @@ import com.alibaba.fastjson.JSONObject;
 import com.persagy.ibms.core.data.SceneDataPrimitive;
 import com.persagy.ibms.core.data.SceneDataValue;
 import com.persagy.ibms.core.util.DataUtil;
+import com.persagy.ibms.data.sdk.service.websocket.server.WebSocketUtil;
 import com.persagy.ibms.data.sdk.util.RepositoryContainer;
 
 import lombok.extern.slf4j.Slf4j;
@@ -44,10 +46,21 @@ public class IOTWebSocketClient extends WebSocketClient {
 		// this.close_connect();
 	}
 
+	Date lastTime = new Date();
+	volatile int recent_count = 0;
+
 	@Override
 	public void onMessage(String arg0) {
 		try {
+			Date currTime = new Date();
+			if (currTime.getTime() / (1000L * 60) != lastTime.getTime() / (1000L * 60)) {
+				lastTime = currTime;
+				log.warn("IOTWebSocketClient rece count: " + recent_count);
+				recent_count = 0;
+			}
+
 			JSONObject json = (JSONObject) JSON.parse(arg0);
+			WebSocketUtil.ProcessIOTReceived(json);
 			String type = json.getString("type");
 			if (type.equals("iot") || type.equals("text")) {
 				String message = json.getString("data");
@@ -55,6 +68,7 @@ public class IOTWebSocketClient extends WebSocketClient {
 				// log.info(message);
 
 				String[] splits = message.split(";");
+				recent_count += splits.length / 4;
 				for (int i = 0; i < splits.length; i += 4) {
 					// String time = splits[i + 0];
 					String meter = splits[i + 1];
@@ -86,6 +100,7 @@ public class IOTWebSocketClient extends WebSocketClient {
 			} else if (type.equals("pointset")) {
 				String message = json.getString("data");
 				String[] splits = message.split(";");
+				recent_count += splits.length / 4;
 				int i = 0;
 				{
 					// String time = splits[i + 0];
@@ -107,7 +122,6 @@ public class IOTWebSocketClient extends WebSocketClient {
 						// log.info("onMessage" + "\t" + meter + "-" + funcid + "\t" + value);
 					}
 				}
-
 			}
 		} catch (Exception e) {
 			log.error(e.getMessage(), e);

+ 1 - 0
ibms-data-sdk/src/main/resources/application.properties

@@ -1,4 +1,5 @@
 server.port=8808
+websocket.port=8818
 spring.application.name=ibms-data-sdk
 
 autolog.app-name: ${spring.application.name}