诉求

模拟服务端主动推送消息给客户端，同时展示客户端发送给客户端的消息以及服务端推送给客户的消息。

相关步骤

项目创建

创建一个Spring Boot项目，并在pom.xml文件中添加Websocket依赖。完整依赖如下


4.0.0org.springframework.bootspring-boot-starter-parent2.7.0com.exampledemo0.0.1-SNAPSHOTwardemoDemo project for Spring Boot8UTF-81.7.30org.springframework.bootspring-boot-starter-weborg.springframework.bootspring-boot-starter-websocketorg.springframework.bootspring-boot-starter-tomcatprovidedorg.slf4jslf4j-api${slf4j.version}org.slf4jslf4j-log4j12${slf4j.version}org.apache.logging.log4jlog4j-to-slf4j2.14.0org.apache.maven.pluginsmaven-compiler-plugin3.1${java.version}${java.version}${java.encoding}org.apache.maven.pluginsmaven-surefire-plugin2.6org.apache.maven.pluginsmaven-release-plugin-Preleaseorg.apache.maven.pluginsmaven-source-plugin2.1truecompilejar

编写相关配置

编写WebSocketConfig和WebSocketHandler配置类，实现对WebSocket的配置。

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.*;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;/*** @author * @date 2023年01月30日 14:07*/
@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {//    @Override
//    public void configureMessageBroker(MessageBrokerRegistry registry) {
//        registry.enableSimpleBroker("/topic");
//        registry.setApplicationDestinationPrefixes("/app");
//    }@Overridepublic void registerStompEndpoints(StompEndpointRegistry registry) {registry.addEndpoint("/websocket").withSockJS();}@Beanpublic ServerEndpointExporter serverEndpointExporter() {return new ServerEndpointExporter();}}

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;import java.io.IOException;
import java.util.ArrayList;
import java.util.List;/*** handler * @date 2023年01月30日 14:08*/
@Component
public class WebSocketHandler extends TextWebSocketHandler {private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketHandler.class);private static final List sessions = new ArrayList<>();@Overridepublic void handleTextMessage(WebSocketSession session, TextMessage message) {LOGGER.info("Received message: {}", message.getPayload());for (WebSocketSession webSocketSession : sessions) {try {webSocketSession.sendMessage(message);} catch (IOException e) {LOGGER.error("Error: {}", e.getMessage());}}}@Overridepublic void afterConnectionEstablished(WebSocketSession session) {sessions.add(session);}@Overridepublic void afterConnectionClosed(WebSocketSession session, CloseStatus status) {sessions.remove(session);}
}

消息逻辑处理

编写一个WebSocketServer类，实现服务端主动推送消息的逻辑(此处服务端推送客户端消息的逻辑采用定时任务自动推送，⚠️注意:必须在启动类中加入@EnableScheduling注解，开启定时任务的支持)。


import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;/*** @author * @date 2023年01月30日 11:19*/
@ServerEndpoint("/websocket")
@Component
public class WebSocketServer {private Session session;/*     用于存储websocket连接，key为sessionId  */private static ConcurrentHashMap webSocketServerConcurrentHashMap = new ConcurrentHashMap();@OnOpenpublic void onOpen(Session session) {this.session = session;webSocketServerConcurrentHashMap.put(session.getId(), this);System.out.println("WebSocket opened: " + session.getId());}@OnMessagepublic void onMessage(String message, Session session) {System.out.println("WebSocket message received: " + message);String dateStr = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(System.currentTimeMillis());try {//发送的消息也返回给当前连接，用于展示session.getBasicRemote().sendText(dateStr + "发送消息:" + message);//写入DB或者其他存储系统中。。。} catch (IOException e) {e.printStackTrace();}}@OnClosepublic void onClose(Session session, CloseReason closeReason) {webSocketServerConcurrentHashMap.remove(session.getId());System.out.println("WebSocket closed: " + closeReason);}@OnErrorpublic void onError(Session session, Throwable throwable) {System.out.println("WebSocket error: " + throwable);}/*** 模拟服务端消息推送,5s推送一次(服务端 -> 客户端)*/@Scheduled(fixedRate = 5000)public void sendMessageToClient() {//没有连接时不做任何事情if (CollectionUtils.isEmpty(webSocketServerConcurrentHashMap)){return;}System.out.println("服务端发送消息到客户端");String dateStr = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(System.currentTimeMillis());long number = new Random().nextInt(10000);webSocketServerConcurrentHashMap.forEach((k, v) -> {try {v.session.getBasicRemote().sendText(dateStr + "收到消息:" + number);//写入DB或者其他存储系统中。。。} catch (IOException e) {e.printStackTrace();}});}}