Spring Gateway转发websocket原理

Spring Cloud Gateway简称Spring Gateway，它可以转发请求到后端微服务。Spring Gateway除了转发HTTP请求，也支持websocket请求。我们看下它是怎么实现的吧。

配置支持websocket转发

支持websocket转发，需要用到spring-cloud-starter-gateway ，不要搞错成spring-cloud-starter-gateway-web 。引入maven配置：

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-gateway</artifactId>
    <version>4.1.4</version>
</dependency>

然后注册需要路由的规则，可以通过yml配置。

spring:
  cloud:
    gateway:
      routes:
        - id: ws1
          uri: ws://localhost:8080
          predicates:
            - Path=/ws

Java配置方式，与yml方式等效。

@Bean
public RouteLocator customRouteLocator(RouteLocatorBuilder builder) {
    return builder.routes()
            .route("ws1", r -> r.path("/ws")
                    .uri("ws://localhost:8080"))
            .build();
}

websocket转发原理

处理websocket协议转发的类是org.springframework.cloud.gateway.filter.WebsocketRoutingFilter。它的filter方法会过滤出ws和wss两类请求。

@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
	changeSchemeIfIsWebSocketUpgrade(exchange);
 
	URI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR);
	String scheme = requestUrl.getScheme();
 
	if (isAlreadyRouted(exchange) || (!"ws".equals(scheme) && !"wss".equals(scheme))) {
		return chain.filter(exchange);
	}
	setAlreadyRouted(exchange);
 
	HttpHeaders headers = exchange.getRequest().getHeaders();
	HttpHeaders filtered = filterRequest(getHeadersFilters(), exchange);
 
	List<String> protocols = getProtocols(headers);
 
	return this.webSocketService.handleRequest(exchange,
			new ProxyWebSocketHandler(requestUrl, this.webSocketClient, filtered, protocols));
}

可以看到方法的最后一行的handleRequest()方法，exchange是当前发给网关的ws握手请求，ProxyWebSocketHandler用来处理网关和客户端建立完websocket链接成功后的事件。重点看看ProxyWebSocketHandler的handle()方法。

 
@Override
public Mono<Void> handle(WebSocketSession session) {
	// pass headers along so custom headers can be sent through
	return client.execute(url, this.headers, new WebSocketHandler() {
	  // 省略部分代码... 
	});
}

handle()方法里用client（WebSocketClient ）给后端websocket地址发来一个握手请求。

到这里，网关握手的流程就清晰了。前端客户端 —[连接]—>网关—[连接]—>后端websocket服务，总共会产生2条websocket连接。

然后就是发消息和断开连接的方式，就在上面省略代码里。

new WebSocketHandler() {
 
	@Override
	public Mono<Void> handle(WebSocketSession proxySession) {
		Mono<Void> serverClose = proxySession.closeStatus().filter(__ -> session.isOpen())
				.map(this::adaptCloseStatus).flatMap(session::close);
		Mono<Void> proxyClose = session.closeStatus().filter(__ -> proxySession.isOpen())
				.map(this::adaptCloseStatus).flatMap(proxySession::close);
		// Use retain() for Reactor Netty
		Mono<Void> proxySessionSend = proxySession
				.send(session.receive().doOnNext(WebSocketMessage::retain).doOnNext(webSocketMessage -> {
					if (log.isTraceEnabled()) {
						log.trace("proxySession(send from client): " + proxySession.getId()
								+ ", corresponding session:" + session.getId() + ", packet: "
								+ webSocketMessage.getPayloadAsText());
					}
				}));
		// .log("proxySessionSend", Level.FINE);
		Mono<Void> serverSessionSend = session.send(
				proxySession.receive().doOnNext(WebSocketMessage::retain).doOnNext(webSocketMessage -> {
					if (log.isTraceEnabled()) {
						log.trace("session(send from backend): " + session.getId()
								+ ", corresponding proxySession:" + proxySession.getId() + " packet: "
								+ webSocketMessage.getPayloadAsText());
					}
				}));
		// .log("sessionSend", Level.FINE);
		// Ensure closeStatus from one propagates to the other
		Mono.when(serverClose, proxyClose).subscribe();
		// Complete when both sessions are done
		return Mono.zip(proxySessionSend, serverSessionSend).then();
	}
}

websocket连接关闭，是serverClose和proxyClose这两行代码，当后端的websocket连接断开时，就会把断开的转发设置到网关的websocket连接上；网关的websocket连接断开时，就会把断开的转发设置到后端的websocket连接上。这样，两个websocket连接的断开状态就一致了。

websocket发送消息，是proxySessionSend和serverSessionSend这两行代码，当网关收到客户端消息时，就会把消息发送给后端websocket服务；当网关收到后端websocket发来的消息时，就会把消息转发给客户端。

至此，网关在websocket连接、发消息、断开连接就和后端websocket服务保持一致了。