SpringBoot实现WebSocket即时通讯的示例代码

1、引入依赖

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.3</version>
</dependency>

2、WebSocketConfig 开启WebSocket

package com.shucha.deveiface.web.config;

/**
* @author tqf
* @Description
* @Version 1.0
* @since 2022-04-12 15:35
*/
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

/**
* 开启WebSocket
*/
@Configuration
public class WebSocketConfig {
   @Bean
   public ServerEndpointExporter serverEndpointExporter(){
       return new ServerEndpointExporter();
   }
}

3、WebSocketServer

package com.shucha.deveiface.web.ws;

/**
* @author tqf
* @Description
* @Version 1.0
* @since 2022-04-12 15:33
*/
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.WebSocketSession;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

@Component
@ServerEndpoint("/webSocket/{userId}")
@Slf4j
public class WebSocketServer {
   private Session session;
   private String userId;
   /**静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。*/
   private static int onlineCount = 0;
   private static CopyOnWriteArraySet<WebSocketServer> webSocketSet = new CopyOnWriteArraySet<>();

/**
    * concurrent包的线程安全set，用来存放每个客户端对应的MyWebSocket对象
    */
   private static ConcurrentHashMap<String,WebSocketServer> webSocketMap = new ConcurrentHashMap();

/**
    * 为了保存在线用户信息，在方法中新建一个list存储一下【实际项目依据复杂度，可以存储到数据库或者缓存】
    */
   private final static List<Session> SESSIONS = Collections.synchronizedList(new ArrayList<>());

/**
    * 建立连接
    * @param session
    * @param userId
    */
   @OnOpen
   public void onOpen(Session session, @PathParam("userId") String userId) {
       this.session = session;
       this.userId = userId;
       webSocketSet.add(this);
       SESSIONS.add(session);
       if (webSocketMap.containsKey(userId)) {
           webSocketMap.remove(userId);
           webSocketMap.put(userId,this);
       } else {
           webSocketMap.put(userId,this);
           addOnlineCount();
       }
       // log.info("【websocket消息】有新的连接, 总数:{}", webSocketSet.size());
       log.info("[连接ID:{}] 建立连接, 当前连接数:{}", this.userId, webSocketMap.size());
   }

/**
    * 断开连接
    */
   @OnClose
   public void onClose() {
       webSocketSet.remove(this);
       if (webSocketMap.containsKey(userId)) {
           webSocketMap.remove(userId);
           subOnlineCount();
       }
       // log.info("【websocket消息】连接断开, 总数:{}", webSocketSet.size());
       log.info("[连接ID:{}] 断开连接, 当前连接数:{}", userId, webSocketMap.size());
   }

/**
    * 发送错误
    * @param session
    * @param error
    */
   @OnError
   public void onError(Session session, Throwable error) {
       log.info("[连接ID:{}] 错误原因:{}", this.userId, error.getMessage());
       error.printStackTrace();
   }

/**
    * 收到消息
    * @param message
    */
   @OnMessage
   public void onMessage(String message) {
       // log.info("【websocket消息】收到客户端发来的消息:{}", message);
       log.info("[连接ID:{}] 收到消息:{}", this.userId, message);
   }

/**
    * 发送消息
    * @param message
    * @param userId
    */
   public void sendMessage(String message,Long userId) {
       WebSocketServer webSocketServer = webSocketMap.get(String.valueOf(userId));
       if (webSocketServer!=null){
           log.info("【websocket消息】推送消息, message={}", message);
           try {
               webSocketServer.session.getBasicRemote().sendText(message);
           } catch (Exception e) {
               e.printStackTrace();
               log.error("[连接ID:{}] 发送消息失败, 消息:{}", this.userId, message, e);
           }
       }
   }

/**
    * 群发消息
    * @param message
    */
   public void sendMassMessage(String message) {
       try {
           for (Session session : SESSIONS) {
               if (session.isOpen()) {
                   session.getBasicRemote().sendText(message);
                   log.info("[连接ID:{}] 发送消息:{}",session.getRequestParameterMap().get("userId"),message);
               }
           }
       } catch (Exception e) {
           e.printStackTrace();
       }
   }

/**
    * 获取当前连接数
    * @return
    */
   public static synchronized int getOnlineCount() {
       return onlineCount;
   }

/**
    * 当前连接数加一
    */
   public static synchronized void addOnlineCount() {
       WebSocketServer.onlineCount++;
   }

/**
    * 当前连接数减一
    */
   public static synchronized void subOnlineCount() {
       WebSocketServer.onlineCount--;
   }

}

4、测试连接发送和接收消息

package com.shucha.deveiface.web.controller;

import com.alibaba.fastjson.JSONObject;
import com.shucha.deveiface.web.ws.WebSocketServer;
import lombok.Data;
import lombok.experimental.Accessors;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
* @author tqf
* @Description
* @Version 1.0
* @since 2022-04-12 15:44
*/
@RestController
@RequestMapping("/web")
public class TestWebSocket {
   @Autowired
   private WebSocketServer webSocketServer;

/**
    * 消息发送测试
    */
   @GetMapping("/test")
   public void test(){
       for (int i=1;i<4;i++) {
           WebsocketResponse response = new WebsocketResponse();
           response.setUserId(String.valueOf(i));
           response.setUserName("姓名"+ i);
           response.setAge(i);
           webSocketServer.sendMessage(JSONObject.toJSONString(response), Long.valueOf(String.valueOf(i)));
       }
   }

/**
    * 群发消息测试(给当前连接用户发送)
    */
   @GetMapping("/sendMassMessage")
   public void sendMassMessage(){
       WebsocketResponse response = new WebsocketResponse();
       response.setUserName("群发消息模板测试");
       webSocketServer.sendMassMessage(JSONObject.toJSONString(response));
   }

@Data
   @Accessors(chain = true)
   public static class WebsocketResponse {
       private String userId;
       private String userName;
       private int age;
   }
}

5、在线测试地址

websocket 在线测试

6、测试截图

访问测试发送消息：http://localhost:50041//web/test

测试访问地址：ws://192.168.0.115:50041/webSocket/1   wss://192.168.0.115:50041/webSocket/2

来源：https://blog.csdn.net/tanqingfu1/article/details/124127295