springboot整合websocket简单版本

野性酷女 2024-04-01 17:40 16阅读 0赞

这是springboot整合的简单的websocket,可实现简单的前后端实时通讯。

第一步：添加依赖

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

第二步：WebSocket配置类

mport org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.web.socket.server.standard.ServerEndpointExporter;
    
    @Configuration
    public class WebSocketConfig {
        /**
         * 	注入ServerEndpointExporter，
         * 	这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint
         */
        @Bean
        public ServerEndpointExporter serverEndpointExporter() {
            return new ServerEndpointExporter();
        }
        
    }

第三步：WebSocket操作类，通过该类WebSocket可以进行群推送以及单点推送

WebSocketServer ：下面是websocket的重点

@ServerEndpoint("/api/pushMessage/\{userId\}") 前端通过此 URI 和后端交互，建立连接  
@Component 不用说将此类交给 spring 管理  
@OnOpen websocket 建立连接的注解，前端触发上面 URI 时会进入此注解标注的方法  
@OnMessage 收到前端传来的消息后执行的方法  
@OnClose 顾名思义关闭连接，销毁 session  
因为WebSocket是类似客户端服务端的形式(采用ws协议)，那么这里的WebSocketServer其实就相当于一个ws协议的Controller  
新建一个ConcurrentHashMap webSocketMap 用于接收当前userId的WebSocket，方便IM之间对userId进行推送消息  
核心代码：

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

## 消息推送 ##

至于推送新信息，可以在自己的Controller写个方法调用WebSocketServer.sendInfo()即可。  
程序中使用定任务不停的向客户端发送消息。

@Controller
    @RequestMapping("/api/test")
    @Api(description = "服务器向客户端推送消息接口", tags = "Test")
    public class TestController {
    
        @Autowired
        private TestServiceImpl testServiceImpl;
        /**
         * 启动页面
         * @return
         */
        @GetMapping("/start")
        public String start(){
            return "index";
        }
    
        @PostMapping("/pushToWeb")
        @ApiOperation(value = "服务器端向客户端推送消息", notes = "服务器端向客户端推送消息")
        public ResponseBean<?> pushToWeb(@RequestBody @ApiParam(value = "回收人编码和医院编码", required = true) CodesInfo info){
    
            testServiceImpl.printTime();
            return new ResponseBean<>(200, "success", "123456");
        }
    
    }

@Service
    @EnableScheduling
    public class TestServiceImpl {
    
        //打印时间
        @Scheduled(fixedRate=1000) //1000毫秒执行一次
        public  void  printTime(){
    
            SimpleDateFormat dateFormat = new SimpleDateFormat("HH:mm:ss");
            String date = dateFormat.format(new Date());
            WebSocketServer.sendInfo(date,"10");
            System.out.println(date);
        }
    
    }

## 建立websocket连接 ##

前端页面直接使用ws协议建立连接，const socketUrl = “**ws://localhost:9091/api/pushMessage/**” + $("\#userId").val();

<!DOCTYPE html>
    <html>
    <head>
        <meta charset="utf-8">
        <title>websocket通讯</title>
    </head>
    <script src="https://cdn.bootcss.com/jquery/3.3.1/jquery.js"></script>
    <script>
        let socket;
        function openSocket() {
    
            const socketUrl = "ws://localhost:9091/api/pushMessage/" + $("#userId").val();
            console.log(socketUrl);
            if(socket!=null){
                socket.close();
                socket=null;
            }
            socket = new WebSocket(socketUrl);
            //打开事件
            socket.onopen = function() {
                console.log("websocket已打开");
            };
            //获得消息事件
            socket.onmessage = function(msg) {
                console.log(msg.data);
                //发现消息进入,开始处理前端触发逻辑
            };
            //关闭事件
            socket.onclose = function() {
                console.log("websocket已关闭");
            };
            //发生了错误事件
            socket.onerror = function() {
                console.log("websocket发生了错误");
            }
        }
        function sendMessage() {
    
            socket.send('{"toUserId":"'+$("#toUserId").val()+'","contentText":"'+$("#contentText").val()+'"}');
            console.log('{"toUserId":"'+$("#toUserId").val()+'","contentText":"'+$("#contentText").val()+'"}');
        }
    </script>
    <body>
    <p>【socket开启者的ID信息】：<div><input id="userId" name="userId" type="text" value="10"></div>
    <p>【客户端向服务器发送的内容】：<div><input id="toUserId" name="toUserId" type="text" value="20">
        <input id="contentText" name="contentText" type="text" value="hello websocket"></div>
    <p>【操作】：<div><a onclick="openSocket()">开启socket</a></div>
    <p>【操作】：<div><a onclick="sendMessage()">发送消息</a></div>
    </body>
    
    </html>

如果是发布到服务器上，用nginx的话，需要配置支持websocket

location  /prod-api {
          #proxy_pass              https://debug.saizhuge.net/api/;
          proxy_set_header Host $http_host;
    	  proxy_set_header X-Real-IP $remote_addr;
    	  proxy_set_header REMOTE-HOST $remote_addr;
    	  proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    	  proxy_pass http://127.0.0.1:9080/prod-api;
    	  proxy_send_timeout 1800;
          proxy_read_timeout 1800;
          proxy_connect_timeout 1800;
          #启用支持websocket连接
          proxy_set_header Upgrade $http_upgrade;
          proxy_set_header Connection "upgrade";
        }

这样就完成了！