摘要：本文将介绍如何在Spring WebFlux中使用WebSocket实现高级功能，包括连接建立和断开时的操作、消息收发和广播等。
继WebFlux使用案例后拓展讲解
在现代的Web应用程序中，实时性和即时通信变得越来越重要。WebSocket是一种在Web应用程序中实现实时双向通信的协议，允许服务器主动向客户端推送消息。在Spring WebFlux中，我们可以使用WebFlux的强大功能和响应式编程模型来实现WebSocket，并且还可以利用其高级功能来满足更复杂的需求。
本文将介绍如何在Spring WebFlux中使用WebSocket的高级功能，包括连接建立和断开时的操作、消息收发和广播等。让我们逐步深入了解这些功能。
在WebSocket连接建立时，我们可以执行一些操作来处理连接的初始化。例如，我们可以进行身份验证、订阅特定主题或加载初始数据。在Spring WebFlux中，我们可以通过实现WebSocketHandler接口，并在handle()方法中重写相应的逻辑来实现这些操作。
以下是一个示例：
public class MyWebSocketHandler implements WebSocketHandler {
@Override
public Mono<Void> handle(WebSocketSession session) {
// 连接建立时执行的操作
System.out.println("WebSocket连接建立");
// 处理接收到的消息...
return session.send(/* 响应消息给客户端 */)
.doFinally(signalType -> {
// 连接关闭时执行的操作
System.out.println("WebSocket连接关闭");
});
}
}
在上述示例中，我们在handle()方法中打印一条消息表示连接已建立。在连接关闭时，我们使用doFinally()操作符注册一个回调函数，在连接关闭时执行相应的操作。
您可以根据实际需求扩展这些操作，例如执行身份验证、订阅主题或加载初始数据。
在WebSocket连接建立后，客户端和服务器之间可以相互发送消息。在Spring WebFlux中，我们可以使用WebSocketSession对象来处理接收到的消息，并使用send()方法将响应消息发送给客户端。
以下是一个示例：
public class MyWebSocketHandler implements WebSocketHandler {
@Override
public Mono<Void> handle(WebSocketSession session) {
// 连接建立时执行的操作...
// 处理接收到的消息
Flux<WebSocketMessage> messageFlux = session.receive()
.doOnNext(message -> {
// 处理接收到的消息
System.out.println("接收到消息：" + message.getPayloadAsText());
});
// 发送消息给客户端
Flux<WebSocketMessage> outputMessageFlux = /* 构造要发送的消息 */
return session.send(messageFlux.concatWith(outputMessageFlux))
.doFinally(signalType -> {
// 连接关闭时执行的操作...
});
}
}
在上述示例中，我们通过session.receive()来接收客户端发送的消息，并使用doOnNext()操作符处理接收到的消息。您可以根据需求执行相应的业务逻辑。
我们还创建了一个Flux来构造要发送给客户端的消息，并使用session.send()将消息发送给客户端。
根据实际需求，您可以进一步扩展消息的处理和发送逻辑。
在某些场景下，需要将消息广播给多个连接或订阅者。例如，在聊天室或实时通知应用中，您可能希望将消息发送给所有在线用户。在Spring WebFlux中，我们可以使用外部容器（如Map或Set）来维护连接或订阅者列表，并在接收到消息时遍历列表，将消息发送给每个连接或订阅者。
以下是一个示例：
public class MyWebSocketHandler implements WebSocketHandler {
private static Set<WebSocketSession> sessions = Collections.synchronizedSet(new HashSet<>());
@Override
public Mono<Void> handle(WebSocketSession session) {
// 连接建立时执行的操作...
sessions.add(session);
// 处理接收到的消息...
return session.send(/* 响应消息给客户端 */)
.doFinally(signalType -> {
// 连接关闭时执行的操作...
sessions.remove(session);
});
}
public void broadcastMessage(String message) {
sessions.forEach(session -> {
// 发送消息给每个连接或订阅者
session.send(/* 构造要发送的消息 */);
});
}
}
在上述示例中，我们使用一个静态的Set来维护所有连接的会话。在连接建立时，我们将会话添加到集合中。在连接关闭时，我们从集合中移除会话。
我们还定义了一个broadcastMessage()方法，用于将消息广播给所有连接。在该方法中，我们遍历所有会话，并使用session.send()将消息发送给每个会话。
您可以根据需求扩展广播逻辑，例如只广播给特定的订阅者或根据条件过滤消息。
首先，让我们创建一个WebSocket拦截器，用于进行身份验证和日志记录：
public class WebSocketInterceptor implements WebSocketHandlerInterceptor {
@Override
public boolean beforeHandshake(ServerWebExchange exchange, WebSocketHandler handler, Map<String, Object> attributes) {
// 在握手之前执行的操作，例如身份验证
// 如果验证失败，可以通过返回false来拒绝连接
String token = exchange.getRequest().getHeaders().getFirst("Authorization");
if (isValidToken(token)) {
attributes.put("userId", extractUserIdFromToken(token));
return true;
}
return false;
}
@Override
public void afterHandshake(ServerWebExchange exchange, WebSocketHandler handler, Exception exception) {
// 在握手之后执行的操作，例如记录日志
String userId = (String) exchange.getAttributes().get("userId");
log.info("WebSocket连接建立，用户ID: {}", userId);
}
private boolean isValidToken(String token) {
// 验证令牌的逻辑...
}
private String extractUserIdFromToken(String token) {
// 从令牌中提取用户ID的逻辑...
}
}
在上述示例中，我们实现了WebSocketHandlerInterceptor接口，并重写了beforeHandshake()和afterHandshake()方法。在beforeHandshake()方法中，我们执行身份验证逻辑，并将验证通过的用户ID存储在attributes映射中。在afterHandshake()方法中，我们记录了连接建立的日志，包含了用户ID信息。
接下来，让我们创建一个自定义的消息编解码器，用于处理自定义的消息格式：
public class CustomMessageCodec implements WebSocketMessageCodec {
@Override
public List<WebSocketMessage<?>> decode(DataBuffer buffer, ResolvableType messageType,
@Nullable String mimeType, @Nullable Map<String, Object> hints) {
// 解码消息的逻辑...
}
@Override
public DataBuffer encode(WebSocketMessage<?> message, DataBufferFactory bufferFactory,
ResolvableType messageType, @Nullable String mimeType, @Nullable Map<String, Object> hints) {
// 编码消息的逻辑...
}
}
在上述示例中，我们实现了WebSocketMessageCodec接口，并重写了decode()和encode()方法。在decode()方法中，我们根据自定义的消息格式解码消息。在encode()方法中，我们根据自定义的消息格式编码消息。您可以根据实际需求自定义消息的格式和编解码逻辑。
最后，让我们创建一个WebSocket处理程序，用于处理连接和广播消息：
public class MyWebSocketHandler implements WebSocketHandler {
private static Set<WebSocketSession> sessions = Collections.synchronizedSet(new HashSet<>());
@Override
public Mono<Void> handle(WebSocketSession session) {
// 连接建立时执行的操作...
sessions.add(session);
// 处理接收到的消息
Flux<WebSocketMessage<?>> messageFlux = session.receive()
.doOnNext(message -> {
// 处理接收到的消息
String userId = (String) session.getAttributes().get("userId");
log.info("收到来自用户ID为 {} 的消息：{}", userId, message.getPayload());
});
// 发送消息给客户端
Flux<WebSocketMessage<?>> outputMessageFlux = /* 构造要发送的消息 */
return session.send(messageFlux.concatWith(outputMessageFlux))
.doFinally(signalType -> {
// 连接关闭时执行的操作...
sessions.remove(session);
});
}
public void broadcastMessage(String message) {
sessions.forEach(session -> {
// 发送消息给每个连接
session.send(/* 构造要发送的消息 */);
});
}
}
在上述示例中，我们维护了一个静态的Set来存储所有连接的会话。在连接建立时，我们将会话添加到集合中。在连接关闭时，我们从集合中移除会话。在handle()方法中，我们处理接收到的消息，并根据需要发送消息给客户端。在broadcastMessage()方法中，我们遍历所有连接的会话，并向每个会话发送广播消息。