springboot java实现Webscoket客户端/服务端,心跳以及重连附代码

WebScoket介绍

是一种网络通讯协议,是HTML5开始提供的一种在单个TCP连接上的双全功通讯协议.

已经有HTTP协议,为什么还需要另外一个协议?

1.HTTP协议是一种无状态,无连接的,单向的应用层协议.通讯只能由客户端发起,无法实现服务器主动想客户端推送消息
2.既然是一个单向请求,那么如果服务器有连续的状态变化,客户端要获取就会很麻烦, 采用轮询
  的效率低,非常浪费资源(因为必须不停的链接,或者HTTP链接始终打开),由此WebScoket应运而生.

WebScoket特点介绍

最大特点:服务端可以主动向客户端推送消息,客户端也可以主动向服务器发送消息,是双向
平等对话,属于服务器推送技术的一种.

其他特点:
1. 建立在TCP协议上的,服务器端易于实现
2. 与HTTP协议有着良好的兼容性,默认端口也是80和443,并且握手阶段采用HTTP协议,
因此握手时不用一屏蔽,能通过各种http代理服务器
3. 数据格式比较轻量,性能开销小,通信高效
4. 可以发送文本,也可以发送二进制数据
5. 没有同源限制,客户端可以与任意服务器通信
6. 协议标识符是ws (如加密,则为wss) ,服务器网址就是URL(ws://localhost:8080/api/v1/example)

服务端实现

由于项目代码相对较多,此处只贴出核心代码配置,具体链接地址后续会附上

添加maven依赖。

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

创建Webscoket服务器, implements WebSocketHandler 或者 extends TextWebSocketHandler 或 BinaryWebSocketHandler。

@Component
public class MyWebSocketHandler  implements WebSocketHandler {
    @Override
    public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception {
    	//TODO  接受来自客户端的消息
    }
}

WebSocketHandler 源码如下，这意味着你的处理器大概可以处理哪些 WebSocket 事件。

public interface WebSocketHandler {

   /**
    * 建立连接后触发的回调
    */
   void afterConnectionEstablished(WebSocketSession session) throws Exception;

   /**
    * 收到消息时触发的回调
    */
   void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception;

   /**
    * 传输消息出错时触发的回调
    */
   void handleTransportError(WebSocketSession session, Throwable exception) throws Exception;

   /**
    * 断开连接后触发的回调
    */
   void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception;

   /**
    * 是否处理分片消息
    */
   boolean supportsPartialMessages();

}

配置WebScoket的访问通道。

import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        //注册通道
        registry.addHandler(myHandler(), "/ws-service").setAllowedOrigins("*").addInterceptors(myInterceptor());
        // withSockJS() 方法声明我们想要使用 SockJS 功能，如果WebSocket不可用的话，会使用 SockJS；
        registry.addHandler(myHandler(), "/sockjs/wsservice").setAllowedOrigins("*").addInterceptors(myInterceptor()).withSockJS();
    }

    @Bean
    public WebSocketHandler myHandler() {
        return new MyHandler();
    }

}

WebScoket握手,在进行消息交互的前后,可以调用HandshakeInterceptor来进行有效的消息过滤,处理需要的信息,。

import org.springframework.http.server.ServerHttpRequest;
import org.springframework.http.server.ServerHttpResponse;
import org.springframework.http.server.ServletServerHttpRequest;
import org.springframework.util.StringUtils;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.server.HandshakeInterceptor;
import org.springframework.web.socket.server.support.HttpSessionHandshakeInterceptor;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpSession;
import java.util.Map;

public class MyWebSocketHandshakeInterceptor extends HttpSessionHandshakeInterceptor implements HandshakeInterceptor {

	@Override
	public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map<String, Object> attributes) throws Exception {
		//TODO  
	}

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

客户端配置

maven依赖。

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

客户端实现 implements WebSocketHandler 接口。

@Component
public class MyWebSocketHandler  implements WebSocketHandler {
    @Override
    public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception {
    	//TODO  接受来自服务端的消息
    }
}

添加Listener监听器,服务启动建立链接,贴出核心实现。

	/**
	 * 定义websocket配置
	 * 
	 * @return
	 */
	@Bean(name = "wsCloudConnectionManager")
	public WebSocketConnectionManager wsCloudConnectionManager() {
		StandardWebSocketClient webSocketClient = new StandardWebSocketClient();
		WebSocketConnectionManager manager = new WebSocketConnectionManager(webSocketClient, myWebSocketClientHandler,
				"ws://localhost:8080/ws-service?shopId=001&appId=002");
		manager.setAutoStartup(true);
		return manager;
	}

添加一个定时器,定时检测心跳,WebScoket有对应配置,类似于(ping/pong) 客户端做出相应,确保服务一直存活。

    /**
     * 每30秒发送一个心跳,检测断开后重连
     */
    @Scheduled(cron = "${websocket.pong.schedule.cron}")
    public void heartBeat() {
        System.out.println("执行定时任务开始");
        System.out.println(isConnected());
        try {
            if (isConnected()) {
                this.clientSession.sendMessage(new PongMessage(ByteBuffer.wrap("1".getBytes())));
            } else {
                System.out.println("Send Ping Message fail, not connect ");
                System.out.println("try " + currentConnectionTimes + " times, connection fail, reconnecting");
                currentConnectionTimes++;
                // 重连
                wsConnectionManager.startInternal();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

至此SpringBoot java实现WebScoket消息通讯完成,实现心跳,重连
demo地址 : https://gitee.com/wangshisuifeng123/web-scoket
官方文档 : https://docs.spring.io/spring/docs/current/spring-framework-reference/web.html#websocket
参考文档: http://www.ruanyifeng.com/blog/2017/05/websocket.html
STOMP协议 : https://stomp.github.io/stomp-specification-1.2.html