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

< 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 >

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();

     }

}

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--;

     }

}

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;

     }

}