Spring Boot WebFlux 中的 WebSocket 提供了一種高效、異步的方式來(lái)處理客戶端與服務(wù)器之間的雙向通信。WebSocket 連接的生命周期包括連接建立、消息傳輸、連接關(guān)閉以及資源清理等過(guò)程。此外，為了確保 WebSocket 連接的穩(wěn)定性和可靠性，我們可以加入重試機(jī)制，以處理斷開(kāi)或網(wǎng)絡(luò)問(wèn)題時(shí)自動(dòng)重新連接。

1. WebSocket 連接建立

WebSocket 的連接是通過(guò) HTTP 的 Upgrade 機(jī)制從普通的 HTTP/HTTPS 請(qǐng)求升級(jí)而來(lái)的。具體流程如下：

1.1 客戶端請(qǐng)求 WebSocket 連接

客戶端通過(guò) ws:// 或 wss:// 協(xié)議來(lái)訪問(wèn) WebSocket 服務(wù)器，并發(fā)送 HTTP Upgrade 請(qǐng)求頭，要求服務(wù)器將連接升級(jí)為 WebSocket 協(xié)議：

GET /ws HTTP/1.1
Host: example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: random-generated-key
Sec-WebSocket-Version: 13

1.2 服務(wù)器端處理 WebSocket 連接

Spring WebFlux 通過(guò) WebSocketHandler 來(lái)處理 WebSocket 請(qǐng)求。以下是一個(gè)簡(jiǎn)單的 WebSocketHandler 實(shí)現(xiàn)：

@Component
public class MyWebSocketHandler implements WebSocketHandler {@Overridepublic Mono<Void> handle(WebSocketSession session) {return session.receive().doOnNext(message -> System.out.println("Received: " + message.getPayloadAsText())).then();}
}

當(dāng)服務(wù)器收到 HTTP Upgrade 請(qǐng)求后，它會(huì)檢查 Sec-WebSocket-Key 并返回 Sec-WebSocket-Accept 進(jìn)行握手，建立連接。

1.3 握手成功，連接建立

如果握手成功，服務(wù)器會(huì)返回 101 Switching Protocols 響應(yīng)，表示 WebSocket 連接已建立：

HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: (calculated key)

2. WebSocket 消息處理

連接建立后，WebSocket 進(jìn)入消息傳輸階段，包括消息的接收和發(fā)送。

2.1 消息接收

服務(wù)器端可以通過(guò) WebSocketSession.receive() 方法來(lái)接收客戶端發(fā)送的消息：

session.receive().map(WebSocketMessage::getPayloadAsText).doOnNext(msg -> System.out.println("Received: " + msg)).then();

session.receive() 返回一個(gè) Flux<WebSocketMessage>，可以處理流式消息，每次接收到新消息時(shí)執(zhí)行 doOnNext() 中的處理邏輯。

2.2 消息發(fā)送

服務(wù)器端可以通過(guò) WebSocketSession.send() 方法發(fā)送消息給客戶端：

Flux<String> messages = Flux.interval(Duration.ofSeconds(1)).map(i -> "Message " + i);
return session.send(messages.map(session::textMessage));

send() 方法接收一個(gè) Publisher<WebSocketMessage>，可以使用 Flux 來(lái)生成消息流。textMessage() 方法用于創(chuàng)建文本消息。

3. WebSocket 連接關(guān)閉

WebSocket 連接可以由客戶端、服務(wù)器或網(wǎng)絡(luò)異常等原因主動(dòng)關(guān)閉。連接關(guān)閉的主要方式如下：

3.1 正常關(guān)閉

• 客戶端主動(dòng)關(guān)閉：客戶端可以通過(guò)調(diào)用 WebSocket.close() 發(fā)送 Close Frame，服務(wù)器接收到后會(huì)關(guān)閉連接。
• 服務(wù)器主動(dòng)關(guān)閉：服務(wù)器通過(guò) WebSocketSession.close() 關(guān)閉連接：

  session.close(CloseStatus.NORMAL);
  

3.2 異常關(guān)閉

• 網(wǎng)絡(luò)異常：如網(wǎng)絡(luò)斷開(kāi)或客戶端崩潰等，連接會(huì)被強(qiáng)制關(guān)閉。
• 心跳超時(shí)：如果使用 ping/pong 機(jī)制檢測(cè) WebSocket 是否存活，超時(shí)未收到 pong 響應(yīng)時(shí)，連接會(huì)關(guān)閉。

  session.send(Flux.just(session.pingMessage(ByteBuffer.wrap(new byte[0]))));
  

