SpringBoot整合Websocket（Java websocket怎么使用）

1 Websocket是什么

WebSocket 是一种基于 TCP 协议的全双工通信协议，可以在浏览器和服务器之间建立实时、双向的数据通信。可以用于在线聊天、在线游戏、实时数据展示等场景。与传统的 HTTP 协议不同，WebSocket 可以保持长连接，实时传输数据，避免了频繁的 HTTP 请求和响应，节省了网络带宽和服务器资源，提高了应用程序的性能和用户体验。

2 Websocket可以做什么

项目中大部分的请求都是前台主动发送给后台，后台接收后返回数据给前台，返回数据后这个连接就终止了。如果要实现实时通信，通用的方式是采用 HTTP 协议

不断发送请求。但这种方式即浪费带宽（HTTP HEAD 是比较大的），又消耗服务器 CPU 占用（没有信息也要接受请求）。

websocket可以建立长连接实现双向通信，客户端和服务端都可以主动的向对方发送消息。

例如：

假设张三今天有个快递快到了，但是张三忍耐不住，就每隔十分钟给快递员或者快递站打电话，询问快递到了没，每次快递员就说还没到，等到下午张三的快递到了，但是，快递员不知道哪个电话是张三的，（可不是只有张三打电话，还有李四，王五），所以只能等张三打电话，才能通知他，你的快递到了。

而最好的情况是，张三给快递员第一次打电话时，说明自己的身份，快递员记录下来，让自己和快递员之间形成一对一的关系可以互相联系到。张三也不用再次给快递员打电话了，快递到了快递员会主动联系张三通知他来取。

后者就是websocket模式，在客户端断开WebSocket连接或Server端中断连接前，不需要客户端和服务端重新发起连接请求。在海量并发及客户端与服务器交互负载流量大的情况下，极大的节省了网络带宽资源的消耗，有明显的性能优势，且客户端发送和接受消息是在同一个持久连接上发起，实现了“真·长链接”，实时性优势明显。

在项目中聊天功能也是类似的逻辑，A发送了消息B立刻就要收到，A和B都属于前台客户端，不可能直接从一个前台不走服务器传输给另一个前台，过程一定是前台 —> 服务器 -> 前台。那前台客户端B接收消息是被动的，需要服务器主动发送消息请求，这就用到了WebSocket。大体流程如下图：

3 Springboot整合Websocket

3.1 服务端

1. 添加依赖

   <dependency>
       <groupId>org.springframework.bootgroupId>
       <artifactId>spring-boot-starter-websocketartifactId>
   dependency>
   1
   2
   3
   4

2. 添加Websocket配置文件

   @Configuration
   public class WebSocketConfig {
       /**
        *     注入ServerEndpointExporter，
        *     这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket
        */
       @Bean
       public ServerEndpointExporter serverEndpointExporter() {
           return new ServerEndpointExporter();
       }
   }
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11

3. Webscoket操作类

   import org.springframework.stereotype.Component;
   import javax.websocket.*;
   import javax.websocket.server.PathParam;
   import javax.websocket.server.ServerEndpoint;
   import java.util.concurrent.ConcurrentHashMap;
   import java.util.concurrent.CopyOnWriteArraySet;
   import java.util.logging.Logger;
   import static java.util.logging.Level.WARNING;
   
   @Component
   @ServerEndpoint("/websocket/{userId}")  // 接口路径 ws://localhost:9001/webSocket/userId;
   public class WebSocket {
   
       private static final Logger log = Logger.getLogger(WebSocket.class.getName());
   
       //与某个客户端的连接会话，需要通过它来给客户端发送数据
       private Session session;
       /**
        * 用户ID
        */
       private String userId;
   
       //concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
       //虽然@Component默认是单例模式的，但springboot还是会为每个websocket连接初始化一个bean，所以可以用一个静态set保存起来。
       //  注：底下WebSocket是当前类名
       private static CopyOnWriteArraySet<WebSocket> webSockets =new CopyOnWriteArraySet<>();
       // 用来存在线连接用户信息
       private static ConcurrentHashMap<String,Session> sessionPool = new ConcurrentHashMap<String,Session>();
   
       /**
        * 链接成功调用的方法
        */
       @OnOpen
       public void onOpen(Session session, @PathParam(value="userId")String userId) {
           try {
               this.session = session;
               this.userId = userId;
               webSockets.add(this);
               sessionPool.put(userId, session);
               log.info("【websocket消息】有新的连接，总数为:"+webSockets.size());
           } catch (Exception e) {
           }
       }
   
       /**
        * 链接关闭调用的方法
        */
       @OnClose
       public void onClose() {
           try {
               webSockets.remove(this);
               sessionPool.remove(this.userId);
               log.info("【websocket消息】连接断开，总数为:"+webSockets.size());
           } catch (Exception e) {
           }
       }
       /**
        * 收到客户端消息后调用的方法
        *
        * @param message
        */
       @OnMessage
       public void onMessage(String message) {
           log.info("【websocket消息】收到客户端消息:"+message);
       }
   
       /** 发送错误时的处理
        * @param session
        * @param error
        */
       @OnError
       public void onError(Session session, Throwable error) {
   
           log.log(WARNING,"用户错误,原因:"+error.getMessage());
           error.printStackTrace();
       }
   
   	/**
   	*	下面为服务端向客户端发送消息
   	*/
       // 此为广播消息
       public void sendAllMessage(String message) {
           log.info("【websocket消息】广播消息:"+message);
           for(WebSocket webSocket : webSockets) {
               try {
                   if(webSocket.session.isOpen()) {
                       webSocket.session.getAsyncRemote().sendText(message);
                   }
               } catch (Exception e) {
                   e.printStackTrace();
               }
           }
       }
   
       // 此为单点消息
       public void sendOneMessage(String userId, String message) {
           Session session = sessionPool.get(userId);
           if (session != null&&session.isOpen()) {
               try {
                   log.info("【websocket消息】 单点消息:"+message);
                   session.getAsyncRemote().sendText(message);
               } catch (Exception e) {
                   e.printStackTrace();
               }
           }
       }
   
       // 此为单点消息(多人)
       public void sendMoreMessage(String[] userIds, String message) {
           for(String userId:userIds) {
               Session session = sessionPool.get(userId);
               if (session != null&&session.isOpen()) {
                   try {
                       log.info("【websocket消息】 单点消息:"+message);
                       session.getAsyncRemote().sendText(message);
                   } catch (Exception e) {
                       e.printStackTrace();
                   }
               }
           }
   
       }
   }
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23
   24
   25
   26
   27
   28
   29
   30
   31
   32
   33
   34
   35
   36
   37
   38
   39
   40
   41
   42
   43
   44
   45
   46
   47
   48
   49
   50
   51
   52
   53
   54
   55
   56
   57
   58
   59
   60
   61
   62
   63
   64
   65
   66
   67
   68
   69
   70
   71
   72
   73
   74
   75
   76
   77
   78
   79
   80
   81
   82
   83
   84
   85
   86
   87
   88
   89
   90
   91
   92
   93
   94
   95
   96
   97
   98
   99
   100
   101
   102
   103
   104
   105
   106
   107
   108
   109
   110
   111
   112
   113
   114
   115
   116
   117
   118
   119
   120
   121
   122
   123

   上面的操作类中有几个主要的点需要注意

   • @ServerEndpoint注解：注解的value属性为调用时路径。类似于@RequestMapping("")设置的路径。添加该注解，才会被注册。
   • @OnOpen：链接成功调用的方法。
   • @OnMessage：客户端可以主动给服务端发送消息，此方法接受数据并处理。
   • @OnError：发送错误时的处理。

