thumbnail

1 概述

1.1 WebSocket简介

WebSocket是一种协议,用于在Web应用程序和服务器之间建立实时、双向的通信连接。它通过一个单一的TCP连接提供了持久化连接,这使得Web应用程序可以更加实时地传递数据。WebSocket协议最初由W3C开发,并于2011年成为标准。

WebSocket使得客户端和服务器之间的数据交换变得更加简单,允许服务端主动向客户端推送数据。

在WebSocket API中,浏览器和服务器只需要完成一次握手,两者之间就直接可以创建持久性的连接,并进行双向数据传输。

1.2 WebSocket优劣势

优势:

  • 实时性: 由于WebSocket的持久化连接,它可以实现实时的数据传输,避免了Web应用程序需要不断地发送请求以获取最新数据的情况。
  • 双向通信(全双工): WebSocket协议支持双向通信,这意味着服务器可以主动向客户端发送数据,而不需要客户端发送请求。
  • 减少网络负载: 由于WebSocket的持久化连接,它可以减少HTTP请求的数量,从而减少了网络负载。

劣势:

  • 需要浏览器和服务器都支持: WebSocket是一种相对新的技术,需要浏览器和服务器都支持。一些旧的浏览器和服务器可能不支持WebSocket。
  • 需要额外的开销: WebSocket需要在服务器上维护长时间的连接,这需要额外的开销,包括内存和CPU。
  • 安全问题: 由于WebSocket允许服务器主动向客户端发送数据,可能会存在安全问题。服务器必须保证只向合法的客户端发送数据

1.3 与HTTP协议的区别

HTTP协议有一个缺陷:通信只能由客户端发起。因为一般的请求都是HTTP请求(单向通信),HTTP是一个短连接(非持久化),且通信只能由客户端发起,HTTP协议做不到服务器主动向客户端推送消息。而WebSocket的最大特点就是,服务器可以主动向客户端推送信息,客户端也可以主动向服务器发送信息,是真正的双向平等对话,属于服务器推送技术的一种。

2 基本概念

2.1 协议

WebSocket 协议是一种基于TCP的协议,用于在客户端和服务器之间建立持久连接,并且可以在这个连接上实时地交换数据。WebSocket协议有自己的握手协议,用于建立连接,也有自己的数据传输格式。

WebSocket连接在端口80 (ws)或者443 (wss)上创建,与HTTP使用的端口相同,这样,基本上所有的防火墙都不会阻止WebSocket连接。此外WebSocket支持跨域,可以避免Ajax的限制。

总之,WebSocket协议是一种可靠的、高效的、双向的、持久的通信协议,它适用于需要实时通信的Web应用程序,如在线游戏、实时聊天等。

2.2 生命周期

WebSocket 生命周期描述了 WebSocket 连接从创建到关闭的过程。一个 WebSocket 连接包含以下四个主要阶段:

1.连接建立阶段(Connection Establishment): 在这个阶段,客户端和服务器之间的 WebSocket 连接被建立。客户端发送一个 WebSocket 握手请求,服务器响应一个握手响应,然后连接就被建立了。

2.连接开放阶段(Connection Open): 在这个阶段,WebSocket 连接已经建立并开放,客户端和服务器可以在连接上互相发送数据。

3.连接关闭阶段(Connection Closing): 在这个阶段,一个 WebSocket 连接即将被关闭。它可以被客户端或服务器发起,通过发送一个关闭帧来关闭连接。

4.连接关闭完成阶段(Connection Closed): 在这个阶段,WebSocket 连接已经完全关闭。客户端和服务器之间的任何交互都将无效。

sequenceDiagram
Client ->>Server: 1.WebSocket握手请求
Server ->>Client: 2.WebSocket握手响应
Client ->> Server: 3.WebSocket连接开放
Server ->> Client: 3.WebSocket连接开放
Server ->> Client: 4.WebSocket连接关闭

2.3 消息格式

WebSocket 消息格式由两个部分组成:消息头和消息体。

消息头包含以下信息:

FIN: 表示这是一条完整的消息,一般情况下都是1。

RSV1、RSV2、RSV3: 暂时没有使用,一般都是0。

Opcode: 表示消息的类型,包括文本消息、二进制消息等。

Mask: 表示消息是否加密。

Payload length: 表示消息体的长度。

