Springboot+Websocket+Scheduled开发案例总结

案例简介

本人最近由于公司项目需求，有幸接触websocket，公司项目系统消息模板欲实现添加公告时实现用户的定向推送，分立即推送和定时推送，本人实现项目思路是前端页面在登录成功时通过后台redis验证用户身份，以用户id标识安全订阅消息推送接口，在添加系统公告成功时开启定时任务每隔30秒查询未推送的消息实现定向推送，使用Websocket，STOMP技术完成项目需求，期间由于没有heartbeats导致会话可能会过期在服务器端报错。我最终每隔20000ms从浏览器启用STOMP心跳，以保持Spring会话最后访问时间在STOMP心跳上更新而不会丢失会话。服务端我使用AbstractWebSocketMessageBrokerConfigurer作为基础来启用握手前身份认证和心跳设置。初次接触websocket，不足之处，请见解。

消息通知websocket结构图

关键技术

WebSocket介绍：

很多网站为了实现实时推送技术，所用的技术都是轮询。轮询是在特定的时间间隔（如每1秒），由浏览器对服务器发出HTTP请求，然后由服务器返回最新的数据给客户端。这种传统的模式带来的缺点很明显，即浏览器需要不断的向服务器发出请求，然而HTTP请求包含较多的请求头信息，而其中真正有效的数据只是很小的一部分，显然这样会浪费很多的带宽等资源。在这种情况下，HTML5定义了WebSocket协议，能更好的节省服务器资源和带宽，并且能够更实时地进行通讯。

WebSocket，一种在单个TCP连接上进行全双工通讯的协议，建立在 TCP 协议之上，服务器端的实现比较容易，使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。在WebSocket API中，浏览器和服务器只需要完成一次握手，两者之间就可以创建持久性的连接，并进行双向数据传输。与 HTTP 协议有着良好的兼容性。默认端口也是80和443，并且握手阶段采用 HTTP 协议，因此握手时不容易屏蔽，能通过各种 HTTP 代理服务器。同 HTTP 在 TCP 套接字上添加请求-响应模型层一样，STOMP 在 WebSocket 之上提供了一个基于帧的线路格式层，用来定义消息语义

STOMP介绍：

STOMP，面向消息的简单文本协议，WebSocket处在TCP上非常薄的一层，会将字节流转换为文本/二进制消息，因此，对于实际应用来说，WebSocket的通信形式层级过低，因此，可以在 WebSocket 之上使用 STOMP协议，来为浏览器和 server间的通信增加适当的消息语义。

SockJS介绍：

SockJS 是 WebSocket 技术的一种模拟。为了应对许多浏览器不支持WebSocket协议的问题，设计了备选

SockJs

。

案例环境要求

JDK1.8+
Spring Boot 1.5.11.RELEASE
Spring Boot-WebSocket
sockjs.js
stomp.js
jquery.js

添加pom.xml websocket 支持

<dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

