SpringBoot+Stomp实现WebSocket消息推送

- 日理万妓 2022-05-18 10:29 827阅读 0赞

一、Stomp概念

STOMP是在WebSocket之上提供了一个基于帧的线路格式层，用于定义消息的语义。 比起原生WebSocket，稳定性和功能性都好得多。

原理可参考：[https://blog.csdn.net/a617137379/article/details/78765025][https_blog.csdn.net_a617137379_article_details_78765025]  
STOMP帧由命令、一个或多个头信息以及负载所组成！举例发送数据的一个STOMP帧：

SEND
    destination:/app/sendTest
    content-length:23
    
    {"name":"asdfsadfsadf"}

*  这里STOMP的命令是SEND，后面接发送的目标地址，消息内容长度，然后是一个空行，最后是发送内容，这个里面是一个JSON消息。
 *  这里需要注意的是destination，目标地址，消息会发送到这个目的地，这个目的地有服务端组件来进行处理。
 *  Spring使用STOMP需要进行配置，并且Spring为STOMP消息提供了基于SpringMVC的编程模型！

二、项目搭建（github项目：[https://github.com/suncht/sun-test/tree/master/springboot.websocket.test][https_github.com_suncht_sun-test_tree_master_springboot.websocket.test]）

1、SpringBoot的POM依赖：

<dependency>
    		    <groupId>org.springframework.boot</groupId>
    		    <artifactId>spring-boot-starter-websocket</artifactId>
    		</dependency>
    
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-test</artifactId>
    			<scope>test</scope>
    		</dependency>
    
    		<dependency>
    			<groupId>com.fasterxml.jackson.core</groupId>
    			<artifactId>jackson-core</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>com.fasterxml.jackson.core</groupId>
    			<artifactId>jackson-databind</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>com.fasterxml.jackson.datatype</groupId>
    			<artifactId>jackson-datatype-joda</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>com.fasterxml.jackson.module</groupId>
    			<artifactId>jackson-module-parameter-names</artifactId>
    		</dependency>

2、配置Stomp：

package com.sample.suncht.config;
    
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.messaging.simp.config.ChannelRegistration;
    import org.springframework.messaging.simp.config.MessageBrokerRegistry;
    import org.springframework.web.socket.config.annotation.AbstractWebSocketMessageBrokerConfigurer;
    import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
    import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
    import org.springframework.web.socket.config.annotation.WebSocketTransportRegistration;
    
    import com.sample.suncht.websocket.HttpSessionIdHandshakeInterceptor;
    import com.sample.suncht.websocket.PresenceChannelInterceptor;
    
    /**
     * 
     * @ClassName: WebSocketStompConfig
     * @Description: springboot websocket stomp配置 
     * 参考：
     * 	https://docs.spring.io/spring/docs/4.0.1.RELEASE/spring-framework-reference/html/websocket.html
     *	https://docs.spring.io/spring/docs/current/spring-framework-reference/web.html#websocket-fallback-sockjs-client
     */
    
    @Configuration
    @EnableWebSocketMessageBroker
    public class WebSocketStompConfig extends AbstractWebSocketMessageBrokerConfigurer {
    
    	/**
    	 * 注册stomp的端点（必须）
    	 */
    	@Override
    	public void registerStompEndpoints(StompEndpointRegistry registry) {
    		// 允许使用socketJs方式访问，访问点为webSocketServer
    		// 在网页上我们就可以通过这个链接 http://localhost:8080/webSocketServer 来和服务器的WebSocket连接
    		registry.addEndpoint("/webSocketServer").withSockJS();
    	}
    
    	/**
    	 * 配置信息代理（必须）
    	 */
    	@Override
    	public void configureMessageBroker(MessageBrokerRegistry registry) {
    		// 订阅Broker名称
    		registry.enableSimpleBroker("/queue", "/topic");
    		// 全局使用的消息前缀（客户端订阅路径上会体现出来）
    		registry.setApplicationDestinationPrefixes("/ms");
    		// 点对点使用的订阅前缀（客户端订阅路径上会体现出来），不设置的话，默认也是/user/
    		// registry.setUserDestinationPrefix("/user/");
    	}
    
    	/**
    	 * 消息传输参数配置（可选）
    	 */
    	@Override
    	public void configureWebSocketTransport(WebSocketTransportRegistration registry) {
    		registry.setMessageSizeLimit(8192) //设置消息字节数大小
    				.setSendBufferSizeLimit(8192)//设置消息缓存大小
    				.setSendTimeLimit(10000); //设置消息发送时间限制毫秒
    	}
    
    	/**
    	 * 输入通道参数设置（可选）
    	 */
    	@Override
    	public void configureClientInboundChannel(ChannelRegistration registration) {
    		registration.taskExecutor().corePoolSize(4) //设置消息输入通道的线程池线程数
    				.maxPoolSize(8)//最大线程数
    				.keepAliveSeconds(60);//线程活动时间
    		registration.setInterceptors(presenceChannelInterceptor());
    	}
    
    	/**
    	 * 输出通道参数设置（可选）
    	 */
    	@Override
    	public void configureClientOutboundChannel(ChannelRegistration registration) {
    		registration.taskExecutor().corePoolSize(4).maxPoolSize(8);
    		registration.setInterceptors(presenceChannelInterceptor());
    	}
    
    	@Bean
    	public HttpSessionIdHandshakeInterceptor httpSessionIdHandshakeInterceptor() {
    		return new HttpSessionIdHandshakeInterceptor();
    	}
    
    	@Bean
    	public PresenceChannelInterceptor presenceChannelInterceptor() {
    		return new PresenceChannelInterceptor();
    	}
    
    }

注意：

（1）注册stomp的端点 registry.addEndpoint("/webSocketServer").withSockJS();  没有添加.setAllowedOrigins("\*")允许跨域

sockjs1.0版本以上必须要setAllowedOrigins("\*")，否则报错： http://localhost:8112/endpointWisely/info?t=1510914882520 的远程资源。（原因：CORS 头缺少 'Access-Control-Allow-Origin'）

sockjs0.3.4版本可以不需要setAllowedOrigins("\*")，也就是不会允许跨域。

本人一项目一开始使用sockjs1.1.4版本，也设置setAllowedOrigins("\*")，但是还是一直上述报错，弄了一整天都没搞定，可能是跟开发环境配置有关吧。最后使用sockjs0.3.4版本就可以了。

（2）上述配置是Spring代码方式配置， 也可以使用XML配置，如下：

<websocket:message-broker application-destination-prefix="/app">
            <websocket:transport message-size="131072" send-timeout="1000" send-buffer-size="8192"/>
    
            <websocket:stomp-endpoint path="/webSocketServer">
                <websocket:handshake-interceptors>
                    <bean class="com.sample.suncht.websocket.HttpSessionIdHandshakeInterceptor"/>
                </websocket:handshake-interceptors>
                <websocket:sockjs/>
            </websocket:stomp-endpoint>
    
            <websocket:simple-broker prefix="/topic,/queue"/>
        </websocket:message-broker>

3、Controller：

package com.sample.suncht.controller;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.messaging.handler.annotation.MessageMapping;
    import org.springframework.messaging.handler.annotation.SendTo;
    import org.springframework.messaging.simp.annotation.SubscribeMapping;
    import org.springframework.stereotype.Controller;
    
    import com.sample.suncht.model.ClientMessage;
    import com.sample.suncht.model.ServerMessage;
    
    @Controller
    public class WebSocketController {
    
    	private Logger logger = LoggerFactory.getLogger(this.getClass());
    
    	@Autowired
        private SimpMessagingTemplate messagingTemplate;
    	
    	@MessageMapping("/sendTest")
    	@SendTo("/topic/subscribeTest")
    	public ServerMessage sendDemo(ClientMessage message) {
    		logger.info("接收到了信息" + message.getName());
    		return new ServerMessage("你发送的服务返回消息为:" + message.getName());
    	}
    
    	@SubscribeMapping("/subscribeTest")
    	public ServerMessage sub() {
    		logger.info("XXX用户订阅了我。。。");
    		return new ServerMessage("感谢你订阅了我。。。");
    	}
    
    	@RequestMapping("/startStomp.do")
        @ResponseBody
        public String startStomp() {
            final int counter = 10;
            MoreExecutors.newDirectExecutorService().submit(() -> {
                int index = 0;
                while (index++ < counter) {
                    messagingTemplate.convertAndSend("/topic/subscribeTest", new ServerMessage("服务器主动推的数据["+index+"] : " + DateUtils.simpleFormat(new Date())));
                    try {
                        Thread.sleep(RandomUtils.nextInt(0, 3000));
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
    
            return "ok";
        }
    }

4、前端Html：

<!DOCTYPE html>
    <html lang="zh-CN">
    
    <head>
    	<meta charset="UTF-8">
    	<meta http-equiv="content-type" content="text/html; charset=utf-8">
    	<title>stomp</title>
    </head>
    
    <body>
    	Welcome
    	<br />
    	<input id="text" type="text" />
    	<button onclick="send()">发送消息</button>
    	<button onclick="subscribe1()">订阅消息/topic/subscribeTest</button>
    	<hr />
    	<button onclick="closeWebSocket()">关闭WebSocket连接</button>
    	<hr />
    	<div id="message"></div>
    </body>
    
    <script src="/static/stomp.js/stomp.min.js"></script>
    <script src="/static/sockjs-client/0.3.4/sockjs.min.js"></script>
    <script src="/static/WebsocketProxy.js"></script>
    <script type="text/javascript">
    	var socket = new WebsocketProxy("/webSocketServer");
    	socket.connect({}, function connectCallback(frame) {
    		// 连接成功时（服务器响应 CONNECTED 帧）的回调方法
    		setMessageInnerHTML("连接成功");
    		subscribe1();
    	}, function errorCallBack(error) {
    		// 连接失败时（服务器响应 ERROR 帧）的回调方法
    		setMessageInnerHTML("连接失败");
    	});
    
    	//发送消息
    	function send() {
    		var message = document.getElementById('text').value;
    		var messageJson = JSON.stringify({
    			"name" : message
    		});
    		socket.send("/ms/sendTest", {}, messageJson);
    		//stompClient.send("/ms/sendTest", {}, messageJson);
    		//setMessageInnerHTML("/ms/sendTest 你发送的消息:" + message);
    	}
    
    	//订阅消息
    	function subscribe1() {
    		socket.subscribe('/topic/subscribeTest', function(response) {
    			setMessageInnerHTML("已成功订阅/topic/subscribeTest");
    			var returnData = JSON.parse(response.body);
    			setMessageInnerHTML("/topic/subscribeTest 你接收到的消息为:" + returnData.responseMessage);
    		});
    	}
    	
    
    	//将消息显示在网页上
    	function setMessageInnerHTML(innerHTML) {
    		document.getElementById('message').innerHTML += innerHTML + '<br/>';
    	}
    </script>
    
    </html>

WebsocketProxy.js封装了Stomp实现：

[https_blog.csdn.net_a617137379_article_details_78765025]: https://blog.csdn.net/a617137379/article/details/78765025
[https_github.com_suncht_sun-test_tree_master_springboot.websocket.test]: https://github.com/suncht/sun-test/tree/master/springboot.websocket.test