Masking key: 仅在消息需要加密时出现,用于对消息进行解密。

消息体就是实际传输的数据,可以是文本或二进制数据。

2.4 API

WebSocket API 是用于在 Web 应用程序中创建和管理 WebSocket 连接的接口集合。WebSocket API 由浏览器原生支持,无需使用额外的 JavaScript 库或框架,可以直接在 JavaScript 中使用。

let ws = new WebSocket('ws://example.com/ws');

WebSocket.send() 方法: WebSocket.send() 方法用于向服务器发送数据。它接受一个参数,表示要发送的数据。数据可以是字符串、Blob 对象或 ArrayBuffer 对象。例如:

ws.send('Hello, server!');

WebSocket.onopen 事件: WebSocket.onopen 事件在 WebSocket 连接成功建立时触发。例如:

ws.onopen = function() {
  console.log('WebSocket 连接已经建立。');
};

WebSocket.onmessage 事件: WebSocket.onmessage 事件在接收到服务器发送的消息时触发。它的 event 对象包含一个 data 属性,表示接收到的数据。例如:

ws.onmessage = function(event) {
  console.log('收到服务器消息:', event.data);
};

WebSocket.onerror 事件: WebSocket.onerror 事件在 WebSocket 连接出现错误时触发。例如:

ws.onerror = function(event) {
  console.error('WebSocket 连接出现错误:', event);
};

WebSocket.onclose 事件: WebSocket.onclose 事件在 WebSocket 连接被关闭时触发。例如:

ws.onclose = function() {
  console.log('WebSocket 连接已经关闭。');
};

3 应用示例

需求:多个前端用户通过WebSocket连接到后端的某个业务接口上,后端定时或实时将数据发送到前端。

3.1 后端

整体架构

  • 依赖
<dependency> 
	<groupId>org.springframework.boot</groupId> 
	<artifactId>spring-boot-starter-websocket</artifactId> 
</dependency>
  • WebSocket配置

固定写法:

/** * websocket * 的配置信息 */ 
@Configuration 
public class WebSocketConfig { 
	@Bean 
	public ServerEndpointExporter serverEndpointExporter() { 
		return new ServerEndpointExporter(); 
	} 
}
  • WebSocket的服务类

相当于三层架构中的Controller。

1.ServerEndpoint("/api/pushMessage/{userId}") 前端通过此 URI 和后端交互,建立连接

2.OnOpen websocket 建立连接的注解,前端触发上面 URI 时会进入此注解标注的方法

3.OnMessage 收到前端传来的消息后执行的方法

4.OnClose 顾名思义关闭连接,销毁 session

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

新建一个ConcurrentHashMap webSocketMap 用于接收当前userId的WebSocket,方便IM之间对userId进行推送消息

注意这里存储的是sessionId,而不是userId。

@Getter  
@Component  
@Slf4j  
@ServerEndpoint("/api/websocket/service1")  
public class WebSocketServerService1 {  

  /* 最大在线人数 */    
  public static int socketMaxOnlineCount = 100;  

  private static final Semaphore socketSemaphore = new Semaphore(socketMaxOnlineCount);  

  /* concurrent包的线程安全Set,用来存放每个客户端对应的Session对象 */    
  /* key: sessionId value: WebSocketServer */    
  private static final Map<String, WebSocketServerService1> webSocketMap = new ConcurrentHashMap<>();  

  /* 当前的Session */  
  private Session session;  

  /* 连接建立成功调用的方法 */    @OnOpen  
  public void onOpen(Session session) {  
    this.session = session;  
    String sessionId = session.getId();  
    // 尝试获取信号量  
    boolean semaphoreFlag = SemaphoreUtils.tryAcquire(socketSemaphore);  
    if (!semaphoreFlag) {  
      // 未获取到信号量  
      log.error("当前在线人数超过限制数 - {}", socketMaxOnlineCount);  
      this.sendMessage(session, "当前在线人数超过限制数:" + socketMaxOnlineCount);  
      try {  
        session.close();  
      } catch (IOException e) {  
        throw new RuntimeException(e);  
      }  
    } else {  
      webSocketMap.put(sessionId, this);  
      log.info("用户连接: {} - {},当前在线人数为: {}", sessionId, session, webSocketMap.size());  
      this.sendMessage(session, "连接成功");  
    }  
  }  