在项目中添加 websocket 的 Spring 配置

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer{


	public static final Logger LOGGER = LoggerFactory.getLogger(WebSocketConfig.class);

	/**
	 * 定义一些消息连接规范（也可不设置）
	 * @param config
	 */
	@Override
	public void configureMessageBroker(MessageBrokerRegistry config) {
		//服务端发送消息给客户端的域,多个用逗号隔开
		config.enableSimpleBroker("/topic", "/user")
				.setTaskScheduler(new DefaultManagedTaskScheduler())
				.setHeartbeatValue(new long[]{20000,20000});	//设置心跳
		config.setApplicationDestinationPrefixes("/app");
	}

	/**
	 * 注册STOMP协议节点并映射url
	 * @param registry
	 */
	@Override
	public void registerStompEndpoints(StompEndpointRegistry registry) {
		registry.addEndpoint("/notify") //注册一个 /chebaba 的 websocket 节点
				.addInterceptors(notifyHandshakeInterceptor()) //添加 websocket握手拦截器
				.setHandshakeHandler(notifyDefaultHandshakeHandler()) //添加 websocket握手处理器
				.setAllowedOrigins("*")//设置允许可跨域的域名
				.withSockJS();//指定使用SockJS协议
	}


	/**
	 * 实施定制来断开会话
	 * @param registration
	 */
	@Override
	public void configureWebSocketTransport(final WebSocketTransportRegistration registration) {
		registration.addDecoratorFactory(new WebSocketHandlerDecoratorFactory() {
			@Override
			public WebSocketHandler decorate(final WebSocketHandler handler) {
				return new WebSocketHandlerDecorator(handler) {

					@Override
					public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
						LOGGER.warn("会话过程出现错误，关闭连接");
						LOGGER.warn(exception.getMessage());
						session.close();
						super.handleTransportError(session, exception);
					}
				};
			}
		});
		super.configureWebSocketTransport(registration);
	}




	/**
	 * WebSocket 握手拦截器
	 * 可做一些用户认证拦截处理
	 */
	private HandshakeInterceptor notifyHandshakeInterceptor(){
		return new HandshakeInterceptor() {
			/**
			 * websocket握手连接
			 * @return 返回是否同意握手
			 */
			@Override
			public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response,
										   WebSocketHandler wsHandler, Map<String, Object> attributes) throws Exception {
				ServletServerHttpRequest req = (ServletServerHttpRequest) request;
				//通过url的query参数获取认证参数
				String token = req.getServletRequest().getParameter("token");
				LOGGER.info("用户身份token:{}",token);
				//根据token认证用户，不通过返回拒绝握手
				Principal user = authenticate(token);
				if(user == null){
					LOGGER.warn("用户身份token:{}，身份验证错误或者用户未登录！",token);
					return false;
				}
				//保存认证用户
				attributes.put("user", user);
				return true;
			}

			@Override
			public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception exception) {
				//LOGGER.warn(exception.getMessage());
			}
		};
	}

	//WebSocket 握手处理器
	private DefaultHandshakeHandler notifyDefaultHandshakeHandler(){
		return new DefaultHandshakeHandler(){
			@Override
			protected Principal determineUser(ServerHttpRequest request, WebSocketHandler wsHandler, Map<String, Object> attributes) {
				//设置认证通过的用户到当前会话中
				return (Principal)attributes.get("user");
			}

			@Override
			protected void doStop() {
				super.doStop();
			}
		};
	}

	/**
	 * 根据token认证授权
	 * @param token
	 */
	private Principal authenticate(String token){
		//实现用户的认证并返回用户信息，如果认证失败返回 null
		LOGGER.info(token);

		if(SessionUtil.isLogin(token)){
			SessionVo sessionVo = SessionUtil.get(token);
			NoticePrincipal noticePrincipal = new NoticePrincipal();
			noticePrincipal.setName(String.valueOf(sessionVo.getStore().get("store_id")));

			return noticePrincipal;
		}

		return null;
	}




	//用户信息需继承 Principal 并实现 getName() 方法，返回全局唯一值
	class NoticePrincipal implements Principal {

		private String userId;

		@Override
		public String getName() {
			return userId;
		}

		public void setName(String userId) {
			this.userId = userId;
		}

		@Override
		public boolean equals(Object o) {
			if (this == o) return true;
			if (o == null || getClass() != o.getClass()) return false;

			NoticePrincipal that = (NoticePrincipal) o;

			return userId != null ? userId.equals(that.userId) : that.userId == null;

		}

		@Override
		public int hashCode() {
			return userId != null ? storeId.hashCode() : 0;
		}
	}

}

3. 实现连接的监听

