WebSocket的6種集成方式介紹

介紹

由于前段時(shí)間我實(shí)現(xiàn)了一個(gè)庫【Spring Cloud】一個(gè)配置注解實(shí)現(xiàn) WebSocket 集群方案

以至于我對(duì)WebSocket的各種集成方式做了一些研究

目前我所了解到的就是下面這些了（就一個(gè)破ws都有這么多花里胡哨的集成方式了？）

Javax

WebMVC

WebFlux

Java-WebSocket

SocketIO

Netty

今天主要介紹一下前3種方式，畢竟現(xiàn)在的主流框架還是Spring Boot

而后3種其實(shí)和Spring Boot并不強(qiáng)行綁定，基于Java就可以支持，不過我也會(huì)對(duì)后3種做個(gè)簡(jiǎn)單的介紹，大家先混個(gè)眼熟就行了

那么接下來我們就來講講前3種方式（Javax，WebMVC，WebFlux）在Spring Boot中的服務(wù)端和客戶端配置（客戶端配置也超重要的有木有，平時(shí)用不到，用到了卻基本找不到文檔，這也太絕望了）

Javax

在java的擴(kuò)展包javax.websocket中就定義了一套WebSocket的接口規(guī)范

服務(wù)端

一般使用注解的方式來進(jìn)行配置

第一步

@Component
@ServerEndpoint("/websocket/{type}")
publicclassJavaxWebSocketServerEndpoint{

@OnOpen
publicvoidonOpen(Sessionsession,EndpointConfigconfig,
@PathParam(value="type")Stringtype){
//連接建立
}

@OnClose
publicvoidonClose(Sessionsession,CloseReasonreason){
//連接關(guān)閉
}

@OnMessage
publicvoidonMessage(Sessionsession,Stringmessage){
//接收文本信息
}

@OnMessage
publicvoidonMessage(Sessionsession,PongMessagemessage){
//接收pong信息
}

@OnMessage
publicvoidonMessage(Sessionsession,ByteBuffermessage){
//接收二進(jìn)制信息，也可以用byte[]接收
}

@OnError
publicvoidonError(Sessionsession,Throwablee){
//異常處理
}
}

我們?cè)陬惿咸砑覢ServerEndpoint注解來表示這是一個(gè)服務(wù)端點(diǎn)，同時(shí)可以在注解中配置路徑，這個(gè)路徑可以配置成動(dòng)態(tài)的，使用{}包起來就可以了

@OnOpen用來標(biāo)記對(duì)應(yīng)的方法作為客戶端連接上來之后的回調(diào)，Session就相當(dāng)于和客戶端的連接啦，我們可以把它緩存起來用于發(fā)送消息；通過@PathParam注解就可以獲得動(dòng)態(tài)路徑中對(duì)應(yīng)值了

@OnClose用來標(biāo)記對(duì)應(yīng)的方法作為客戶端斷開連接之后的回調(diào)，我們可以在這個(gè)方法中移除對(duì)應(yīng)Session的緩存，同時(shí)可以接受一個(gè)CloseReason的參數(shù)用于獲取關(guān)閉原因

@OnMessage用來標(biāo)記對(duì)應(yīng)的方法作為接收到消息之后的回調(diào)，我們可以接受文本消息，二進(jìn)制消息和pong消息

@OnError用來標(biāo)記對(duì)應(yīng)的方法作為拋出異常之后的回調(diào)，可以獲得對(duì)應(yīng)的Session和異常對(duì)象

第二步

implementation'org.springframework.boot:spring-boot-starter-websocket'
@Configuration(proxyBeanMethods=false)
publicclassJavaxWebSocketConfiguration{

@Bean
publicServerEndpointExporterserverEndpointExporter(){
returnnewServerEndpointExporter();
}
}

依賴Spring的WebSocket模塊，手動(dòng)注入ServerEndpointExporter就可以了

需要注意ServerEndpointExporter是Spring中的類，算是Spring為了支持javax.websocket的原生用法所提供的支持類

冷知識(shí)

javax.websocket庫中定義了PongMessage而沒有PingMessage