  /* 连接关闭调用的方法 */    @OnClose  
  public void onClose(Session session) {  
    String sessionId = session.getId();  
    webSocketMap.remove(sessionId);  
    // 获取到信号量则需释放  
    SemaphoreUtils.release(socketSemaphore);  
    log.info("用户退出: {} - {},当前在线人数为: {}", sessionId, session, webSocketMap.size());  
  }  

  /* 收到客户端消息后调用的方法 */    
  @OnMessage  
  public void onMessage(String message, Session session) {  
    String sessionId = session.getId();  
    log.info("收到用户消息: {} - {},报文: {}", sessionId, session, message);  
    // 可以群发消息  
    // 消息保存到数据库、redis  
    if(StringUtils.isNotBlank(message)){  
      try {  
        // 解析发送的报文  
        JSONObject jsonObject = JSON.parseObject(message);  
        // 追加发送人(防止篡改)  
        jsonObject.put("fromSessionId", sessionId);  
        String toSessionId = jsonObject.getString("toSessionId");  
        // 传送给对应toUserId用户的websocket  
        if (StringUtils.isNotBlank(toSessionId) && webSocketMap.containsKey(toSessionId)) {  
          webSocketMap.get(toSessionId).getSession().getBasicRemote().sendText(message);  
        } else {  
          // 否则不在这个服务器上,发送到mysql或者redis  
          log.error("请求的sessionId:{} 不在该服务器上", toSessionId);  
        }  
      }catch (Exception e){  
        e.printStackTrace();  
      }  
    }  
  }  

  /* 连接报错时的处理方法 */    
  @OnError  
  public void onError(Session session, Throwable error) {  
    String sessionId = session.getId();  
    if (session.isOpen()) {  
      // 关闭连接  
      try {  
        session.close();  
      } catch (IOException e) {  
        throw new RuntimeException(e);  
      }  
    }  
    // 移出用户  
    webSocketMap.remove(sessionId);  
    // 获取到信号量则需释放  
    SemaphoreUtils.release(socketSemaphore);  
    log.error("用户错误: {} - {},原因:{}", sessionId, session, error.getMessage());  
  }  

  /* 实现服务器主动推送 */    
  public void sendMessage(Session session, String message) {  
    try {  
      session.getBasicRemote().sendText(message);  
    } catch (IOException e) {  
      log.error("推送报错:{}", e.toString());  
    }  
  }  

  /* 群发:发送自定义消息 */    
  public static void sendInfo(String message) {  
    // 发送到每一个已连接的session中  
    for (WebSocketServerService1 service : webSocketMap.values()) {  
      try {  
        Session session = service.getSession();  
        session.getBasicRemote().sendText(message);  
        log.info("发送消息报文至:{} - {}: {}", session.getId(), session, message);  
      } catch (IOException e) {  
        throw new RuntimeException(e);  
      }  
    }  
  }  
}
  • 业务层和控制器
@Service  
public class HelloServiceImpl implements HelloService {  

  // 每3s推送至前端  
  @Scheduled(fixedRate = 3000)  
  @Override  
  public void printTimeToWeb() {  
    SimpleDateFormat dateFormat = new SimpleDateFormat("HH:mm:ss");  
    String date = dateFormat.format(new Date());  
    WebSocketServerService1.sendInfo(date);  
  }  
}
@RestController  
@RequestMapping("/api/hello")  
public class HelloController {  

  @Resource  
  public HelloService helloService;  

  @PostMapping("/pushToWeb")  
  public void pushToWeb(){  
    helloService.printTimeToWeb();  
  }  
}

3.2 前端

<!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:8080/api/websocket/service1";
      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 closeSocket() {
      socket.close();
    }
    function sendMessage() {
      socket.send('{"toSessionId":"'+$("#toSessionId").val()+'","contentText":"'+$("#contentText").val()+'"}');
    }
  </script>
  <body>
    <p>【socket开启者的ID信息】:<div><input id="userId" name="userId" type="text" value="10"></div>
    <p>【客户端向服务器发送的内容】:<div><input id="toSessionId" name="toSessionId" 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="closeSocket()">关闭socket</a></div>
    <p>【操作】:<div><a onclick="sendMessage()">发送消息</a></div>
  </body>
</html>
测试
连接成功时:

数据推送:

关闭连接:

如果有问题可以通过以下的邮箱联系我帮助你解决哦!