SpringBoot集成websocket

1. 前言

​ 这里我们使用springboot搭建一个轻量级的WEBSocket服务，同时提供4个入参。使用websocket服务可以轻松和微信小程序、支付宝小程序、网页就行双向通讯，非常实用方便。

1. wss地址
   这里是我们自己搭建的中转服务websocket地址。

   wss://xxxx.cn/netgate/auth/pid/sn/openid

2. 参数说明

   参数名称	参数说明
   auth	自定义标识符，用于鉴权处理
   pid	设备产品ID【设备二维码中包含此信息】
   sn	设备序列号【设备二维码中包含此信息】
   openid	用户自己的id。可以使用个人平台的用户唯一id。相同ID会挤掉前一个用户的ws连接

2. 引入依赖

      org.springframework.boot     spring-boot-starter-websocket 

3. 配置文件

​ 新增WebSocketConfig.java的配置文件。用来初始化websocket

@Configuration@EnableWebSocketpublic class WebSocketConfig implements WebSocketConfigurer{    @Override    public void reGISterWebSocketHandlers(WebSocketHandlerRegistry registry) {       registry.addHandler(webSocketHandler(),"/netgate/{auth}/{pid}/{sn}/{openid}")                          //注册Handler           .addInterceptors(new WebSocketHandshakeInterceptor())           //注册Interceptor           .setAllowedOrigins("*");     //注册Interceptor//     //2.注册Sockjs，提供SockJS支持(主要是兼容ie8)//     String sockjs_url = "/sockjs/socketServer.do";                          //设置sockjs的地址//     registry.addHandler(netgateHandler, sockjs_url)                         //注册Handler//         .addInterceptors(new WebSocketHandshakeInterceptor())               //注册Interceptor//         .withSockJS();                          //支持sockjs协议    }    @Bean    public ServletServerContainerFactoryBean createWebSocketContainer() {        ServletServerContainerFactoryBean container = new ServletServerContainerFactoryBean();        container.setMaxTextMessageBufferSize(2*1024*1024);//8192*1024 1024*1024*1024        container.setMaxBinaryMessageBufferSize(2*1024*1024);        container.setAsyncSendTimeout(55000l);        container.setMaxSessionIdleTimeout(55000l);//心跳        return container;    }    @Bean    public TextWebSocketHandler webSocketHandler() {        return new NetgateHandler();    }}

4. Websocket握手过滤器

​ 过滤器的作用主要是用来做连接接入的鉴权，和参数解析。

​ 新增WebSocketHandshakeInterceptor.java文件

public class WebSocketHandshakeInterceptor implements HandshakeInterceptor {    private final static Logger LOGGER = LoggerFactory.getLogger(WebSocketHandshakeInterceptor.class);     public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse serverHttpResponse, WebSocketHandler webSocketHandler, Map attributes) throws Exception {        if (request instanceof ServletServerHttpRequest) {        String path = request.getURI().getPath();        if(requestIsValid(path)){        String[] params = getParams(path);        attributes.put("WEBSOCKET_AUTH", params[0]);        attributes.put("WEBSOCKET_PID", params[1]);        attributes.put("WEBSOCKET_SN", params[2]);        attributes.put("WEBSOCKET_OPENID", params[3]);        attributes.put("WEBSOCKET_FIRSTONE","yes");        }        }        System.out.println("================Before Handshake================");        return true;    }     public void afterHandshake(ServerHttpRequest serverHttpRequest, ServerHttpResponse serverHttpResponse, WebSocketHandler webSocketHandler, Exception e) {    System.out.println("================After Handshake================");    if(e!=null) e.printStackTrace();    System.out.println("================After Handshake================");    }        private boolean requestIsValid(String url){        //在这里可以写上具体的鉴权逻辑    boolean isvalid = false;    if(StringUtils.isNotEmpty(url)    && url.startsWith("/netgate/")    && url.split("/").length==6){    isvalid = true;    }    return isvalid;    }        private String[] getParams(String url){    url = url.replace("/netgate/","");    return url.split("/");    }    }

5. Websocket处理器

​ 在这里可以做消息的接收和发送。

​ 这里MQttGateway是springboot整合MQtT客户端的服务类。具体可以参考下一篇的springboot集成mqtt

​ 新建NetgateHandler.java文件

@Componentpublic class NetgateHandler extends TextWebSocketHandler {@Autowiredprivate MqttGateway mqttGateway;        private static ConcurrentHashMap> netgates = new ConcurrentHashMap>();            @Override    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {    if(!session.isOpen()) {System.out.println("连接已关闭，不再处理该连接的消息！");return;}    String mes = ObjectUtils.toString(message.getPayload(),"");    String pid = session.getAttributes().get("WEBSOCKET_PID").toString();    String sn = session.getAttributes().get("WEBSOCKET_SN").toString();if(message==null || "".equals(mes)){System.out.println(getSysDate()+"============接收到空消息，不予处理。");}else if(mes.length()==1){//心跳消息过滤掉return;}else {//转发成mqtt消息String topic = "pay/"+pid+"/server/"+sn;System.out.println(topic);System.out.println(getSysDate()+"============消息处理完成："+mes);mqttGateway.sendToMqtt(topic,mes);}    }          @Override    public void afterConnectionEstablished(WebSocketSession session) throws Exception {    System.out.println(getSysDate()+"============正在初始化连接："+session.getId());        try {            //初始化连接，把session存储起来this.initUsers(session);} catch (Exception e) {System.out.println(getSysDate()+"============初始化连接异常-开始："+e.getMessage());e.printStackTrace();System.out.println(getSysDate()+"============初始化连接异常-结束："+e.getMessage());}        System.out.println(getSysDate()+"============初始化连接完成："+session.getId());    }         @Override    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {    System.out.println(getSysDate()+"============正在关闭连接："+session.getId()+",isOpen:"+session.isOpen()+";code："+status.getCode());    try {    System.out.println("断开连接状态值"+status.getCode());this.removeSession(session);} catch (Exception e) {System.out.println(getSysDate()+"============关闭连接异常-开始："+e.getMessage());e.printStackTrace();System.out.println(getSysDate()+"============关闭连接异常-结束："+e.getMessage());}    System.out.println(getSysDate()+"============正在关闭完成："+session.getId()+",isOpen:"+session.isOpen()+";code："+status.getCode());    }         @Override    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {    System.out.println(getSysDate()+"============发生传输错误："+session.getId()+";session.isOpen():"+session.isOpen()+";exception："+exception.getMessage());    exception.printStackTrace();    if (session.isOpen()) {        //try { session.close(); } catch (Exception e) {e.printStackTrace();}        }else {        try {    this.removeSession(session);    } catch (Exception e) {    System.out.println(getSysDate()+"============传输错误处理异常-开始："+e.getMessage());    e.printStackTrace();    System.out.println(getSysDate()+"============传输错误处理异常-结束："+e.getMessage());    }        }    System.out.println(getSysDate()+"============错误处理结束："+session.getId()+";session.isOpen():"+session.isOpen()+";exception："+exception.getMessage());    }public synchronized void sendMsgToNetgateSn(String sn, String msg)  {if(netgates.size()>0 && netgates.containsKey(sn) && !netgates.get(sn).isEmpty()){//获取EID对应的后台管理连接 多个for (WebSocketSession ws: netgates.get(sn).values()){System.out.println("对网关指令开始发送啦:sn="+sn+"消息内容"+msg);try {ws.sendMessage(new TextMessage(msg));} catch (IOException e) {System.out.println(getSysDate()+"发生了异常："+e.getMessage());e.printStackTrace();continue;}}}}    //连接接入的处理方法private synchronized void initUsers(WebSocketSession session){String pid = (String) session.getAttributes().get("WEBSOCKET_PID");String sn = (String) session.getAttributes().get("WEBSOCKET_SN");String openid = (String) session.getAttributes().get("WEBSOCKET_OPENID");if(StringUtils.isNotEmpty(pid) && StringUtils.isNotEmpty(sn) && StringUtils.isNotEmpty(openid)){ConcurrentHashMap netgate = netgates.get(sn);if(netgate == null){netgate = new ConcurrentHashMap();}WebSocketSession session_exist = netgate.get(sn);if(session_exist!=null) {System.out.println("检测到相同SN重复连接,SN:"+sn+",连接ID:"+session_exist.getId()+",准备清理失效的连接。。。");try {session_exist.close();} catch (IOException e) {e.printStackTrace();}}netgate.putIfAbsent(openid, session);netgates.put(sn,netgate);}}    //连接被关闭时处理集合private synchronized void removeSession(WebSocketSession session){String sn = (String) session.getAttributes().get("WEBSOCKET_SN");String openid = (String) session.getAttributes().get("WEBSOCKET_OPENID");if(netgates.get(sn).containsKey(openid)) {WebSocketSession exist_session = netgates.get(sn).get(openid);//确保是同一个session 不是同一个session则不应该进行下一步的处理if(exist_session.getId()!=null && exist_session.getId().equals(session.getId())) {netgates.get(sn).remove(openid);System.out.println("有一网关连接关闭！SN："+sn+",当前在线数量为"+netgates.get(sn).keySet().size());}else {System.out.println("检测到关闭session异常,程序中断处理,关闭sessionId："+session.getId()+",当前实际sessionId:"+exist_session.getId());}}else {System.out.println("检测到关闭session异常,程序中断处理,系统中未找到对应的session,Sn="+sn+"openid="+openid);}}private String getSysDate() { SimpleDateFORMat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式        return df.format(new Date());}}

6. 下一篇

springboot集成mqtt

来源地址：https://blog.csdn.net/qq_35921773/article/details/127451903