private static final Logger LOGGER = LoggerFactory.getLogger(WebsocketEventListener.class);

	/**
	 * 静态变量，用来记录当前在线连接数。应该把它设计成线程单例的。
	 */
	private static  AtomicLong connectionIds = new AtomicLong(0);


	//concurrent包的线程安全HashMap，用来存放每个客户端对应的webSocket对象。
	private static ConcurrentHashMap<Object, Integer> concurrentHashMap = new ConcurrentHashMap<>();

	@EventListener
	public void handleWebSocketConnectedListener(SessionConnectedEvent event) {
		StompHeaderAccessor headerAccessor = StompHeaderAccessor.wrap(event.getMessage());
		if (headerAccessor.getUser() != null) {
			LOGGER.info("Received a new web socket connectionc - {}", headerAccessor);
            String storeId = headerAccessor.getUser().getName();
			//保存用户访问数量
			if (concurrentHashMap.containsKey(storeId)) {
				concurrentHashMap.put(userId, concurrentHashMap.get(userId)+1);
			} else {
				concurrentHashMap.put(userId, 1);
			}
			int count = concurrentHashMap.size();
			//在线连接数加1
			LOGGER.info("user_id为：{}的用户链接访问成功，用户访问数量为：{},当前链接数为：{}",headerAccessor.getUser().getName(),count,connectionIds.incrementAndGet());
		}

	}

	@EventListener
	public void handleWebSocketSubscribeListener(SessionSubscribeEvent event) {
		StompHeaderAccessor headerAccessor = StompHeaderAccessor.wrap(event.getMessage());
		if (headerAccessor.getUser() != null) {
			LOGGER.info("A web socket client Subscribed a topic - {}", headerAccessor);
			LOGGER.info("user_id为：{}的用户链接已订阅成功,等待消息通知",headerAccessor.getUser().getName());
		}


	}

	@EventListener
	public void handleWebSocketUnsubscribeListener(SessionUnsubscribeEvent event) {
		StompHeaderAccessor headerAccessor = StompHeaderAccessor.wrap(event.getMessage());

		if (headerAccessor.getUser() != null) {
			LOGGER.info("A web socket client unsubscribed a topic - {}", headerAccessor);
			LOGGER.info("user_id为：{}的用户链接已取消订阅成功",headerAccessor.getUser().getName());
		}

	}

	@EventListener
	public void handleWebSocketDisconnectListener(SessionDisconnectEvent event) {
		StompHeaderAccessor headerAccessor = StompHeaderAccessor.wrap(event.getMessage());


		if (headerAccessor.getUser() != null) {

			String storeId = headerAccessor.getUser().getName();
			//统计每个用户人数，为0移除
			if(null == concurrentHashMap.get(userId)){
				LOGGER.warn("计数过程出错");
				return;
			}
			if (concurrentHashMap.get(storeId) - 1 == 0) {
				concurrentHashMap.remove(storeId);
			} else {
				concurrentHashMap.put(storeId,concurrentHashMap.get(storeId) - 1);
			}

			int count = concurrentHashMap.size();
			//在线连接数减1
			LOGGER.info("user_id为：{}的用户链接断开，用户访问数量为：{},当前链接数为：{}",headerAccessor.getUser().getName(),count,connectionIds.decrementAndGet());
		}

	}

4. 前端页面示例

