当前位置：主页 > 查看内容

SpringBoot使用WebSocket实现即时消息

发布时间：2021-04-16 00:00| 有 位朋友查看

简介：环境：SpringBoot2.3.9.RELEASE 依赖 dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-websocket/artif……

环境：SpringBoot2.3.9.RELEASE

依赖

  1. <dependency> 
  2.         <groupId>org.springframework.boot</groupId> 
  3.         <artifactId>spring-boot-starter-web</artifactId> 
  4. </dependency> 
  5. <dependency> 
  6.         <groupId>org.springframework.boot</groupId> 
  7.         <artifactId>spring-boot-starter-websocket</artifactId> 
  8. </dependency> 

定义消息类型

抽象消息对象

  1. public class AbstractMessage { 
  2.     /** 
  3.      *  消息类型 
  4.      */ 
  5.     protected String type ; 
  6.      
  7.     /** 
  8.      *  消息内容 
  9.      */ 
  10.     protected String content ; 
  11.     /** 
  12.      *  消息日期 
  13.      */ 
  14.     protected String date ; 
  15. } 

消息对象子类

1、Ping检查消息

  1. public class PingMessage extends AbstractMessage { 
  2.     public PingMessage() {} 
  3.     public PingMessage(String type) { 
  4.         this.type = type ; 
  5.     } 
  6. } 

2、系统消息

  1. public class SystemMessage extends AbstractMessage { 
  2.     public SystemMessage() {} 
  3.     public SystemMessage(String type, String content) { 
  4.         this.type = type ; 
  5.         this.content = content ; 
  6.     } 
  7. } 

3、点对点消息

  1. public class PersonMessage extends AbstractMessage { 
  2.     private String fromName ; 
  3.     private String toName ; 
  4. } 

消息类型定义

  1. public enum MessageType { 
  2.      
  3.     /** 
  4.      *  系统消息 0000;心跳检查消息 0001;点对点消息2001 
  5.      */ 
  6.     SYSTEM("0000"), PING("0001"), PERSON("2001") ; 
  7.      
  8.     private String type ; 
  9.      
  10.     private MessageType(String type) { 
  11.         this.type = type ; 
  12.     } 
  13.  
  14.     public String getType() { 
  15.         return type; 
  16.     } 
  17.  
  18.     public void setType(String type) { 
  19.         this.type = type; 
  20.     } 
  21.      
  22. } 

 WebSocket服务端点

该类作用就是定义客户端连接的地址

  1. @ServerEndpoint(value = "/message/{username}",  
  2.     encoders = {WsMessageEncoder.class}, 
  3.     decoders = {WsMessageDecoder.class}, 
  4.     subprotocols = {"gmsg"}, 
  5.     configurator = MessageConfigurator.class)   
  6. @Component   
  7. public class GMessageListener {   
  8.    
  9.     public static ConcurrentMap<String, UserSession> sessions = new ConcurrentHashMap<>(); 
  10.     private static Logger logger = LoggerFactory.getLogger(GMessageListener.class) ; 
  11.    
  12.     private String username ; 
  13.      
  14.     @OnOpen   
  15.     public void onOpen(Session session, EndpointConfig config, @PathParam("username") String username){ 
  16.         UserSession userSession = new UserSession(session.getId(), username, session) ; 
  17.         this.username = username ; 
  18.         sessions.put(username, userSession) ; 
  19.         logger.info("【{}】用户进入, 当前连接数：{}", username, sessions.size()) ;  
  20.     }   
  21.    
  22.     @OnClose   
  23.     public void onClose(Session session, CloseReason reason){   
  24.         UserSession userSession = sessions.remove(this.username) ; 
  25.         if (userSession != null) { 
  26.             logger.info("用户【{}】, 断开连接, 当前连接数：{}", username, sessions.size()) ; 
  27.         } 
  28.     } 
  29.      
  30.     @OnMessage 
  31.     public void pongMessage(Session session, PongMessage message) { 
  32.         ByteBuffer buffer = message.getApplicationData() ; 
  33.         logger.debug("接受到Pong帧【这是由浏览器发送】：" + buffer.toString()); 
  34.     } 
  35.      
  36.     @OnMessage 
  37.     public void onMessage(Session session, AbstractMessage message) { 
  38.         if (message instanceof PingMessage) { 
  39.             logger.debug("这里是ping消息"); 
  40.             return ; 
  41.         } 
  42.         if (message instanceof PersonMessage) { 
  43.             PersonMessage personMessage = (PersonMessage) message ; 
  44.             if (this.username.equals(personMessage.getToName())) { 
  45.                 logger.info("【{}】收到消息：{}", this.username, personMessage.getContent()); 
  46.             } else { 
  47.                 UserSession userSession = sessions.get(personMessage.getToName()) ; 
  48.                 if (userSession != null) { 
  49.                     try { 
  50.                         userSession.getSession().getAsyncRemote().sendText(new ObjectMapper().writeValueAsString(message)) ; 
  51.                     } catch (JsonProcessingException e) { 
  52.                         e.printStackTrace(); 
  53.                     } 
  54.                 } 
  55.             } 
  56.             return ; 
  57.         } 
  58.         if (message instanceof SystemMessage) { 
  59.             logger.info("接受到消息类型为【系统消息】") ;  
  60.             return ; 
  61.         } 
  62.     } 
  63.      
  64.     @OnError 
  65.     public void onError(Session session, Throwable error) { 
  66.         logger.error(error.getMessage()) ; 
  67.     } 
  68. } 