4. 服务端主动向客户端发送消息

   测试用例

   @Resource
   private WebSocket webSocket;
   
   @GetMapping("sendMessage")
   public AjaxResult queryById(@Validated @NotNull Long id){
       //创建业务消息信息
       JSONObject obj = new JSONObject();
       obj.put("msgId", "00000001");//消息id
   	obj.put("msgTxt", "服务端->客户端发送消息");//消息内容
       //全体发送
       webSocket.sendAllMessage(obj.toJSONString());
       //单个用户发送 (userId为用户id)
       //webSocket.sendOneMessage(userId, obj.toJSONString());
       //多个用户发送 (userIds为多个用户id，逗号‘,’分隔)
       //webSocket.sendMoreMessage(userIds, obj.toJSONString());
       return AjaxResult.success("执行成功");
   }
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17

3.2 客户端

<script>
export default {
  name: "index",
  data() {
    return {
      websock:null
    };
  },
  mounted() {
    //初始化websocket
    this.initWebSocket()
  },
  destroyed: function () {
    //关闭连接
    this.websocketclose();
  },
  methods: {
    initWebSocket: function () { // 建立连接
      // WebSocket与普通的请求所用协议有所不同，ws等同于http，wss等同于https
      var userId = "user-001";
      var url = "ws://localhost:9001/websocket/" + userId;
      this.websock = new WebSocket(url);
      this.websock.onopen = this.websocketonopen;
      this.websock.onerror = this.websocketonerror;
      this.websock.onmessage = this.websocketonmessage;
      this.websock.onclose = this.websocketclose;
    },
    // 连接成功后调用
    websocketonopen: function () {
      console.log("WebSocket连接成功");
    },
    // 发生错误时调用
    websocketonerror: function (e) {
      console.log("WebSocket连接发生错误");
    },
	// 接收后端消息
    websocketonmessage: function (e) {
      console.log("eee",e)
      var data = eval("(" + e.data + ")"); 
    },
    // 关闭连接时调用
    websocketclose: function (e) {
      console.log("connection closed (" + e.code + ")");
    },
    //向后台发送消息
    sendMessage(){
      let params = {
        id:"00000",
        msg:"前端消息测试"
      }
      let a = JSON.stringify(params);
      this.websock.send(a)
    },
  }
</script>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

websockke中内置了连接、错误、接收消息、接收消息、关闭的回调，下面自己定义的websocketonopen、websocketonmessage等方法的名字可以随便起，但需要在初始化时赋值给websocket对应的属性。

下面测试下完整流程

• 创建连接

  ws同http，wss同https，后面路径为服务段@ServerEndpoint注解的值，以此选择连接不同连接端。
  

• 前台客户端主动发送消息给服务端

  调用websock.send()方法，但消息的类型需要注意，socket本质是传输字节流，所以不能把任意类型的数据直接传入send方法，限制类型如下：
  

  等接收到数据以后通过IO包装类都可以把数据还原。

  服务端成功接收到消息：
  

• 服务端主动向客户端发送消息

  客户端成功接收
  
  

  可以看到是一个请求长连接：

  • 绿色向上的箭头是客户端发送给服务端
  • 红色向下的箭头是服务端发送给客户端

上述测试的流程如下：