一、引入依赖

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

二、添加配置

新增配置文件

config/WebSocketConfig.java

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;@Configuration
public class WebSocketConfig {/*** 这个Bean的作用是自动注册使用了@ServerEndpoint注解的Bean*/@Beanpublic ServerEndpointExporter serverEndpointExporter(){return new ServerEndpointExporter();}
}

三、新建互通类

/*** description:  平台同步*/
@Component
@Slf4j
// 类似于controlelr 服务点
@ServerEndpoint(value = "/webSocket/{username}")
public class PlatformAsyncWebSocket {/**静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。*/private static int onlineCount = 0;/**concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。*/private static ConcurrentHashMap<String,PlatformAsyncWebSocket> webSocketMap = new ConcurrentHashMap<>();/**与某个客户端的连接会话，需要通过它来给客户端发送数据*/private Session session;/**接收userId*/private String username="";/*** 连接建立成功调用的方法*/@OnOpenpublic void onOpen(Session session,@PathParam("username") String username) {this.session = session;this.username=username;if(webSocketMap.containsKey(username)){webSocketMap.remove(username);webSocketMap.put(username,this);//加入set中}else{webSocketMap.put(username,this);//加入set中addOnlineCount();//在线数加1}log.info("用户连接:"+username+",当前在线人数为:" + getOnlineCount());try {sendMessage("连接成功");} catch (IOException e) {log.error("用户:"+username+",网络异常!!!!!!");}}/*** 连接关闭调用的方法*/@OnClosepublic void onClose() {if(webSocketMap.containsKey(username)){webSocketMap.remove(username);//从set中删除subOnlineCount();}log.info("用户退出:"+username+",当前在线人数为:" + getOnlineCount());}/*** 收到客户端消息后调用的方法** @param message 客户端发送过来的消息*/@OnMessagepublic void onMessage(String message, Session session) {log.info("用户消息:"+username+",报文:"+message);//可以群发消息//消息保存到数据库、redisif(StringUtils.isNotBlank(message)){try {//解析发送的报文JSONObject jsonObject = JSON.parseObject(message);//追加发送人(防止串改)jsonObject.put("fromUserId",this.username);String toUserId=jsonObject.getString("toUserId");//传送给对应toUserId用户的websocketif(StringUtils.isNotBlank(toUserId)&&webSocketMap.containsKey(toUserId)){webSocketMap.get(toUserId).sendMessage(jsonObject.toJSONString());}else{log.error("请求的userId:"+toUserId+"不在该服务器上");//否则不在这个服务器上，发送到mysql或者redis}}catch (Exception e){e.printStackTrace();}}}/**** @param session* @param error*/@OnErrorpublic void onError(Session session, Throwable error) {log.error("用户错误:"+this.username+",原因:"+error.getMessage());error.printStackTrace();}/*** 实现服务器主动推送*/public void sendMessage(String message) throws IOException {this.session.getBasicRemote().sendText(message);}/*** 发送自定义消息* */public static void sendMsg(String message,@PathParam("username") String username) throws IOException {log.info("发送消息到:"+username+"，报文:"+message);if(StringUtils.isNotBlank(username)&&webSocketMap.containsKey(username)){webSocketMap.get(username).sendMessage(message);}else{log.error("用户"+username+",不在线！");}}public static synchronized int getOnlineCount() {return onlineCount;}public static synchronized void addOnlineCount() {PlatformAsyncWebSocket.onlineCount++;}public static synchronized void subOnlineCount() {PlatformAsyncWebSocket.onlineCount--;}}