3.3 連接關(guān)閉后的處理

服務(wù)器可以使用 session.receive().doOnTerminate() 監(jiān)聽(tīng)連接關(guān)閉事件，執(zhí)行清理操作：

session.receive().doOnTerminate(() -> System.out.println("WebSocket connection closed")).then();

4. WebSocket 生命周期總結(jié)

階段	說(shuō)明
連接建立	客戶端發(fā)起 WebSocket 連接請(qǐng)求，服務(wù)器接受并返回 101 Switching Protocols 響應(yīng)，連接建立。
消息傳輸	服務(wù)器和客戶端可以雙向傳輸文本或二進(jìn)制消息。
連接關(guān)閉	連接可由客戶端、服務(wù)器、網(wǎng)絡(luò)異常等原因關(guān)閉。
資源清理	連接關(guān)閉后需要進(jìn)行資源清理操作，如取消訂閱、清理狀態(tài)等。

5. 完整示例：WebFlux WebSocket 服務(wù)器

以下是一個(gè)完整的 WebSocket 服務(wù)器配置示例，展示了如何在 Spring Boot WebFlux 中配置 WebSocket：

import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketSession;
import org.springframework.web.reactive.socket.server.support.WebSocketHandlerAdapter;
import org.springframework.web.reactive.socket.server.support.WebSocketHandlerMapping;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import reactor.core.publisher.Flux;
import java.time.Duration;
import java.util.Map;@Configuration
public class WebSocketConfig {@Beanpublic WebSocketHandler webSocketHandler() {return session -> {Flux<String> output = Flux.interval(Duration.ofSeconds(1)).map(time -> "Server time: " + time);return session.send(output.map(session::textMessage));};}@Beanpublic WebSocketHandlerMapping handlerMapping(WebSocketHandler handler) {return new WebSocketHandlerMapping(Map.of("/ws", handler));}@Beanpublic WebSocketHandlerAdapter handlerAdapter() {return new WebSocketHandlerAdapter();}
}

說(shuō)明：

• WebSocketHandler 處理 WebSocket 連接，發(fā)送定時(shí)消息。
• WebSocketHandlerMapping 將 /ws 端點(diǎn)映射到 WebSocketHandler。
• WebSocketHandlerAdapter 用于適配 WebSocket 處理器。

6. 服務(wù)器端發(fā)起 WebSocket 連接

如果你希望服務(wù)器主動(dòng)連接到其他 WebSocket 服務(wù)器，可以使用 WebSocketClient。Spring WebFlux 提供了 ReactorNettyWebSocketClient 來(lái)發(fā)起 WebSocket 連接。

6.1 示例：服務(wù)器端發(fā)起 WebSocket 連接

import org.springframework.stereotype.Service;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.WebSocketSession;
import org.springframework.web.reactive.socket.client.ReactorNettyWebSocketClient;
import reactor.core.publisher.Mono;
import java.net.URI;@Service
public class WebSocketClientService {private final ReactorNettyWebSocketClient client = new ReactorNettyWebSocketClient();public Mono<Void> connectToWebSocketServer() {return client.execute(URI.create("ws://example.com/socket"), session -> {Mono<Void> sendMessage = session.send(Mono.just(session.textMessage("Hello Server!")));session.receive().map(WebSocketMessage::getPayloadAsText).doOnNext(System.out::println).subscribe();return sendMessage;});}
}

6.2 在 Spring Boot 啟動(dòng)時(shí)自動(dòng)連接

通過(guò)在 @PostConstruct 中調(diào)用連接方法，可以確保 WebSocket 客戶端在 Spring Boot 啟動(dòng)時(shí)自動(dòng)連接：

import jakarta.annotation.PostConstruct;
import org.springframework.stereotype.Component;@Component
public class WebSocketClientInitializer {private final WebSocketClientService webSocketClientService;public WebSocketClientInitializer(WebSocketClientService webSocketClientService) {this.webSocketClientService = webSocketClientService;}@PostConstructpublic void init() {webSocketClientService.connectToWebSocketServer().subscribe();}
}

7. WebSocket 連接重試機(jī)制

在 WebSocket 的生命周期中，由于網(wǎng)絡(luò)問(wèn)題或服務(wù)器錯(cuò)誤，WebSocket 連接可能會(huì)中斷。為了提高 WebSocket 連接的可靠性，我們可以為 WebSocket 客戶端添加重試機(jī)制，以確保斷開(kāi)后能夠重新連接。

