WebSocket使用总结

小灰灰 2022-12-01 14:29 93阅读 0赞

# WebSocket使用总结 #

## 是什么 ##

WebSocket 协议在2008年诞生，2011年成为国际标准。所有浏览器都已经支持了。

它的最大特点就是，服务器可以主动向客户端推送信息，客户端也可以主动向服务器发送信息，是真正的双向平等对话，属于[服务器推送技术][Link 1]的一种。

其他特点包括：

（1）建立在 TCP 协议之上，服务器端的实现比较容易。

（2）与 HTTP 协议有着良好的兼容性。默认端口也是80和443，并且握手阶段采用 HTTP 协议，因此握手时不容易屏蔽，能通过各种 HTTP 代理服务器。

（3）数据格式比较轻量，性能开销小，通信高效。

（4）可以发送文本，也可以发送二进制数据。

（5）没有同源限制，客户端可以与任意服务器通信。

（6）协议标识符是`ws`（如果加密，则为`wss`），服务器网址就是 URL。

## 怎么用（java+springboot） ##

### 配置 ###

@Configuration
    public class WebSocketConfig { 
        public WebSocketConfig() { 
        }
    
        @Bean
        public ServerEndpointExporter serverEndpointExporter() { 
            return new ServerEndpointExporter();
        }
    }

ServerEndpointExporter的作用是将@ServerEndpoint 注解的类 注入到容器中,源码如下getBeanNamesForAnnotation(ServerEndpoint.class)

