springboot集成websocket
1. 前言
这里我们使用springboot搭建一个轻量级的websocket服务，同时提供4个入参。使用websocket服务可以轻松和微信小程序、支付宝小程序、网页就行双向通讯，非常实用方便。
-
wss地址
这里是我们自己搭建的中转服务websocket地址。wss://xxxx.cn/netgate/auth/pid/sn/openid -
参数说明
参数名称 参数说明 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. 下一篇
来源地址：https://blog.csdn.net/qq_35921773/article/details/127451903