7.1 使用 retry() 方法重試連接

WebFlux 提供了 retry() 方法來(lái)自動(dòng)重試操作。以下是一個(gè)簡(jiǎn)單的重試機(jī)制示例：

import reactor.core.publisher.Mono;public class WebSocketClientService {private final ReactorNettyWebSocketClient client = new ReactorNettyWebSocketClient();public Mono<Void> connectToWebSocketServer() {return client.execute(URI.create("ws://example.com/socket"), session -> {Mono<Void> sendMessage = session.send(Mono.just(session.textMessage("Hello Server!")));session.receive().map(WebSocketMessage::getPayloadAsText).doOnNext(System.out::println).subscribe();return sendMessage;}).retry(5);  // 最大重試5次}
}

在這個(gè)例子中，retry(5) 表示如果 WebSocket 連接失敗，最多會(huì)重試 5 次。

7.2 使用 retryWhen() 實(shí)現(xiàn)自定義重試邏輯

我們還可以通過(guò) retryWhen() 來(lái)實(shí)現(xiàn)更復(fù)雜的重試策略，例如設(shè)置重試間隔時(shí)間或根據(jù)錯(cuò)誤類型決定是否重試：

import reactor.core.publisher.Mono;
import reactor.core.publisher.Flux;
import java.time.Duration;public class WebSocketClientService {private final ReactorNettyWebSocketClient client = new ReactorNettyWebSocketClient();public Mono<Void> connectToWebSocketServer() {return client.execute(URI.create("ws://example.com/socket"), session -> {Mono<Void> sendMessage = session.send(Mono.just(session.textMessage("Hello Server!")));session.receive().map(WebSocketMessage::getPayloadAsText).doOnNext(System.out::println).subscribe();return sendMessage;}).retryWhen(errors ->errors.zipWith(Flux.range(1, 5), (error, count) -> count)  // 重試次數(shù).flatMap(retryCount -> Mono.delay(Duration.ofSeconds(retryCount)))  // 增加重試間隔);}
}

在這個(gè)例子中，retryWhen() 會(huì)根據(jù)錯(cuò)誤進(jìn)行自定義重試邏輯，設(shè)置每次重試間隔遞增。

8. 連接關(guān)閉后的重試機(jī)制

為了確保連接在關(guān)閉后重新建立，我們可以監(jiān)聽(tīng)連接關(guān)閉事件并嘗試重試：

session.receive().doOnTerminate(() -> {System.out.println("WebSocket connection closed");reconnect();  // 重新連接}).then();private void reconnect() {connectToWebSocketServer().retry(3)  // 重試3次.subscribe();
}

8.1 完整的客戶端重試代碼

public Mono<Void> connectWithRetry() {return client.execute(URI.create("ws://example.com/socket"), session -> {Mono<Void> sendMessage = session.send(Mono.just(session.textMessage("Hello Server!")));session.receive().map(WebSocketMessage::getPayloadAsText).doOnNext(System.out::println).doOnTerminate(() -> reconnect())  // 連接關(guān)閉后重試.subscribe();return sendMessage;}).retryWhen(errors ->errors.zipWith(Flux.range(1, 5), (error, count) -> count).flatMap(retryCount -> Mono.delay(Duration.ofSeconds(retryCount))));
}

9. 結(jié)論

Spring Boot WebFlux 中 WebSocket 的生命周期包括：

1. 連接建立：通過(guò) HTTP Upgrade 握手建立 WebSocket 連接。
2. 消息收發(fā)：服務(wù)器和客戶端之間通過(guò) receive() 和 send() 方法進(jìn)行消息交換。
3. 連接關(guān)閉：連接可以通過(guò)正常關(guān)閉、異常關(guān)閉或主動(dòng)關(guān)閉的方式結(jié)束。
4. 資源清理：連接關(guān)閉后需要進(jìn)行資源清理操作，確保系統(tǒng)穩(wěn)定。
5. 重試機(jī)制：通過(guò) retry() 和 retryWhen() 方法為 WebSocket 連接添加自動(dòng)重試機(jī)制，提高連接的可靠性。

通過(guò) WebSocket，Spring Boot WebFlux 提供了高效的異步通信方式，特別適合用于實(shí)時(shí)數(shù)據(jù)流應(yīng)用。