通過我的測(cè)試發(fā)現(xiàn)基本上所有的WebSocket包括前端js自帶的，都實(shí)現(xiàn)了自動(dòng)回復(fù)；也就是說當(dāng)接收到一個(gè)ping消息之后，是會(huì)自動(dòng)回應(yīng)一個(gè)pong消息，所以沒有必要再自己接受ping消息來處理了，即我們不會(huì)接受到ping消息

當(dāng)然我上面講的ping和pong都是需要使用框架提供的api，如果是我們自己通過Message來自定義心跳數(shù)據(jù)的話是沒有任何的處理的，下面是對(duì)應(yīng)的api

//發(fā)送ping
session.getAsyncRemote().sendPing(ByteBufferbuffer);

//發(fā)送pong
session.getAsyncRemote().sendPong(ByteBufferbuffer);

然后我又發(fā)現(xiàn)js自帶的WebSocket是沒有發(fā)送ping的api的，所以是不是可以猜想當(dāng)初就是約定服務(wù)端發(fā)送ping，客戶端回復(fù)pong

客戶端

客戶端也是使用注解配置

第一步

@ClientEndpoint
publicclassJavaxWebSocketClientEndpoint{

@OnOpen
publicvoidonOpen(Sessionsession){
//連接建立
}

@OnClose
publicvoidonClose(Sessionsession,CloseReasonreason){
//連接關(guān)閉
}

@OnMessage
publicvoidonMessage(Sessionsession,Stringmessage){
//接收文本消息
}

@OnMessage
publicvoidonMessage(Sessionsession,PongMessagemessage){
//接收pong消息
}

@OnMessage
publicvoidonMessage(Sessionsession,ByteBuffermessage){
//接收二進(jìn)制消息
}

@OnError
publicvoidonError(Sessionsession,Throwablee){
//異常處理
}
}

客戶端使用@ClientEndpoint來標(biāo)記，其他的@OnOpen，@OnClose，@OnMessage，@OnError和服務(wù)端一模一樣

第二步

WebSocketContainercontainer=ContainerProvider.getWebSocketContainer();
Sessionsession=container.connectToServer(JavaxWebSocketClientEndpoint.class,uri);

我們可以通過ContainerProvider來獲得一個(gè)WebSocketContainer，然后調(diào)用connectToServer方法將我們的客戶端類和連接的uri傳入就行了

冷知識(shí)

通過ContainerProvider#getWebSocketContainer獲得WebSocketContainer其實(shí)是基于SPI實(shí)現(xiàn)的

在Spring的環(huán)境中我更推薦大家使用ServletContextAware來獲得，代碼如下

@Component
publicclassJavaxWebSocketContainerimplementsServletContextAware{

privatevolatileWebSocketContainercontainer;

publicWebSocketContainergetContainer(){
if(container==null){
synchronized(this){
if(container==null){
container=ContainerProvider.getWebSocketContainer();
}
}
}
returncontainer;
}

@Override
publicvoidsetServletContext(@NonNullServletContextservletContext){
if(container==null){
container=(WebSocketContainer)servletContext
.getAttribute("javax.websocket.server.ServerContainer");
}
}
}

發(fā)消息

Sessionsession=...

//發(fā)送文本消息
session.getAsyncRemote().sendText(Stringmessage);

//發(fā)送二進(jìn)制消息
session.getAsyncRemote().sendBinary(ByteBuffermessage);

//發(fā)送對(duì)象消息，會(huì)嘗試使用Encoder編碼
session.getAsyncRemote().sendObject(Objectmessage);

//發(fā)送ping
session.getAsyncRemote().sendPing(ByteBufferbuffer);

//發(fā)送pong
session.getAsyncRemote().sendPong(ByteBufferbuffer);

WebMVC

依賴肯定是必不可少的

implementation'org.springframework.boot:spring-boot-starter-websocket'

服務(wù)端

第一步

importorg.springframework.web.socket.WebSocketHandler;
importorg.springframework.web.socket.WebSocketMessage;
importorg.springframework.web.socket.WebSocketSession;

