目录

WebSocket是一种在单个TCP连接上进行全双工通信的协议，常用于实时通信的场景。在没有使用高层级线路协议的情况下，直接使用WebSocket是很难实现发布订阅的功能。而STOMP是在WebSocket之上提供了一个基于帧的线路格式层，STOMP客户端可以同时作为生产者和消费者两种模式。为发布订阅的功能提供了基础。

STOMP is a simple text-orientated messaging protocol. It defines an interoperable wire format so that any of the available STOMP clients can communicate with any STOMP message broker to provide easy and widespread messaging interoperability among languages and platforms (the STOMP web site has a list of STOMP client and server implementations.

文档地址：http://jmesnil.net/stomp-websocket/doc/

SpringBoot集成WebSocket非常方便，只需要简单的三个步骤：导包、配置、提供接口

  1. compile('org.springframework.boot:spring-boot-starter-websocket')

第一步：创建WebSocketConfig类，通过@EnableWebSocketMessageBroker 启用代理支持的消息传递。

第二步：重写registerStompEndpoints和configureMessageBroker方法。

第三步：注册对外可访问的stomp端点、访问方式和连接跨域设置。

第四步：配置消息代理。可设置广播模式和点对点通讯。也可以添加订阅通道的前缀。

  1. package com.itdragon.server.config
  3. import org.springframework.context.annotation.Configuration
  4. import org.springframework.messaging.simp.config.MessageBrokerRegistry
  5. import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker
  6. import org.springframework.web.socket.config.annotation.StompEndpointRegistry
  7. import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer
  9. @Configuration
  10. @EnableWebSocketMessageBroker
  11. class WebSocketConfig : WebSocketMessageBrokerConfigurer {
  13.     override fun configureMessageBroker(config: MessageBrokerRegistry) {
  14.         // 设置订阅Broker名称,/topic为广播模式
  15.         config.enableSimpleBroker("/topic")
  16.         // 设置应用程序全局目标前缀
  17.         config.setApplicationDestinationPrefixes("/itdragon")
  18.     }
  20.     override fun registerStompEndpoints(registry: StompEndpointRegistry) {
  21.         // 允许使用socketJs方式访问，访问端点为socket，并允许跨域
  22.         registry.addEndpoint("/socket").setAllowedOrigins("*").withSockJS()
  23.     }
  25. }

注意：

若使用了setApplicationDestinationPrefixes方法，则作用主要体现在@SubscribeMapping和@MessageMapping上。如控制层配置@MessageMapping(“/sendToServer”)，则客户端发送的地址是 /itdragon/sendToServer

第一步：创建WebSocket的控制层类，并注入用于发送消息的SimpMessagingTemplate。

第二步：配置通过@MessageMapping注解修饰的方法来接收客户端SEND的操作。

第三步：配置通过@SubscribeMapping注解修饰的方法来接收客户端SUBSCRIBE的操作。

第四步：配置通过@SendTo注解的方法来直接将消息推送的指定地址上。

  1. package com.itdragon.server.api.rest
  3. import org.springframework.beans.factory.annotation.Autowired
  4. import org.springframework.messaging.handler.annotation.MessageMapping
  5. import org.springframework.messaging.handler.annotation.SendTo
  6. import org.springframework.messaging.simp.SimpMessagingTemplate
  7. import org.springframework.messaging.simp.annotation.SubscribeMapping
  8. import org.springframework.stereotype.Controller
  9. import org.springframework.web.bind.annotation.RequestMapping
  10. import java.time.Instant
  12. @Controller
  13. class WebSocketController {
  15.     @Autowired
  16.     lateinit var simpMessagingTemplate: SimpMessagingTemplate
  18.     /**
  19.      * 订阅广播，服务器主动推给连接的客户端
  20.      * 通过Http请求的方式触发订阅操作
  21.      */
  22.     @RequestMapping("/subscribeTopic")
  23.     fun subscribeTopicByHttp() {
  24.         while (true) {
  25.             // 可以灵活设置成通道地址，实现发布订阅的功能
  26.             val channel = "/topic/subscribeTopic" 
  27.             simpMessagingTemplate.convertAndSend(channel, Instant.now())
  28.             Thread.sleep(10*1000)
  29.         }
  30.     }
  32.     /**
  33.      * 订阅广播，服务器主动推给连接的客户端
  34.      * 通过Websocket的subscribe操作触发订阅操作
  35.      */
  36.     @SubscribeMapping("/subscribeTopic")
  37.     fun subscribeTopicByWebSocket(): Long {
  38.         return Instant.now().toEpochMilli()
  39.     }
  41.     /**
  42.      * 服务端接收客户端发送的消息，类似OnMessage方法
  43.      */
  44.     @MessageMapping("/sendToServer")
  45.     fun handleMessage(message: String) {
  46.         println("message:{$message}")
  47.     }
  49.     /**
  50.      * 将客户端发送的消息广播出去
  51.      */
  52.     @MessageMapping("/sendToTopic")
  53.     @SendTo("/topic/subscribeTopic")
  54.     fun sendToTopic(message: String): String {
  55.         return message
  56.     }
  58. }

WebSocket的订阅功能，可以用@SubscribeMapping注解，也可以用HTTP的方式触发。ITDragon龙 比较倾向HTTP的方式，因为在实现身份验证的功能上会比较方便。在客户端发送订阅操作之前，先发送HTTP请求做身份验证，验证成功后再返回指定的订阅通道地址。

在做消息通道对接的测试中，最常见的对话就是：连上了吗？没连上；收到了吗？没收到；收到了吗？收到了，后端报错……. 作为技术人员，我们有必要对各个领域的知识都有一定的了解。只有清楚明白了前端和移动端的开发思维，我们才能提供更合适的接口。

  1. <!DOCTYPE html>
  2. <html>
  3. <head>
  4. <meta charset="UTF-8" />
  5. <title>WebSocket 发布订阅</title>
  6. <link rel="stylesheet" type="text/css" href="https://cdn.staticfile.org/antd/3.23.6/antd.min.css">
  7. <script src="https://cdn.bootcss.com/sockjs-client/1.1.4/sockjs.min.js"></script>
  8. <script src="https://cdn.bootcss.com/stomp.js/2.3.3/stomp.min.js"></script>
  9. <script src="https://cdn.bootcss.com/jquery/3.3.1/jquery.min.js"></script>
  10. </head>
  11. <body>
  12.     <h2 id="connStatus"></h2>
  13.     <div style="padding-left:20px;">
  14.         <p>/topic/subscribeTopic 订阅广播通道；/sendToServer 向服务端推送消息；/sendToTopic 将消息广播出去</p>
  15.         <div>
  16.             <label>订阅通道: </label> <input type="text" id="subscribeChannel" value=""/><br>
  17.             <label>推送通道: </label> <input type="text" id="sendChannel" value=""/><br>
  18.             <button id="subscribe" onclick="subscribe()">订阅</button>
  19.             <button id="connect" onclick="connect()">连接</button>
  20.             <button id="disconnect"  onclick="disconnect();">断开连接</button>
  21.         </div>
  22.         <br>
  23.         <div>
  24.             <label>用户名称: </label> <input type="text" id="name"/><br>
  25.             <label>输入消息: </label> <input type="text" id="message"/><br>
  26.             <button id="send" onclick="sendMsg();">发送</button>
  27.             <p id="response"></p>
  28.         </div>
  29.     </div>
  30.     
  31.     <script type="text/javascript">
  32.         var stompClient = null;
  33.         var host="http://localhost:8809";
  34.         function subscribe() {
  35.             $.get(host+'/subscribeTopic');
  36.         }
  37.         
  38.         function connect() {
  39.             var socket = new SockJS(host+'/socket');
  40.             stompClient = Stomp.over(socket);
  41.             stompClient.connect({}, function(frame) {
  42.                 $('#connStatus').html('Connected:' + frame)
  43.                 stompClient.subscribe($('#subscribeChannel').val(), function(response) {
  44.                     $('#response').html(response.body);
  45.                 });
  46.             },function(err){
  47.                 console.log(err)
  48.             });
  49.         }
  50.         
  51.         function disconnect() {
  52.             if (stompClient != null) {
  53.                 stompClient.disconnect();
  54.             }
  55.             $('#connStatus').html('Disconnected')
  56.         }
  57.         
  58.         function sendMsg() {
  59.             var message = $('#message').val();
  60.             stompClient.send($('#sendChannel').val(), {}, message);
  61.         }
  62.     </script>
  63. </body>
  64. </html>

简单测试了发布和订阅功能

GIF

有的特殊场景需要检测WebSocket的生命周期，还是会用到原生的WebSocket配置，这里记录一下对应的坑。

在任意一个配置类中添加ServerEndpointExporter的Bean配置

  1. @Bean
  2. fun serverEndpointExporter(): ServerEndpointExporter {
  3.     return ServerEndpointExporter()
  4. }

问题：

  • 1）添加后单元测试启动失败，服务可以正常启动。网上说可以移除代码，由SpringBoot管理。可是移除后websocket链接会出现问题。解决方法目前未找到。

第一步：通过@ServerEndpoint注解修饰类，表示该类是WebSocket的Server，并对外暴露连接地址。

第二步：通过@OnOpen、@OnClose、@OnMessage、@OnError注解修饰方法，监控WebSocket的生命周期。

第三步：通过静态、私有、ConcurrentHashMap 修饰的变量管理客户端。

第四步：为程序其他类提供发送消息的方法。

  1. package com.itdragon.server.config
  3. import org.slf4j.LoggerFactory
  4. import org.springframework.stereotype.Component
  5. import java.io.IOException
  6. import java.util.concurrent.ConcurrentHashMap
  7. import javax.websocket.*
  8. import javax.websocket.server.PathParam
  9. import javax.websocket.server.ServerEndpoint
  11. @ServerEndpoint("/nativeSocket/{clientKey}")
  12. @Service
  13. class WebSocketServer {
  15.     private var logger = LoggerFactory.getLogger(WebSocketServer::class.java)
  16.     private var session: Session? = null
  17.     private var clientKey = ""
  19.     @OnOpen
  20.     fun onOpen(session: Session, @PathParam("clientKey") clientKey: String) {
  21.         this.session = session
  22.         this.clientKey = clientKey
  23.         if (webSocketMap.containsKey(clientKey)) {
  24.             webSocketMap.remove(clientKey)
  25.             webSocketMap[clientKey] = this
  26.         } else {
  27.             webSocketMap[clientKey] = this
  28.         }
  29.         logger.info("客户端:$clientKey 连接成功")
  31.     }
  33.     @OnClose
  34.     fun onClose() {
  35.         if (webSocketMap.containsKey(clientKey)) {
  36.             webSocketMap.remove(clientKey)
  37.         }
  38.         logger.warn("客户端:$clientKey 连接关闭")
  40.     }
  42.     @OnMessage
  43.     fun onMessage(message: String, session: Session) {
  44.         logger.info("客户端:$clientKey 收到消息:$message")
  45.     }
  47.     @OnError
  48.     fun onError(session: Session, error: Throwable) {
  49.         logger.error("WebSocket客户端(${this.clientKey})错误: ${error.message}")
  50.     }
  52.     @Throws(IOException::class)
  53.     fun sendMessage(message: String) {
  54.         this.session!!.basicRemote.sendText(message)
  55.     }
  57.     companion object {
  58.         private val webSocketMap = ConcurrentHashMap<String, WebSocketServer>()
  60.         @Throws(IOException::class)
  61.         fun sendMessage(clientKey: String, message: String) {
  62.             webSocketMap[clientKey]?.sendMessage(message)
  63.         }
  65.         fun getStatus(clientKey: String): Boolean? {
  66.             return webSocketMap[clientKey]?.session?.isOpen
  67.         }
  69.     }
  70. }

问题：

  1. <!DOCTYPE html>
  2. <html>
  3. <head>
  4.     <meta charset="UTF-8" />
  5.     <title>WebSocket 简单通讯</title>
  6.     <link rel="stylesheet" type="text/css" href="https://cdn.staticfile.org/antd/3.23.6/antd.min.css">
  7.     <script src="https://cdn.bootcss.com/jquery/3.3.1/jquery.js"></script>
  8. </head>
  9. <body>
  10. <div>
  11.     <label>订阅通道: </label> <input type="text" id="clientId" value=""/><br>
  12.     <label>推送消息: </label> <input type="text" id="message" value=""/><br>
  13.     <button id="subscribe" onclick="openSocket()">开启socket</button>
  14.     <button id="connect" onclick="sendMessage()">发送消息</button>
  15.     <button id="disconnect"  onclick="closeSocket();">关闭socket</button>
  16. </div>
  17. <script>
  18.     var socket;
  19.     function openSocket() {
  20.         if(typeof(WebSocket) == "undefined") {
  21.             console.log("您的浏览器不支持WebSocket");
  22.         }else{
  23.             console.log("您的浏览器支持WebSocket");
  24.             var socketUrl="ws://localhost:8809/nativeSocket/"+$("#clientId").val();
  25.             if(socket!=null){
  26.                 socket.close();
  27.                 socket=null;
  28.             }
  29.             socket = new WebSocket(socketUrl);
  30.             socket.onopen = function() {
  31.                 console.log("websocket已打开");
  32.             };
  33.             socket.onmessage = function(msg) {
  34.                 console.log(msg.data);
  35.             };
  36.             socket.onclose = function() {
  37.                 console.log("websocket已关闭");
  38.             };
  39.             socket.onerror = function() {
  40.                 console.log("websocket发生了错误");
  41.             }
  42.         }
  43.     }
  44.     function sendMessage() {
  45.         if(socket!=null){
  46.             socket.send($("#message").val());
  47.         }
  48.     }
  49.     function closeSocket() {
  50.         socket.close();
  51.     }
  52. </script>
  53. </body>
  54. </html>

通过stomp客户端发起的认证操作可以看一下这篇文章：https://www.cnblogs.com/jmcui/p/8999998.html

版权声明：本文为itdragon原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
本文链接：https://www.cnblogs.com/itdragon/p/12442980.html