<!DOCTYPE html>
<html>
<head>
    <title>Hello WebSocket</title>
    <script src="https://cdn.bootcss.com/stomp.js/2.3.3/stomp.min.js"></script>
    <script src="https://cdn.bootcss.com/sockjs-client/1.1.1/sockjs.min.js"></script>
    <script type="text/javascript">

		document.addEventListener('DOMContentLoaded', function() {// request permission on page load
			if (Notification.permission !== "granted")
				Notification.requestPermission();
		});

		/**
			The public notify function
		 */
		function notifyBox(title, icon, body, url) {
			if (!Notification) {
				alert('Desktop notifications not available in your browser. Try Chromium.');
				return;
			}
			if (Notification.permission !== "granted")
				Notification.requestPermission();
			else {
				var notification = new Notification(title, {
					icon : icon,
					body : body,
				});

				notification.onclick = function() {
					window.open(url);
				};
			}
		}

        // var socket = new SockJS('http://localhost:8080/hello');
        // var stompClient = Stomp.over(socket);
		var stompClient = null;

		var userId = 'abc';

		function setConnected(connected) {
			document.getElementById('connect').disabled = connected;
			document.getElementById('disconnect').disabled = !connected;
			// document.getElementById('conversationDiv').style.visibility = connected ? 'visible'
			// 		: 'hidden';
			document.getElementById('response').innerHTML = '';
		}

        // document.getElementById('conversationDiv').style.visibility = 'visible';

		function connect() {
            var token='amber';
			var socket = new SockJS('http://localhost:8080/notify-websocket/notify?token='+token);
			stompClient = Stomp.over(socket);
            stompClient.heartbeat.outgoing = 20000;  // client will send heartbeats every 20000ms
            stompClient.heartbeat.incoming = 20000;      // client does not want to receive heartbeats from the server
            var connect_callback = function(frame) {
                // called back after the client is connected and authenticated to the STOMP server
                setConnected(true);
                console.log('Connected: ' + frame);
                
                stompClient.subscribe('/topic/notify', function(
                    msg) {
                    // showGreeting(JSON.parse(greeting.body).content);
                    notifyBox('Test',
                        'https://www.baidu.com',
                        JSON.parse(msg.body).responseMsg,
                        'https://www.baidu.com');
                });
                stompClient.subscribe('/user/' + userId + '/msg', function(
                		msg) {
                    console.log(msg);
                	//showGreeting(JSON.parse(greeting.body).content);
                	notifyBox('Test',
                			'https://www.baidu.com',
                			JSON.parse(msg.body).responseMsg,
                			'https://www.baidu.com');
                });
            };
            var error_callback = function(error) {
                // display the error's message header:
                sleep(5000); //当前方法暂停5秒
                // alert(error);
                connect();//重新连接


            };
			stompClient.connect({},connect_callback,error_callback);
		}

		function disconnect() {
			if (stompClient != null) {
				stompClient.disconnect();
			}
			setConnected(false);
			console.log("Disconnected");
		}

		function sendName() {
			var name = document.getElementById('name').value;
			stompClient.send("/app/hello", {}, JSON.stringify({
				'name' : name
			}));
		}

		function showGreeting(message) {
			var response = document.getElementById('response');
			var p = document.createElement('p');
			p.style.wordWrap = 'break-word';
			p.appendChild(document.createTextNode(message));
			response.appendChild(p);
		}

        function sleep(d){
            for(var t = Date.now();Date.now() - t <= d;);
        }

	</script>
</head>
<body onload="disconnect()">
<noscript><h2 style="color: #ff0000">Seems your browser doesn't support Javascript! Websocket relies on Javascript being enabled. Please enable
    Javascript and reload this page!</h2></noscript>
<div>
    <div>
        <button id="connect" onclick="connect();">Connect</button>
        <button id="disconnect" disabled="disabled" onclick="disconnect();">Disconnect</button>
    </div>
    <div id="conversationDiv">
        <label>What is your name?</label><input type="text" id="name" />
        <button id="sendName" onclick="sendName();">Send</button>
        <p id="response"></p>
    </div>
</div>
</body>
</html>

5.定时推送消息

@Component
public class ReleaseNoticeTask {

    private static final Logger LOGGER = LoggerFactory.getLogger(ReleaseNoticeTask.class);

    @Autowired
    private NoticeService noticeService;

    @Autowired
    private WebsocketService websocketService;

    /**
     * 每隔30秒推送消息
     */
    @Scheduled(fixedDelay = 30*1000)
    public void task(){

        Date now=new Date();

        List<NoticeLog> list=noticeService.listNeedSendNotices(now);

        if (null != list && list.size() > 0) {
            //根据user_id点对点推送消息
            list.stream().forEach((x)->{
                try{
                    
                    websocketService.sendMsg2User(x.getUserId(),x);
                    
                }catch (Exception e){
                   
                    LOGGER.error(e.getMessage());
                    throw new InterfaceException(InterfaceExceptionEnum.INTERNAL_SERVER_ERROR,"消息推送中出现错误");
                }

            });
        }

    }

}

总结：由于第一次接触websocket，难免会遇到一些困难，主要是以下几个问题：

1.关于ws与wss

WebSocket可以使用 ws 或 wss 来作为统一资源标志符，类似于 HTTP 或 HTTPS。其中，wss 表示在 TLS 之上的 WebSocket，相当于 HTTPS。默认情况下，WebSocket的 ws 协议基于Http的 80 端口；当运行在TLS之上时，wss 协议默认是基于Http的 443 端口。说白了，wss 就是 ws 基于 SSL 的安全传输，与 HTTPS 一样样的道理。所以，如果你的网站是 HTTPS 协议的，那你就不能使用 ws:// 了，浏览器会 block 掉连接，和 HTTPS 下不允许 HTTP 请求一样。