WsMessageEncoder.java类

该类的主要作用是，当发送的消息是对象时，该如何转换

  1. public class WsMessageEncoder implements Encoder.Text<AbstractMessage> { 
  2.     private static Logger logger = LoggerFactory.getLogger(WsMessageDecoder.class) ; 
  3.     @Override 
  4.     public void init(EndpointConfig endpointConfig) { 
  5.     } 
  6.     @Override 
  7.     public void destroy() { 
  8.     } 
  9.     @Override 
  10.     public String encode(AbstractMessage tm) throws EncodeException { 
  11.         String message = null ; 
  12.         try { 
  13.             message = new ObjectMapper().writeValueAsString(tm); 
  14.         } catch (JsonProcessingException e) { 
  15.             logger.error("JSON处理错误：{}", e) ; 
  16.         } 
  17.         return message; 
  18.     } 
  19. } 

WsMessageDecoder.java类

该类的作用是，当接收到消息时如何转换成对象。

  1. public class WsMessageDecoder implements  Decoder.Text<AbstractMessage> { 
  2.  
  3.     private static Logger logger = LoggerFactory.getLogger(WsMessageDecoder.class) ; 
  4.     private static Set<String> msgTypes = new HashSet<>() ; 
  5.      
  6.     static { 
  7.         msgTypes.add(MessageType.PING.getType()) ; 
  8.         msgTypes.add(MessageType.SYSTEM.getType()) ; 
  9.         msgTypes.add(MessageType.PERSON.getType()) ; 
  10.     } 
  11.     @Override 
  12.     @SuppressWarnings("unchecked") 
  13.     public AbstractMessage decode(String s) throws DecodeException { 
  14.         AbstractMessage message = null ; 
  15.         try { 
  16.             ObjectMapper mapper = new ObjectMapper() ; 
  17.             Map<String,String> map = mapper.readValue(s, Map.class) ; 
  18.             String type = map.get("type") ; 
  19.             switch(type) { 
  20.                 case "0000": 
  21.                     message = mapper.readValue(s, SystemMessage.class) ; 
  22.                     break; 
  23.                 case "0001": 
  24.                     message = mapper.readValue(s, PingMessage.class) ; 
  25.                     break; 
  26.                 case "2001": 
  27.                     message = mapper.readValue(s, PersonMessage.class) ; 
  28.                     break; 
  29.             } 
  30.         } catch (JsonProcessingException e) { 
  31.             logger.error("JSON处理错误：{}", e) ; 
  32.         } 
  33.         return message ; 
  34.     } 
  35.  
  36.     // 该方法判断消息是否可以被解码（转换） 
  37.     @Override 
  38.     @SuppressWarnings("unchecked") 
  39.     public boolean willDecode(String s) { 
  40.         Map<String, String> map = new HashMap<>() ; 
  41.         try { 
  42.             map = new ObjectMapper().readValue(s, Map.class); 
  43.         } catch (JsonProcessingException e) { 
  44.             e.printStackTrace(); 
  45.         } 
  46.         logger.debug("检查消息：【" + s + "】是否可以解码") ; 
  47.         String type = map.get("type") ; 
  48.         if (StringUtils.isEmpty(type) || !msgTypes.contains(type)) { 
  49.             return false ; 
  50.         } 
  51.         return true ; 
  52.     } 
  53.     @Override 
  54.     public void init(EndpointConfig endpointConfig) { 
  55.     } 
  56.     @Override 
  57.     public void destroy() { 
  58.     } 
  59. } 

MessageConfigurator.java类

