springboot整合WebSocket

WebSocket通信过程

客户端构建一个websocket实例，并且为它绑定一个需要连接到的服务器地址，当客户端连接服务端的候，会向服务端发送一个http get报文，告诉服务端需要将通信协议切换到websocket，服务端收到http请求后将通信协议切换到websocket,同时发给客户端一个响应报文，返回的状态码为101，表示同意客户端协议转请求，并转换为websocket协议。以上过程都是利用http通信完成的，称之为websocket协议握手(websocket Protocol handshake)，经过握手之后，客户端和服务端就建立了websocket连接，以后的通信走的都是websocket协议了。

1.pom文件添加依赖

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

2.启用Springboot对WebSocket的支持

package com.lby.websocket.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

/**
 * @author Liby
 * @date 2022-04-25 16:18
 * @description:
 * @version:
 */
@Configuration
public class WebSocketConfig {

    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

3.核心配置：WebSocketServer

因为WebSocket是类似客户端服务端的形式(采用ws协议)，那么这里的WebSocketServer其实就相当于一个ws协议的Controller

• @ ServerEndpoint 注解是一个类层次的注解，它的功能主要是将目前的类定义成一个websocket服务器端, 注解的值将被用于监听用户连接的终端访问URL地址,客户端可以通过这个URL来连接到WebSocket服务器端
• 新建一个ConcurrentHashMap webSocketMap 用于接收当前userId的WebSocket，方便传递之间对userId进行推送消息。

下面是具体业务代码：

package com.lby.websocket.component;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.stereotype.Component;
import cn.hutool.core.util.StrUtil;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Liby
 * @date 2022-04-25 16:21
 * @description:
 * @version:
 */
@ServerEndpoint(value = "/websocket/{userId}")
@Component
public class WebSocket {
    private final static Logger logger = LogManager.getLogger(WebSocket.class);

    /**
     * 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的
     */

    private static int onlineCount = 0;

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

    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
     */

    private Session session;
    private String userId;


    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        this.session = session;
        this.userId = userId;
        //加入map
        webSocketMap.put(userId, this);
        addOnlineCount();           //在线数加1
        logger.info("用户{}连接成功,当前在线人数为{}", userId, getOnlineCount());
        try {
            sendMessage(String.valueOf(this.session.getQueryString()));
        } catch (IOException e) {
            logger.error("IO异常");
        }
    }


    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        //从map中删除
        webSocketMap.remove(userId);
        subOnlineCount();           //在线数减1
        logger.info("用户{}关闭连接！当前在线人数为{}", userId, getOnlineCount());
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        logger.info("来自客户端用户：{} 消息:{}",userId, message);

        //群发消息
        /*for (String item : webSocketMap.keySet()) {
            try {
                webSocketMap.get(item).sendMessage(message);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }*/
    }

    /**
     * 发生错误时调用
     *
     * @OnError
     */
    @OnError
    public void onError(Session session, Throwable error) {
        logger.error("用户错误:" + this.userId + ",原因:" + error.getMessage());
        error.printStackTrace();
    }

    /**
     * 向客户端发送消息
     */
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
        //this.session.getAsyncRemote().sendText(message);
    }

    /**
     * 通过userId向客户端发送消息
     */
    public void sendMessageByUserId(String userId, String message) throws IOException {
        logger.info("服务端发送消息到{},消息：{}",userId,message);
     if(StrUtil.isNotBlank(userId)&&webSocketMap.containsKey(userId)){
         webSocketMap.get(userId).sendMessage(message);
     }else{
         logger.error("用户{}不在线",userId);
     }

    }

    /**
     * 群发自定义消息
     */
    public static void sendInfo(String message) throws IOException {
        for (String item : webSocketMap.keySet()) {
            try {
                webSocketMap.get(item).sendMessage(message);
            } catch (IOException e) {
                continue;
            }
        }
    }

    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        WebSocket.onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        WebSocket.onlineCount--;
    }

}

4.测试controller

package com.lby.websocket.controller;

import com.lby.websocket.component.WebSocket;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.IOException;

/**
 * @author Liby
 * @date 2022-04-26 10:28
 * @description:建立WebSocket连接
 * @version:
 */