2.WebSocket connection to ‘wss://{域名}/‘ failed: Error during WebSocket handshake: Unexpected response code: 400

Nginx默认不支持websocket，用域名访问websocket时需修改配置，当 Nginx 代理服务器拦截到一个客户端发来的 Upgrade 请求时，需要我们显式的配置Connection、Upgrade头信息，并使用 101（交换协议）返回响应，在客户端、代理服务器和后端应用服务之间建立隧道来支持 WebSocket，添加以下配置问题可解决。

server {
    lister 8020;
    location / {
                    proxy_ pass http://websacket:
                    proxy_ set_ header Host $host:8020; #注意，原host必须配置，否则传递给后台的值是websocket,端口如果没有输入的话会是80,这会导致连接失败
                    proxy_ http_ version 1.1;
                    proxy_ set_ header Upgrade $http_ upgrade; 
                    proxy_ set_ header Connection "Upgrade";
                }
        }

3. No messages received after 466014 ms. Closing WebSocketServerSockJsSession[id=3x0zvzhj].

这属于STOMP协议后台不发送心跳的问题，在springboot中websocket配置文件设置心跳，配置setHeartbeatValue(new long[]{HEART_BEAT,HEART_BEAT}).setTaskScheduler(new DefaultManagedTaskScheduler());。前一个是配置心跳，后一个使用一个线程发送心跳。断网重连的时候后台捕获到异常，关闭连接，前端再发起重连接操作。最近发现是spring版本，版本4.3.15已经把日志的客户端断开连接报错error级别改为info级别。

@Override
	public void configureMessageBroker(MessageBrokerRegistry config) {
		//服务端发送消息给客户端的域,多个用逗号隔开
		config.enableSimpleBroker("/topic", "/user")
				.setTaskScheduler(new DefaultManagedTaskScheduler())
				.setHeartbeatValue(new long[]{20000,20000});	//设置心跳
		config.setApplicationDestinationPrefixes("/app");
	}


    /**
	 * 实施定制来断开会话
	 * @param registration
	 */
	@Override
	public void configureWebSocketTransport(final WebSocketTransportRegistration registration) {
		registration.addDecoratorFactory(new WebSocketHandlerDecoratorFactory() {
			@Override
			public WebSocketHandler decorate(final WebSocketHandler handler) {
				return new WebSocketHandlerDecorator(handler) {

					@Override
					public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
						LOGGER.warn("会话过程出现错误，关闭连接");
						LOGGER.warn(exception.getMessage());
						session.close();
						super.handleTransportError(session, exception);
					}
				};
			}
		});
		super.configureWebSocketTransport(registration);
	}

参考链接 https://stackoverflow.com/questions/39220647/spring-stomp-over-websockets-not-scheduling-heartbeats

4.最后还有一个小问题，个人问题，由于本地和测试环境使用的是同一个数据库，导致本地项目和测试环境同时运行时，数据库要推送的数据可能造成本地环境浏览器和测试环境浏览器只有一个能看到浏览器通知，造成信息发送的紊乱，影响错误排查，做事写代码多考虑周全点，也是个人追求的。

Springboot+Websocket+Scheduled开发案例总结

案例简介

关键技术

案例环境要求

继续阅读

关于Gradle配置的小结

Java小案例——随机数猜测随机数猜测

nginx location中斜线的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的简单使用

neo4j之cypher使用文档

GitHub连夜封杀！这份阿里 10W 字内部 Java 字面试手册到底有多强？

spark/scala关于【资源文件】加载方法概述外部文件加载方案测试资源文件打包入jar包中小结

mybatis_入门程序Mybatis入门

AOP编程_Android优雅权限框架(1)概念基础，2021金三银四前言正文大纲正文

Effective Java 8:通用程序设计

OOM三种类型

工厂模式-三种类型

【递归】高效率求2的n次幂

win10本地scala和spark安装安装scala安装spark

scala (3) Function 和 Method