publicclassServletWebSocketServerHandlerimplementsWebSocketHandler{

@Override
publicvoidafterConnectionEstablished(@NonNullWebSocketSessionsession)throwsException{
//連接建立
}

@Override
publicvoidhandleMessage(@NonNullWebSocketSessionsession,@NonNullWebSocketMessagemessage)throwsException{
//接收消息
}

@Override
publicvoidhandleTransportError(@NonNullWebSocketSessionsession,@NonNullThrowableexception)throwsException{
//異常處理
}

@Override
publicvoidafterConnectionClosed(@NonNullWebSocketSessionsession,@NonNullCloseStatuscloseStatus)throwsException{
//連接關(guān)閉
}

@Override
publicbooleansupportsPartialMessages(){
//是否支持接收不完整的消息
returnfalse;
}
}

我們實(shí)現(xiàn)一個(gè)WebSocketHandler來處理WebSocket的連接，關(guān)閉，消息和異常

第二步

@Configuration
@EnableWebSocket
publicclassServletWebSocketServerConfigurerimplementsWebSocketConfigurer{

@Override
publicvoidregisterWebSocketHandlers(@NonNullWebSocketHandlerRegistryregistry){
registry
//添加處理器到對(duì)應(yīng)的路徑
.addHandler(newServletWebSocketServerHandler(),"/websocket")
.setAllowedOrigins("*");
}
}

首先需要添加@EnableWebSocket來啟用WebSocket

然后實(shí)現(xiàn)WebSocketConfigurer來注冊(cè)WebSocket路徑以及對(duì)應(yīng)的WebSocketHandler

握手?jǐn)r截

提供了HandshakeInterceptor來攔截握手

@Configuration
@EnableWebSocket
publicclassServletWebSocketServerConfigurerimplementsWebSocketConfigurer{

@Override
publicvoidregisterWebSocketHandlers(@NonNullWebSocketHandlerRegistryregistry){
registry
//添加處理器到對(duì)應(yīng)的路徑
.addHandler(newServletWebSocketServerHandler(),"/websocket")
//添加握手?jǐn)r截器
.addInterceptors(newServletWebSocketHandshakeInterceptor())
.setAllowedOrigins("*");
}

publicstaticclassServletWebSocketHandshakeInterceptorimplementsHandshakeInterceptor{

@Override
publicbooleanbeforeHandshake(ServerHttpRequestrequest,ServerHttpResponseresponse,WebSocketHandlerwsHandler,Mapattributes)throwsException{
//握手之前
//繼續(xù)握手返回true,中斷握手返回false
returnfalse;
}

@Override
publicvoidafterHandshake(ServerHttpRequestrequest,ServerHttpResponseresponse,WebSocketHandlerwsHandler,Exceptionexception){
//握手之后
}
}
}

冷知識(shí)