该类的作用是配置服务端点，比如配置握手信息

  1. public class MessageConfigurator extends ServerEndpointConfig.Configurator { 
  2.     private static Logger logger = LoggerFactory.getLogger(MessageConfigurator.class) ; 
  3.     @Override 
  4.     public void modifyHandshake(ServerEndpointConfig sec, HandshakeRequest request, HandshakeResponse response) { 
  5.         logger.debug("握手请求头信息：" + request.getHeaders()); 
  6.         logger.debug("握手响应头信息：" + response.getHeaders()); 
  7.         super.modifyHandshake(sec, request, response); 
  8.     }    
  9. } 

WebSocke配置类

  1. @Configuration 
  2. public class WebSocketConfig { 
  3.      
  4.     @Bean 
  5.     public ServerEndpointExporter serverEndpointExporter (){   
  6.         return new ServerEndpointExporter();   
  7.     }   
  8.      
  9. } 

当以jar包形式运行时需要配置该bean，暴露我们配置的@ServerEndpoint;当我们以war独立tomcat运行时不能配置该bean。

前端页面

  1. <!doctype html> 
  2. <html> 
  3.  <head> 
  4.   <meta charset="UTF-8"> 
  5.   <meta name="Author" content=""> 
  6.   <meta name="Keywords" content=""> 
  7.   <meta name="Description" content=""> 
  8.   <script src="g-messages.js?v=1"></script> 
  9.   <title>WebSocket</title> 
  10.   <style type="text/css"> 
  11.   </style> 
  12.   <script> 
  13.     let gm = null ; 
  14.     let username = null ; 
  15.     function ListenerMsg({url, protocols = ['gmsg'], options = {}}) { 
  16.         if (!url){  
  17.             throw new Error("未知服务地址") ; 
  18.         } 
  19.         gm = new window.__GM({ 
  20.             url: url, 
  21.             protocols: protocols 
  22.         }) ; 
  23.         gm.open(options) ; 
  24.     } 
  25.     ListenerMsg.init = (user) => { 
  26.         if (!user) { 
  27.             alert("未知的当前登录人") ; 
  28.             return ; 
  29.         } 
  30.         let url = `ws://localhost:8080/message/${user}` ; 
  31.         let msg = document.querySelector("#msg") 
  32.         ListenerMsg({url, options: { 
  33.             onmessage (e) { 
  34.                 let data = JSON.parse(e.data) ; 
  35.                 let li = document.createElement("li") ; 
  36.                 li.innerHTML = "【" + data.fromName + "】对你说：" + data.content ; 
  37.                 msg.appendChild(li) ; 
  38.             } 
  39.         }}) ; 
  40.     } 
  41.     function enter() { 
  42.         username = document.querySelector("#nick").value ; 
  43.         ListenerMsg.init(username) ; 
  44.         document.querySelector("#chat").style.display = "block" ; 
  45.         document.querySelector("#enter").style.display = "none" ; 
  46.         document.querySelector("#cu").innerText = username ; 
  47.     } 
  48.     function send() { 
  49.         let a = document.querySelector("#toname") ; 
  50.         let b = document.querySelector("#content") ; 
  51.         let toName = a.value ; 
  52.         let content = b.value ; 
  53.         gm.sendMessage({type: "2001", content, fromName: username, toName}) ; 
  54.         a.value = '' ; 
  55.         b.value = '' ; 
  56.     } 
  57.   </script> 
  58.  </head> 
  59.  <body> 
  60.     <div id="enter"> 
  61.         <input id="nick"/><button type="button" onclick="enter()">进入</button> 
  62.     </div> 
  63.     <hr/> 
  64.     <div id="chat" style="display:none;"> 
  65.         当前用户：<b id="cu"></b><br/> 
  66.         用户：<input id="toname" name="toname"/><br/><br/> 
  67.         内容：<textarea id="content" rows="3" cols="22"></textarea><br/> 
  68.         <button type="button" onclick="send()">发送</button> 
  69.     </div> 
  70.     <div> 
  71.         <ul id="msg"> 
  72.         </ul> 
  73.     </div> 
  74.  </body> 
  75. </html> 

到此所有的代码完毕，接下来测试

测试

打开两个标签页，以不同的用户进入。

输入对方用户名发送消息

成功了，简单的websocket。我们生产环境还就这么完的，8g内存跑了6w的用户。

完毕!!!


本文转载自网络，原文链接：https://www.toutiao.com/i6942648061537780257/
本站部分内容转载于网络，版权归原作者所有，转载之目的在于传播更多优秀技术内容，如有侵权请联系QQ/微信：153890879删除，谢谢！
上一篇：针对SREs和DevOps的十大开源项目 下一篇：【玩转腾讯云】事件驱动编程

推荐图文

  • 周排行
  • 月排行
  • 总排行

随机推荐