protected void registerEndpoints() { 
    		Set<Class<?>> endpointClasses = new LinkedHashSet<>();
    		if (this.annotatedEndpointClasses != null) { 
    			endpointClasses.addAll(this.annotatedEndpointClasses);
    		}
    
    		ApplicationContext context = getApplicationContext();
    		if (context != null) { 
    			String[] endpointBeanNames = context.getBeanNamesForAnnotation(ServerEndpoint.class);
    			for (String beanName : endpointBeanNames) { 
    				endpointClasses.add(context.getType(beanName));
    			}
    		}

### 使用示例 ###

通过配置ServerEndpointConfig.Configurator将httpsession 加入到websocket中

public class HttpSessionConfigurator extends ServerEndpointConfig.Configurator{ 
        @Override
        public void modifyHandshake(ServerEndpointConfig sec, HandshakeRequest request, HandshakeResponse response) { 
            HttpSession httpSession = (HttpSession) request.getHttpSession();
            sec.getUserProperties().put(HttpSession.class.getName(),httpSession);
            super.modifyHandshake(sec, request, response);
        }
    
    }

@ServerEndpoint(value="/ui/eventMonitorServer/{userId}",configurator = HttpSessionConfigurator.class)
    @Service("eventMonitorServer")
    public class EventMonitorServer { 
    
        //静态变量，用来记录当前在线连接数
        private static AtomicLong onlineCount = new AtomicLong();
    
        //记录每个用户下多个终端的连接
        private static Map<String, Set<EventMonitorServer>> userSocket = new HashMap<>();
    
        //需要session来对用户发送数据, 获取连接特征userId
        private Session session;
    
    
    
        //这个地方是有考究的 不能直接使用autowaire注入
        //https://blog.csdn.net/qq_42109746/article/details/102780642 参考这篇博客
        //不用static 对象会变成null websocket 不是单例，所以使用spring注入一次后，后面的对象就不会再注入了，会报null。
        //spring管理的都是单例（singleton），和 websocket （多对象）相冲突。
        //
        //项目启动时初始化，会初始化 websocket （非用户连接的），spring 同时会为其注入 service，该对象的 service 不是 null，
        // 被成功注入。但是，由于 spring 默认管理的是单例，所以只会注入一次 service。
        // 当新用户进入聊天时，系统又会创建一个新的 websocket 对象，这时矛盾出现了：spring 管理的都是单例，不会给第二个 websocket 对象注入 service，
        // 所以导致只要是用户连接创建的 websocket 对象，都不能再注入了。
        private static UpmRestService upmRestService;
    
        //对于静态变量的初始化 使用set方法 或者使用@PostConstruct
        @Autowired
        public void setUpmRestService(UpmRestService upmRestService){ 
            EventMonitorServer.upmRestService = upmRestService;
        }
    
        //https://www.cnblogs.com/cnsyear/p/12635372.html 网上也推荐使用ApplicationContextAware 从容器中获取对应的bean 测温组件使用
    
    
        /** * 登录用户的Id */
        private String userId;
    
        /** * 当前登录用户是否系统管理员 */
        private volatile boolean isSysadmin;
    
        /** *httpSessionId参数名 */
        private static final String HTTP_SESSION_ID = "httpSessionId";
    
        /** *客户端连接后，一些初始化操作 */
        private void init(String userId, Session session, EndpointConfig config){ 
            HttpSession httpSession = (HttpSession)config.getUserProperties().get(HttpSession.class.getName());
            this.session = session;
            //将httpsessionid设置到websocket中
            if (httpSession!=null){ 
                this.session.getUserProperties().put(HTTP_SESSION_ID,httpSession.getId());
            }else{ 
                this.session.getUserProperties().put(HTTP_SESSION_ID,new Date().toString());
            }
    
            this.userId = userId;
            //对Map加锁访问，避免多线程
            synchronized (userSocket){ 
                //根据该用户当前是否已经在别的终端登录进行添加操作
                if (userSocket.containsKey(this.userId)) { 
                    LOGGER.debug("local user id:{} is login in other terminal",this.userId);
                    userSocket.get(this.userId).add(this); //增加该用户set中的连接实例
                }else { 
                    LOGGER.debug("local user id:{}first terminal:{} login",this.userId,session.getUserProperties().get(HTTP_SESSION_ID));
                    Set<EventMonitorServer> addUserSet = new HashSet<>();
                    addUserSet.add(this);
                    userSocket.put(this.userId, addUserSet);
                }
            }
            isSysadmin = upmRestService.isSysManager(userId);
            if (!isSysadmin){ 
                List<String> resourceCodeList = new ArrayList<>();
                BaseResult<BasePage<UpmResourceRet>> bs = upmRestService.queryUserPrivilege(userId);
                if(bs.getCode().equals("0")){ 
                    if (bs.getData().getTotal().equals(0)){ 
    
                    }else{ 
                        resourceCodeList = bs.getData().getList().stream().map(t->{ return t.getResourceIndexCode();}).collect(Collectors.toList());
                    }
                }
    
                if (!CollectionUtils.isEmpty(resourceCodeList)){ 
                    USER_ID_PRIVILEGE_RESOURCE_CODE.put(userId,resourceCodeList);
                }
            }
    
            LOGGER.debug("after init user {} login terminal num = {}",userId,userSocket.get(this.userId).size());
            LOGGER.debug("after init local user num is：{},terminal num is：{}",userSocket.size(),onlineCount.incrementAndGet());
        }
    
        /** *释放相关缓存信息 */
        private void destory(){ 
            synchronized (userSocket){ 
                //移除当前用户终端登录的websocket信息,如果该用户的所有终端都下线了，则删除该用户的记录
                if (userSocket.get(this.userId).size() == 0) { 
                    userSocket.remove(this.userId);
                }else{ 
                    userSocket.get(this.userId).remove(this);
                }
            }
    
            LOGGER.debug("after destory user {} login terminal num = {}",userId,userSocket.get(this.userId).size());
            LOGGER.debug("after destory local user num is：{},terminal num is：{}",userSocket.size(),onlineCount.decrementAndGet());
        }
    
        /** * @Title: onOpen * @Description: websocekt连接建立时的操作 * @param @param userId 用户id * @param @param session websocket连接的session属性 * @param @throws IOException */
        @OnOpen
        public void onOpen(@PathParam("userId") String userId, Session session, EndpointConfig config) throws IOException { 
            init(userId, session, config);
        }
    
        /** * @Title: onClose * @Description: 连接关闭的操作 */
        @OnClose
        public void onClose(){ 
            destory();
        }
    
        /** * @Title: onMessage * @Description: 收到消息后的操作 * @param @param message 收到的消息 * @param @param session 该连接的session属性 */
        @OnMessage
        public void onMessage(String message, Session session) { 
            LOGGER.debug("recevice from user id：{},terminal is: {} message：{}",this.userId,session.getUserProperties().get(HTTP_SESSION_ID),message);
        }
    
        /** * @Title: onError * @Description: 连接发生错误时候的操作 * @param @param session 该连接的session * @param @param error 发生的错误 */
        @OnError
        public void onError(Session session, Throwable error){ 
            LOGGER.error("user id=：{},terminal:{} connect error",this.userId,session.getUserProperties().get(HTTP_SESSION_ID),error);
        }
    
        /** * 向实时检测client发送检测数据的消息 * @param resourceId 资源Id * @param message 消息内容 */
        public void sendMonitorMessageToUser(String resourceId, String message){ 
            List<EventMonitorServer> clientList = Lists.newArrayList();
            for(Map.Entry<String, Set<EventMonitorServer>> entry : userSocket.entrySet()){ 
                for (EventMonitorServer eventMonitorServer : entry.getValue()){ 
                    clientList.add(eventMonitorServer);
                }
            }
            for (EventMonitorServer eventMonitorServer : clientList){ 
                try { 
    
                    synchronized (eventMonitorServer.session){ 
    
                        if (eventMonitorServer.isSysadmin){ 
                            eventMonitorServer.session.getBasicRemote().sendText(message);
                        }else { 
                            if (!USER_ID_PRIVILEGE_RESOURCE_CODE.containsKey(eventMonitorServer.userId) ||
                                    !USER_ID_PRIVILEGE_RESOURCE_CODE.get(eventMonitorServer.userId).contains(resourceId)){ 
                                LOGGER.debug("local user {}has no pri {}，not send message to {}",eventMonitorServer.userId,
                                        resourceId,eventMonitorServer.session.getUserProperties().get(HTTP_SESSION_ID));
                                continue;
                            }
                            eventMonitorServer.session.getBasicRemote().sendText(message);
                        }
                    }
                }catch (Exception e){ 
                    LOGGER.error(" send to id：{},terminal: {} error",eventMonitorServer.userId,eventMonitorServer.session.getUserProperties().get(HTTP_SESSION_ID),e);
                }finally { 
    
                }
            }
        }
    
    
    }

### webSocket的url传参 ###

//司机端 的url /ui/eventMonitorServer/admin?webType=1
    //web 端的url /ui/eventMonitorServer/admin?webType=0 或者 /ui/eventMonitorServer/admin
    @ServerEndpoint(value="/ui/eventMonitorServer/{userId}",configurator = HttpSessionConfigurator.class)
    
     private void init(String userId, Session session, EndpointConfig config){ 
            HttpSession httpSession = (HttpSession)config.getUserProperties().get(HttpSession.class.getName());
            this.session = session;
            //将httpsessionid设置到websocket中
            if (httpSession!=null){ 
                this.session.getUserProperties().put(HTTP_SESSION_ID,httpSession.getId());
            }else{ 
                this.session.getUserProperties().put(HTTP_SESSION_ID,new Date().toString());
            }
            //使用queryString 获取参数 基本和http类似
            String queryString = session.getQueryString();
            //为空时 是web端
            if(queryString.isEmpty()){ 
                webType = YctmConstant.WEB_TYPE;
            }else{ 
                //司机端只能连接一个用户
                if (queryString.equals("webType=1")){ 
                    webType = YctmConstant.DRIVER_TYPE;
                }else
                { 
                    webType = YctmConstant.WEB_TYPE;
                }
            }
    
            this.userId = userId;

[Link 1]: https://en.wikipedia.org/wiki/Push_technology