@RestController
@RequestMapping("/webSocket")
public class WebSocketController {
    @Autowired
    private WebSocket webSocket;
    @PostMapping("/sentMessage")
    public void sentMessage(String userId,String message){
        try {
            webSocket.sendMessageByUserId(userId,message);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

}

5.webSocket网页客户端工具

http://websocket.jsonin.com/

复制三个tab

WebSocket地址分别输入后连接

ws://127.0.0.1:8092/websocket/1
ws://127.0.0.1.132:8092/websocket/2
ws://127.0.0.1.132:8092/websocket/3

6.通讯测试

1.往客户端发消息

127.0.0.1:8092/webSocket/sentMessage?userId=1&message=请进入视频会议
127.0.0.1:8092/webSocket/sentMessage?userId=2&message=请进入视频会议
127.0.0.1:8092/webSocket/sentMessage?userId=3&message=请进入视频会议

2.客户端收到消息回复“好的”

Springboot整合Websocket

引入maven

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

配置springboot

server:
    port: 5003

编写 ApplicationConfig 类配置websocket

@Configuration
public class ApplicationConfig {

    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

核心代码

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

测试代码

@Component
@Slf4j
@ServerEndpoint("/api/pushMessage/{userId}")
public class MyWebsocket {
    private Session session;

    private String userId;

    private static ConcurrentHashMap<String,MyWebsocket> webSocketSet = new ConcurrentHashMap<>();
    /**
     * 连接建立成功调用的方法
     * session为与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    @OnOpen
    public void OnOpen(Session session, @PathParam(value = "userId") String userId){
        log.info("----------------------------------");
        this.session = session;
        this.userId = userId;
        // userId是用来表示唯一客户端，如果需要指定发送，需要指定发送通过userId来区分
        webSocketSet.put(userId,this);
        log.info("[WebSocket] 连接成功，当前连接人数为：={}",webSocketSet.size());
        log.info("----------------------------------");
        log.info("");

        GroupSending(userId+" 来了");
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void OnClose(){
        webSocketSet.remove(this.userId);
        log.info("[WebSocket] 退出成功，当前连接人数为：={}",webSocketSet.size());

        GroupSending(userId+" 走了");
    }

    /**
     * 收到客户端消息后调用的方法
     */
    @OnMessage
    public void OnMessage(String message_str){
        log.info("[WebSocket] 收到消息：{}",message_str);
        //判断是否需要指定发送，具体规则自定义
        //message_str的格式 TOUSER:user2;message:aaaaaaaaaaaaaaaaaa;
        if(message_str.indexOf("TOUSER") == 0){
            //取出 name和message的值
            String[] split = message_str.split(";");
            String[] split1 = split[0].split(":");
            String[] split2 = split[1].split(":");
            String name = split1[1];
            String message = split2[1];
            //指定发送
            AppointSending(name,message);
        }else{
            //群发
            GroupSending(message_str);
        }
    }

    /**
     * 发生错误时调用
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error){
        log.info("发生错误");
        error.printStackTrace();
    }

    /**
     * 群发
     * @param message
     */
    public void GroupSending(String message){
        for (String name : webSocketSet.keySet()){
            try {
                webSocketSet.get(name).session.getBasicRemote().sendText(message);
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }

    /**
     * 指定发送
     * @param name
     * @param message
     */
    public void AppointSending(String name,String message){
        try {
            webSocketSet.get(name).session.getBasicRemote().sendText(message);
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

测试websocket

在线测试websocket网站： http://coolaf.com/tool/chattest

springboot整合websocket（一）简单聊天室

springboot整合websocket（一）简单聊天室
springboot整合websocket（二）聊天室补充篇

springboot整合websocket（三）上传文件（引导篇）
springboot整合websocket（四）上传文件（终篇）

东西太多了，拆成几章来写（绝对不是骗流量^ _w ^）

这一部分就简单做一个公共聊天室吧

1、引入相关依赖

springboot相关依赖就不写了，这里只写websocket的依赖

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

2、写配置文件

这里注意不要漏了 @EnableWebSocket，用于开启websocket支持，同时 @Configuration将配置类注入spring容器

package com.websocket.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
import org.springframework.web.socket.server.standard.ServletServerContainerFactoryBean;

@Configuration
//开启websocket支持
@EnableWebSocket
public class WebsocketConfig
{
    /**
     * 必须要有的
     *
     * @return
     */
    @Bean
    public ServerEndpointExporter serverEndpointExporter()
    {
        return new ServerEndpointExporter();
    }

    /**
     * websocket 配置信息
     *
     * @return
     */
    @Bean
    public ServletServerContainerFactoryBean createWebSocketContainer()
    {
        ServletServerContainerFactoryBean bean = new ServletServerContainerFactoryBean();
        //文本缓冲区大小
        bean.setMaxTextMessageBufferSize(8192);
        //字节缓冲区大小
        bean.setMaxBinaryMessageBufferSize(8192);
        return bean;
    }
}

3、开始愉快的使用啦

这里先介绍几个websocket的注解

3.1（重要） 将@ServerEndpoint标注在类上，然后依次创建4个方法，参数见上表，方法名随意

注意类上需要有 @Component 扫描哦，我这里用的时@Controller

@Log4j2
@Controller
@ServerEndpoint("/websocket")
public class BaseWebsocketController
{

	//使用 ConcurrentHashMap, 保证线程安全, static全局共享 session
	
	//这里之所以static，是因为这个类不是单例的！！
	//他虽然有@Controller注解，但是不适用Ioc容器中拿对象，每一次请求过来都是一个新的对象
	
    //存放 session
    private final static Map<String, Session> sessions = new ConcurrentHashMap<>();

    //onopen 在连接创建(用户进入聊天室)时触发
    @OnOpen
    public void openSession(Session session, EndpointConfig config)
    {
        
    }

	//响应字符串
    @OnMessage
    public void onMessage(Session session, String message)
    {
        
    }

	//响应字节流
    @OnMessage
    public void onMessage(Session session, byte[] message)
    {
        
    }

    //onclose 在连接断开(用户离开聊天室)时触发
    @OnClose
    public void closeSession(Session session, CloseReason closeReason)
    {
        
    }

    @OnError
    public void sessionError(Session session, Throwable throwable)
    {
        
    }
}

说明

细心的小伙伴可能发现，我有两个 @OnMessage， 这是因为websocket能发送三种请求（我知道的三种），一种是字符串，一种是字节流（用于上传文件），一种是ping-pong（乒乓机制），因为js不好发送ping请求，我这里就只有响应字符串和字节流两种方法。

接下来的篇幅将只演示字符串的，字节流咱再另一篇说，不然太多了看的头痛

3.2 往方法里面写点简单的东西

里面注释写个很清楚哦，慢慢瞅，咱就解释一下流程：
1、再OnOpen中将session存起来，并通知其他用户，有人来啦。
2、有消息来的时候，再OnMessage中，通知其他用户
3、OnClose中，通知其他用户，别人溜了
4、OnError中，有异常就关闭websocket

@Log4j2
@Controller
@ServerEndpoint("/websocket")
public class BaseWebsocketController
{

    //使用 ConcurrentHashMap, 保证线程安全, static全局共享 session

    //这里之所以static，是因为这个类不是单例的！！
    //他虽然有@Controller注解，但是不适用Ioc容器中拿对象，每一次请求过来都是一个新的对象

    //存放 session
    private final static Map<String, Session> sessions = new ConcurrentHashMap<>();

    //onopen 在连接创建(用户进入聊天室)时触发
    @OnOpen
    public void openSession(Session session, EndpointConfig config)
    {
        //将session存起来, 用于服务器向浏览器发送消息
        sessions.put(session.getId(), session);
        sendAll("[" + session.getId() + "]进入房间");
    }

    //响应字符串
    @OnMessage
    public void onMessage(Session session, String message)
    {
        sendAll("[" + session.getId() + "]" + message);
    }

    //响应字节流
    @OnMessage
    public void onMessage(Session session, byte[] message)
    {
        //这个咱以后再说
    }

    //onclose 在连接断开(用户离开聊天室)时触发
    @OnClose
    public void closeSession(Session session, CloseReason closeReason)
    {
        //记得移除相对应的session
        sessions.remove(session.getId());
        
        sendAll("[" + session.getId() + "]离开了房间");
    }

    @OnError
    public void sessionError(Session session, Throwable throwable)
    {
        //通常有异常会关闭session
        try {
            session.close();
        }
        catch (IOException e) {
            e.printStackTrace();
        }
    }

    private void sendAll(String message)
    {
        for (Session s : sessions.values()) {
            //获得session发送消息的对象
            //Basic是同步, 会阻塞
            //Async是异步, 这个会有多线程并发导致异常, 发送消息太快也会有并发异常, 需要有 消息队列 来辅助使用
            final RemoteEndpoint.Basic remote = s.getBasicRemote();
            try {
                //发送消息
                remote.sendText(message);
            }
            catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

3.3 写个页面使用websocket

先上效果图

<!DOCTYPE html>
<html xmlns:th="http://www.thymeleaf.org" lang="en">
<head>
  <meta charset="UTF-8">
  <title>websocket-demo</title>
  
  <link rel="stylesheet" href="https://cdn.bootcdn.net/ajax/libs/twitter-bootstrap/4.2.1/css/bootstrap.min.css">
</head>
<body>
  <div class="container py-3">
    
    <div class="row">
      
      <div class="col-6">
        <div>
          <label for="messageArea">聊天信息:</label>
        </div>
        <div>
          <textarea id="messageArea" readonly class="w-100" style="height: 75vh;"></textarea>
        </div>
      </div>
      
      <div class="col">
        
        <div class="my-1">
          <label for="messageArea">用 户 名:</label>
        </div>
        
        <div class="my-1">
          <input type="text" id="username" autocomplete="off">
        </div>
        
        <div class="my-1">
          <button class="btn-info" id="joinRoomBtn">进入聊天室</button>
          <button class="btn-warning" id="leaveRoomBtn">离开聊天室</button>
        </div>
        
        <hr/>
        
        <div class="my-1">
          <label for="sendMessage">输入消息:</label>
        </div>
        <div>
          <textarea id="sendMessage" rows="5" class="w-100" style="max-height: 50vh"></textarea>
        </div>
        
        <div class="my-1">
          <button class="btn-primary" id="sendBtn">发送消息</button>
        </div>
      
      </div>
    
    </div>
  
  </div>
  
  <script src="https://s3.pstatp.com/cdn/expire-1-M/jquery/3.3.1/jquery.min.js"></script>
  
  <script>
    let webSocket;
    //ip和端口号用自己项目的
    //{websocket}: 其实是刚刚那个@ServerEndpoint("/websocket")中定义的
    let url = 'ws://127.0.0.1:8080/websocket';
    
    $('#username').keyup(function (e) {
      let keycode = e.which;
      if (keycode == 13) {
        $('#joinRoomBtn').click();
      }
    });
    
    //进入聊天室
    $('#joinRoomBtn').click(function () {
      let username = $('#username').val();
      webSocket = new WebSocket(url);
      webSocket.onopen = function () {
        console.log('webSocket连接创建。。。');
      }
      webSocket.onclose = function () {
        console.log('webSocket已断开。。。');
        $('#messageArea').append('websocket已断开\n');
      }
      webSocket.onmessage = function (event) {
        $('#messageArea').append(event.data + '\n');
      }
      webSocket.onerror = function (event) {
        console.log(event)
        console.log('webSocket连接异常。。。');
      }
    });
    
    //退出聊天室
    $('#leaveRoomBtn').click(function () {
      if (webSocket) {
        //关闭连接
        webSocket.close();
      }
    });
    
    //发送消息
    $('#sendBtn').click(function () {
      var msg = $('#sendMessage').val();
      if (msg.trim().length === 0) {
        alert('请输入内容');
        return;
      }
      webSocket.send($('#sendMessage').val());
      
      $('#sendMessage').val('');
    });
  
  </script>

</body>
</html>

4、最后填下坑

4.1@OnOpen、@OnMessage…那些参数是咋来滴？

当然是看注释啦！转到源码，再类的上边有一大串注释，百度翻译看下就行。
我上边的参数还有漏的哦~~太多了解释不过来。

4.2 里面有一段是这个

and Zero to n String or Java primitive parameters
annotated with the {@link javax.websocket.server.PathParam} annotation for server endpoints.

意思是可以再路径中加入参数，然后用 @PathParam注解获得该参数（是不是和web很像呢）

也就是说，可以这样

@ServerEndpoint("/websocket/{username}")

然后这样获得参数（所有方法都可以加），而且可以不止一个

//响应字符串
@OnMessage
public void onMessage(@PathParam("username") String username, Session session, String message)
{
    sendAll("[" + username + "]" + message);
}

再html连接时，url就这么写

let username = $('#username').val();
let url = 'ws://127.0.0.1:8080/websocket';
url = url+'/'+username;

所以啊，刚刚我们demo的用户名是可以显示出来的（这个就交给大家了，懒 _{>_<}）。

End

springboot整合websocket（一）简单聊天室
springboot整合websocket（二）聊天室补充篇

springboot整合websocket（三）上传文件（引导篇）
springboot整合websocket（四）上传文件（终篇）