我在集成的時(shí)候發(fā)現(xiàn)這種方式?jīng)]辦法動(dòng)態(tài)匹配路徑，它的路徑就是固定的，沒辦法使用如/websocket/**這樣的通配符

我在研究了一下之后發(fā)現(xiàn)可以在UrlPathHelper上做點(diǎn)文章

@Configuration
@EnableWebSocket
publicclassServletWebSocketServerConfigurerimplementsWebSocketConfigurer{

@Override
publicvoidregisterWebSocketHandlers(@NonNullWebSocketHandlerRegistryregistry){
if(registryinstanceofServletWebSocketHandlerRegistry){
//替換UrlPathHelper
((ServletWebSocketHandlerRegistry)registry)
.setUrlPathHelper(newPrefixUrlPathHelper("/websocket"));
}

registry
//添加處理器到對(duì)應(yīng)的路徑
.addHandler(newServletWebSocketServerHandler(),"/websocket/**")
.setAllowedOrigins("*");
}

publicclassPrefixUrlPathHelperextendsUrlPathHelper{

privateStringprefix;

@Override
public@NonNullStringresolveAndCacheLookupPath(@NonNullHttpServletRequestrequest){
//獲得原本的Path
Stringpath=super.resolveAndCacheLookupPath(request);
//如果是指定前綴就返回對(duì)應(yīng)的通配路徑
if(path.startsWith(prefix)){
returnprefix+"/**";
}
returnpath;
}
}
}

因?yàn)樗鼉?nèi)部實(shí)際上就是用一個(gè)Map來存的，所以沒有辦法用通配符

主要是有現(xiàn)成的AntPathMatcher實(shí)現(xiàn)通配應(yīng)該不麻煩才對(duì)啊

客戶端

第一步

publicclassServletWebSocketClientHandlerimplementsWebSocketHandler{

@Override
publicvoidafterConnectionEstablished(@NonNullWebSocketSessionsession)throwsException{
//連接建立
}

@Override
publicvoidhandleMessage(@NonNullWebSocketSessionsession,@NonNullWebSocketMessagemessage)throwsException{
//接收消息
}

@Override
publicvoidhandleTransportError(@NonNullWebSocketSessionsession,@NonNullThrowableexception)throwsException{
//異常處理
}

@Override
publicvoidafterConnectionClosed(@NonNullWebSocketSessionsession,@NonNullCloseStatuscloseStatus)throwsException{
//連接關(guān)閉
}

@Override
publicbooleansupportsPartialMessages(){
//是否支持接收不完整的消息
returnfalse;
}
}

和服務(wù)端一樣我們需要先實(shí)現(xiàn)一個(gè)WebSocketHandler來處理WebSocket的連接，關(guān)閉，消息和異常

第二步

WebSocketClientclient=newStandardWebSocketClient();
WebSocketHandlerhandler=newServletWebSocketClientHandler();
WebSocketConnectionManagermanager=newWebSocketConnectionManager(client,handler,uri);
manager.start();

首先我們需要先new一個(gè)StandardWebSocketClient，可以傳入一個(gè)WebSocketContainer參數(shù)，獲得該對(duì)象的方式我之前已經(jīng)介紹過了，這邊就先略過

然后new一個(gè)WebSocketConnectionManager傳入WebSocketClient，WebSocketHandler還有路徑uri

最后調(diào)用一下WebSocketConnectionManager的start方法就可以啦

冷知識(shí)

這里如果大家去看WebSocketClient的實(shí)現(xiàn)類就會(huì)發(fā)現(xiàn)有StandardWebSocketClient還有JettyWebSocketClient等等，所以大家可以根據(jù)自身項(xiàng)目所使用的容器來選擇不同的WebSocketClient實(shí)現(xiàn)類

這里給大家貼一小段Spring適配不同容器WebSocket的代碼

publicabstractclassAbstractHandshakeHandlerimplementsHandshakeHandler,Lifecycle{

privatestaticfinalbooleantomcatWsPresent;

privatestaticfinalbooleanjettyWsPresent;

privatestaticfinalbooleanjetty10WsPresent;

privatestaticfinalbooleanundertowWsPresent;

privatestaticfinalbooleanglassfishWsPresent;

privatestaticfinalbooleanweblogicWsPresent;

privatestaticfinalbooleanwebsphereWsPresent;

static{
ClassLoaderclassLoader=AbstractHandshakeHandler.class.getClassLoader();
tomcatWsPresent=ClassUtils.isPresent(
"org.apache.tomcat.websocket.server.WsHttpUpgradeHandler",classLoader);
jetty10WsPresent=ClassUtils.isPresent(
"org.eclipse.jetty.websocket.server.JettyWebSocketServerContainer",classLoader);
jettyWsPresent=ClassUtils.isPresent(
"org.eclipse.jetty.websocket.server.WebSocketServerFactory",classLoader);
undertowWsPresent=ClassUtils.isPresent(
"io.undertow.websockets.jsr.ServerWebSocketContainer",classLoader);
glassfishWsPresent=ClassUtils.isPresent(
"org.glassfish.tyrus.servlet.TyrusHttpUpgradeHandler",classLoader);
weblogicWsPresent=ClassUtils.isPresent(
"weblogic.websocket.tyrus.TyrusServletWriter",classLoader);
websphereWsPresent=ClassUtils.isPresent(
"com.ibm.websphere.wsoc.WsWsocServerContainer",classLoader);
}
}

發(fā)消息

importorg.springframework.web.socket.*;

WebSocketSessionsession=...

//發(fā)送文本消息
session.sendMessage(newTextMessage(CharSequencemessage);

//發(fā)送二進(jìn)制消息
session.sendMessage(newBinaryMessage(ByteBuffermessage));

//發(fā)送ping
session.sendMessage(newPingMessage(ByteBuffermessage));

//發(fā)送pong
session.sendMessage(newPongMessage(ByteBuffermessage));

WebFlux

WebFlux的WebSocket不需要額外的依賴包

服務(wù)端

第一步

importorg.springframework.web.reactive.socket.WebSocketHandler;
importorg.springframework.web.reactive.socket.WebSocketSession;

publicclassReactiveWebSocketServerHandlerimplementsWebSocketHandler{

@NonNull
@Override
publicMonohandle(WebSocketSessionsession){
Monosend=session.send(Flux.create(sink->{
//可以持有sink對(duì)象在任意時(shí)候調(diào)用next發(fā)送消息
sink.next(WebSocketMessagemessage);
})).doOnError(it->{
//異常處理
});

Monoreceive=session.receive()
.doOnNext(it->{
//接收消息
})
.doOnError(it->{
//異常處理
})
.then();

@SuppressWarnings("all")
Disposabledisposable=session.closeStatus()
.doOnError(it->{
//異常處理
})
.subscribe(it->{
//連接關(guān)閉
});

returnMono.zip(send,receive).then();
}
}

首先需要注意這里的WebSocketHandler和WebSocketSession是reactive包下的

通過WebSocketSession#send方法來持有一個(gè)FluxSink來用于發(fā)送消息

通過WebSocketSession#receive來訂閱消息

通過WebSocketSession#closeStatus來訂閱連接關(guān)閉事件

第二步

@Component
publicclassReactiveWebSocketServerHandlerMappingextendsSimpleUrlHandlerMapping{

publicReactiveWebSocketServerHandlerMapping(){
Mapmap=newHashMap<>();
map.put("/websocket/**",newReactiveWebSocketServerHandler());
setUrlMap(map);
setOrder(100);
}
}

注冊(cè)一個(gè)HandlerMapping同時(shí)配置路徑和對(duì)應(yīng)的WebSocketHandler

第三步

@Configuration(proxyBeanMethods=false)
publicclassReactiveWebSocketConfiguration{

@Bean
publicWebSocketHandlerAdapterwebSocketHandlerAdapter(){
returnnewWebSocketHandlerAdapter();
}
}

注入WebSocketHandlerAdapter

冷知識(shí)

我們自定義的HandlerMapping需要設(shè)置order，如果不設(shè)置，默認(rèn)為Ordered.LOWEST_PRECEDENCE，會(huì)導(dǎo)致這個(gè)HandlerMapping被放在最后，當(dāng)有客戶端連接上來時(shí)會(huì)被其他的HandlerMapping優(yōu)先匹配上而連接失敗

客戶端

第一步

publicclassReactiveWebSocketClientHandlerimplementsWebSocketHandler{

@NonNull
@Override
publicMonohandle(WebSocketSessionsession){
Monosend=session.send(Flux.create(sink->{
//可以持有sink對(duì)象在任意時(shí)候調(diào)用next發(fā)送消息
sink.next(WebSocketMessagemessage);
})).doOnError(it->{
//處理異常
});

Monoreceive=session.receive()
.doOnNext(it->{
//接收消息
})
.doOnError(it->{
//異常處理
})
.then();

@SuppressWarnings("all")
Disposabledisposable=session.closeStatus()
.doOnError(it->{
//異常處理
})
.subscribe(it->{
//連接關(guān)閉
});

returnMono.zip(send,receive).then();
}
}

客戶端WebSocketHandler的寫法和服務(wù)端的一樣

第二步

importorg.springframework.web.reactive.socket.client.WebSocketClient;

WebSocketClientclient=ReactorNettyWebSocketClient();
WebSocketHandlerhandler=newReactiveWebSocketClientHandler();
client.execute(uri,handler).subscribe();

首先我們需要先new一個(gè)ReactorNettyWebSocketClient

然后調(diào)用一下WebSocketClient的execute方法傳入路徑uri和WebSocketHandler并繼續(xù)調(diào)用subscribe方法就行啦

冷知識(shí)

和WebMVC中的WebSocketClient一樣，Reactive包中的WebSocketClient也有很多實(shí)現(xiàn)類，比如ReactorNettyWebSocketClient，JettyWebSocketClient，UndertowWebSocketClient，TomcatWebSocketClient等等，也是需要大家基于自身項(xiàng)目的容器使用不同的實(shí)現(xiàn)類

這里也給大家貼一小段Reactive適配不同容器WebSocket的代碼

publicclassHandshakeWebSocketServiceimplementsWebSocketService,Lifecycle{

privatestaticfinalbooleantomcatPresent;

privatestaticfinalbooleanjettyPresent;

privatestaticfinalbooleanjetty10Present;

privatestaticfinalbooleanundertowPresent;

privatestaticfinalbooleanreactorNettyPresent;

static{
ClassLoaderloader=HandshakeWebSocketService.class.getClassLoader();
tomcatPresent=ClassUtils.isPresent("org.apache.tomcat.websocket.server.WsHttpUpgradeHandler",loader);
jettyPresent=ClassUtils.isPresent("org.eclipse.jetty.websocket.server.WebSocketServerFactory",loader);
jetty10Present=ClassUtils.isPresent("org.eclipse.jetty.websocket.server.JettyWebSocketServerContainer",loader);
undertowPresent=ClassUtils.isPresent("io.undertow.websockets.WebSocketProtocolHandshakeHandler",loader);
reactorNettyPresent=ClassUtils.isPresent("reactor.netty.http.server.HttpServerResponse",loader);
}
}

發(fā)消息

我們需要使用在WebSocketHandler中獲得的FluxSink來發(fā)送消息

importorg.springframework.web.reactive.socket.CloseStatus;
importorg.springframework.web.reactive.socket.WebSocketMessage;
importorg.springframework.web.reactive.socket.WebSocketSession;

publicclassReactiveWebSocket{

privatefinalWebSocketSessionsession;

privatefinalFluxSinksender;

publicReactiveWebSocket(WebSocketSessionsession,FluxSinksender){
this.session=session;
this.sender=sender;
}

publicStringgetId(){
returnsession.getId();
}

publicURIgetUri(){
returnsession.getHandshakeInfo().getUri();
}

publicvoidsend(Objectmessage){
if(messageinstanceofWebSocketMessage){
sender.next((WebSocketMessage)message);
}elseif(messageinstanceofString){
//發(fā)送文本消息
sender.next(session.textMessage((String)message));
}elseif(messageinstanceofDataBuffer){
//發(fā)送二進(jìn)制消息
sender.next(session.binaryMessage(factory->(DataBuffer)message));
}elseif(messageinstanceofByteBuffer){
發(fā)送二進(jìn)制消息
sender.next(session.binaryMessage(factory->factory.wrap((ByteBuffer)message)));
}elseif(messageinstanceofbyte[]){
發(fā)送二進(jìn)制消息
sender.next(session.binaryMessage(factory->factory.wrap((byte[])message)));
}else{
thrownewIllegalArgumentException("Messagetypenotmatch");
}
}

publicvoidping(){
//發(fā)送ping
sender.next(session.pingMessage(factory->factory.wrap(ByteBuffer.allocate(0))));
}

publicvoidpong(){
//發(fā)送pong
sender.next(session.pongMessage(factory->factory.wrap(ByteBuffer.allocate(0))));
}

publicvoidclose(CloseStatusreason){
sender.complete();
session.close(reason).subscribe();
}
}

Java-WebSocket

這是一個(gè)純java的第三方庫，專門用于實(shí)現(xiàn)WebSocket

SocketIO

該庫使用的協(xié)議是經(jīng)過自己封裝的，支持很多的語言，提供了統(tǒng)一的接口，所以需要使用它提供的Server和Client來連接，如socket.io-server-java和socket.io-client-java

這個(gè)庫我了解下來主要用于實(shí)時(shí)聊天等場(chǎng)景，所以如果只是普通的WebSocket功能就有點(diǎn)大材小用了

Netty

這個(gè)大家應(yīng)該都比較熟悉了，就算沒用過肯定也聽過

網(wǎng)上的文檔和示例也非常多，我這里就不介紹有的沒的了，Github傳送門。

審核編輯：劉清