背景

基于更復雜的情況和更高的開發要求，我們可能會遇到必須同時要使用webflux和websocket的情況。

版本約定

• JDK21
• Springboot 3.2.0
• Fastjson2
• lombok

配置文件

<parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>3.2.0</version>
</parent>
<properties><maven.compiler.source>21</maven.compiler.source><maven.compiler.target>21</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties><dependencies><!-- Spring Web --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-webflux</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency><dependency><groupId>com.alibaba.fastjson2</groupId><artifactId>fastjson2</artifactId><version>2.0.54</version></dependency>
</dependencies>

代碼

只要引入webflux，就同時引入了websocket，不需要再次引入websocket

使用webflux

@RestController
@RequestMapping("/user")
public class UserFlux {@Autowiredprivate UserService userService;@GetMapping("/get")public Mono<Result<User>> get() {return Mono.just(Result.httpSuccess(userService.getUser()));}/*** 服務器推送** @return 由服務器決定推送多少次多少數據，推送結束前不會斷開連接** @apiNote (SSE - > Server Send Event)*/@GetMapping(value = "/flux", produces = MediaType.APPLICATION_JSON_VALUE)public Flux<String> flux() {return Flux.fromStream(IntStream.range(1, 5).mapToObj(i -> {try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException ignored) {}JSONObject obj = new JSONObject();obj.put("data", "hello,flux" + i);return obj.toJSONString();}));}
}

這里比較值得注意的是Flux返回值，這個返回值從性質上說有點像會自動close的websocket。我們看下這個/flux的返回值：

{"data": "hello,flux1"
}{"data": "hello,flux2"
}{"data": "hello,flux3"
}{"data": "hello,flux4"
}

注意這不是我拼接的，是調試結果就是這樣。也就是說，/flux是分幀輸出，具有流式的特性。

使用websocket

這里選擇使用手動注冊websocket而非Endpoint自動注解，主要是因為我想對handler做規范化

配置文件

@Configuration(proxyBeanMethods = false)
public class ReactiveWebSocketConfiguration {@Beanpublic WebSocketHandlerAdapter webSocketHandlerAdapter() {return new WebSocketHandlerAdapter();}
}

handler基類

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.TypeReference;
import lombok.extern.slf4j.Slf4j;
import org.springframework.lang.NonNull;
import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.WebSocketSession;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
// 強制規定基類必須有泛型約束入參出參。強制規定必須進行參數校驗
@Slf4j
public abstract class BaseSocketHandler<T, R> implements WebSocketHandler {@Override@NonNullpublic Mono<Void> handle(WebSocketSession session) {String sessionId = session.getId();log.info("與sessionId:【{}】 建立連接", sessionId);Flux<WebSocketMessage> receive = session.receive();Flux<R> fluxHandled = receive.flatMap(webSocketMessage -> {String payloadAsText = webSocketMessage.getPayloadAsText();if (!JSON.isValid(payloadAsText)) {log.error("收取參數不合法:{}", payloadAsText);session.close();throw new IllegalArgumentException("參數不合法");}TypeReference<T> reference = getTypeReference();if (!check(payloadAsText, reference)) {log.error("參數校驗不通過:{}", payloadAsText);session.close();throw new IllegalArgumentException("參數校驗不通過");}return handler(payloadAsText, reference);}).onErrorResume(throwable -> {log.error("連接異常，即將關閉", throwable);session.close();return Mono.error(throwable);});return session.send(Mono.from(fluxHandled).map(payload -> session.textMessage(JSON.toJSONString(payload))));}public abstract boolean check(String payloadObject, TypeReference<T> typeReference);public abstract Mono<R> handler(String payload, TypeReference<T> typeReference);protected abstract TypeReference<T> getTypeReference();
}

實現類

// 這樣繼承基類的handler使用時非常簡單不說，由于上層做了處置，還會更安全更好做日志
public class NoticeHandler extends BaseSocketHandler<User, UserInfo> {@Overridepublic boolean check(String payloadObject, TypeReference<User> userTypeReference) {User user = JSON.parseObject(payloadObject, userTypeReference);return !Objects.isNull(user.getId()) && user.getId() > 0;}@Overridepublic Mono<UserInfo> handler(String payload, TypeReference<User> typeReference) {User user = JSON.parseObject(payload, typeReference);UserInfo userInfo = new UserInfo();BeanUtils.copyProperties(user, userInfo);return Mono.just(userInfo);}@Overrideprotected TypeReference<User> getTypeReference() {return new TypeReference<>() {};}
}

注冊路由

import com.xu.socket.NoticeHandler;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.handler.SimpleUrlHandlerMapping;
import org.springframework.web.reactive.socket.WebSocketHandler;import java.util.HashMap;
import java.util.Map;@Component
public class ReactiveWebSocketServerHandlerMapping extends SimpleUrlHandlerMapping {public ReactiveWebSocketServerHandlerMapping() {Map<String, WebSocketHandler> map = new HashMap<>();map.put("/ws/notice", new NoticeHandler());setUrlMap(map);setOrder(100);}
}