天天看點
傳回

WebSocket的實作與應用

WebSocket的實作與應用

前言

說到websocket，就不得不提http協定的連接配接特點特點與互動模型。

首先，http協定的特點是無狀态連接配接。即http的前一次連接配接與後一次連接配接是互相獨立的。

其次，http的互動模型是請求/應答模型。即互動是通過C/B端向S端發送一個請求，S端根據請求，傳回一個響應。

那麼這裡就有一個問題了--S端無法主動向C/B端發送消息。而互動是雙方的事情，怎麼能限定一方發資料，另一方接資料呢。

傳統解決方案：

傳統的解決方案就倆字：輪詢。

長短連接配接輪詢就不詳細說了，就說說輪詢。大概的場景是這樣的：

用戶端（Request）：有消息不？

服務端（Response）：No

服務端（Response）：有了。你媽叫你回家吃飯。

==================================> loop

看着都累，資源消耗那就更不必說了。尤其有些對實時性要求高的資料，那可能就是1s請求一次。目測伺服器已經淚奔。

websocket解決方案：

那麼websocket的解決方案，總結一下，就是：建立固定連接配接

說白了，就是C/B端與S端就一個websocket服務建立一個固定的連接配接，不斷開。

大概的場景是這樣的：

服務端：我建立了一個chat的websocket，歡迎大家連接配接。

用戶端：我要和你的chat的websocket連接配接，我的sid（唯一辨別）是No.1

服務端：好的，我已經記住你了。如果有發往chat下No.1的消息，我會告訴你的。

用戶端：嗯。謝謝了哈。

==================================> 過了一段時間

（有一個請求調用了chat的websocket，并且指名是給No.1的消息）

服務端（發送消息給No.1）：No.1，有你的消息。你媽媽叫你回家做作業。

用戶端（No.1）：好的。我收到了。謝謝。

由于這次隻是簡單說一下websocket，是以就不深入解讀網絡相關知識了。

應用場景

既然http無法滿足使用者的所有需求，那麼為之誕生的websocket必然有其諸多應用場景。如：

  1. 實時顯示網站線上人數
  2. 賬戶餘額等資料的實時更新
  3. 多玩家網絡遊戲
  4. 多媒體聊天，如聊天室
  5. 。。。

其實總結一下，websocket的應用場景就倆字：實時

無論是多玩家網絡遊戲，網站線上人數等都是由于實時性的需求，才用上了websocket（後面用縮寫ws)。

談幾個在我項目中用到的情景：

  1. 線上教育項目中的課件系統，通過ws實作學生端課件與教師端課件的實時互動
  2. 物聯網項目中的報警系統，通過ws實作報警資訊的實時推送
  3. 大資料項目中的資料展示，通過ws實作資料的實時更新
  4. 物聯網項目中的硬體互動系統，通過ws實作硬體異步響應的展示

當你的項目中存在需要S端向C/B端發送資料的情形，那就可以考慮上一個websocket了。

實作

服務端開發：

引入依賴： 

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

添加配置：

忍不住想要吐槽，為什麼不可以如eureka等元件那樣，直接在啟動類寫一個注解就Ok了呢。看來還得以後自己動手，豐衣足食啊。

package com.renewable.center.warning.configuration;
	
	import org.springframework.context.annotation.Bean;
	import org.springframework.context.annotation.Configuration;
	import org.springframework.web.socket.server.standard.ServerEndpointExporter;
	
	/**
	 * Websocket的配置
	 * 說白了就是引入Websocekt至spring容器
	 */
	@Configuration
	public class WebSocketConfig {  
		
	    @Bean
	    public ServerEndpointExporter serverEndpointExporter() {  
	        return new ServerEndpointExporter();
	    }  
	  
	}

代碼實作：

WebSocketServer的實作： 

package com.renewable.center.warning.controller.websocket;
	
	import lombok.extern.slf4j.Slf4j;
	import org.apache.commons.lang3.StringUtils;
	import org.springframework.stereotype.Component;
	
	import javax.websocket.*;
	import javax.websocket.server.PathParam;
	import javax.websocket.server.ServerEndpoint;
	import java.io.IOException;
	import java.util.concurrent.CopyOnWriteArraySet;
	
	/**
	 * @Description：
	 * @Author: jarry
	 */
	@Component
	@Slf4j
	@ServerEndpoint("/websocket/warning/{sid}")
	public class WarningWebSocketServer {
	
		// JUC包的線程安全Set，用來存放每個用戶端對應的WarningWebSocketServer對象。
		// 用ConcurrentHashMap也是可以的。說白了就是類似線程池中的BlockingQueue那樣作為一個容器
		private static CopyOnWriteArraySet<WarningWebSocketServer> warningWebSocketSet = new CopyOnWriteArraySet<WarningWebSocketServer>();
	
		// 與某個用戶端的連接配接會話，需要通過它來給用戶端發送資料
		private Session session;
	
		// 接收sid
		private String sid="";
	
		/**
		 * 建立websocket連接配接
		 * 看起來很像JSONP的回調，因為前端那裡是Socket.onOpne()
		 * @param session
		 * @param sid
		 */
		@OnOpen
		public void onOpen(Session session, @PathParam("sid") String sid){
			this.session = session;
			this.sid = sid;
			warningWebSocketSet.add(this);
	
			sendMessage("websocket connection has created.");
		}
	
		/**
		 * 關閉websocket連接配接
		 */
		@OnClose
		public void onClose(){
			warningWebSocketSet.remove(this);
			log.info("there is an wsConnect has close .");
		}
	
		/**
		 * websocket連接配接出現問題時的處理
		 */
		@OnError
		public void onError(Session session, Throwable error){
			log.error("there is an error has happen ! error:{}",error);
		}
	
		/**
		 * websocket的server端用于接收消息的（目測是用于接收前端通過Socket.onMessage發送的消息）
		 * @param message
		 */
		@OnMessage
		public void onMessage(String message){
			log.info("webSocketServer has received a message:{} from {}", message, this.sid);
	
			// 調用消息處理方法（此時針對的WarningWebSocektServer對象，隻是一個執行個體。這裡進行消息的單發）
			// 目前這裡還沒有處理邏輯。故為了便于前端調試，這裡直接傳回消息
			this.sendMessage(message);
		}
	
		/**
		 * 伺服器主動推送消息的方法
		 */
		public void sendMessage(String message){
			try {
				this.session.getBasicRemote().sendText(message);
			} catch (IOException e) {
				log.warn("there is an IOException:{}!",e.toString());
			}
		}
	
		public static void sendInfo(String sid, String message){
			for (WarningWebSocketServer warningWebSocketServerItem : warningWebSocketSet) {
				if (StringUtils.isBlank(sid)){
					// 如果sid為空，即群發消息
					warningWebSocketServerItem.sendMessage(message);
					log.info("Mass messaging. the message({}) has sended to sid:{}.", message,warningWebSocketServerItem.sid);
				}
				if (StringUtils.isNotBlank(sid)){
					if (warningWebSocketServerItem.sid.equals(sid)){
						warningWebSocketServerItem.sendMessage(message);
						log.info("single messaging. message({}) has sended to sid:{}.", message, warningWebSocketServerItem.sid);
					}
				}
			}
		}
	
	}

WesocketController

為了便于調試與展示效果，寫一個控制層，用于推送消息

package com.renewable.center.warning.controller.websocket;
	
	import com.renewable.terminal.terminal.common.ServerResponse;
	import org.springframework.stereotype.Controller;
	import org.springframework.web.bind.annotation.GetMapping;
	import org.springframework.web.bind.annotation.RequestMapping;
	import org.springframework.web.bind.annotation.RequestParam;
	import org.springframework.web.bind.annotation.ResponseBody;
	
	import java.io.IOException;
	
	/**
	 * @Description： 用于測試WebsocketServer
	 * @Author: jarry
	 */
	@Controller
	@RequestMapping("/websocket/test/")
	public class WarningWebsocketController {
	
		@GetMapping("link.do")
		@ResponseBody
		public ServerResponse link(@RequestParam(name = "sid") int sid){
			return ServerResponse.createBySuccessMessage("link : "+sid);
		}
	
		/**
		 * 調用WarningWebsocketServer的消息推送方法，進而進行消息推送
		 * @param sid 連接配接WarningWebsocketServer的前端的唯一辨別。如果sid為空，即表示向所有連接配接WarningWebsocketServer的前端發送相關消息
		 * @param message 需要發送的内容主體
		 * @return
		 */
		@ResponseBody
		@RequestMapping("push.do")
		public ServerResponse pushToWeb(@RequestParam(name = "sid", defaultValue = "") String sid, @RequestParam(name = "message")  String message) {
			WarningWebSocketServer.sendInfo(sid, message);
			return ServerResponse.createBySuccessMessage(message+"@"+sid+" has send to target.");
		}
	}

WesocketTestIndex

這裡建立了一個B端頁面，用于與S端進行互動，示範。

<!DOCTYPE html>
	<html lang="en">
	<head>
	    <meta charset="UTF-8">
	    <title>WebsocketTestIndex</title>
	</head>
	<body>
	
	<h1>Websocket Test</h1>
	<script>
	    var socket;
	    if(typeof(WebSocket) == "undefined") {
	        console.log("Your browser not support WebSocket !");
	    }else{
	        console.log("Your browser support WebSocket");
	        // 執行個體化WebSocket對象
	        // 指定要連接配接的伺服器位址與端口
	        // 建立連接配接
	        socket = new WebSocket("ws://localhost:10706/websocket/warning/2");
	        // 打開事件
	        socket.onopen = function() {
	            console.log("You has connect to WebSocketServer");
	        };
	        // 獲得消息事件
	        socket.onmessage = function(msg) {
	            // 列印接收到的消息
	            console.log(msg.data);
	        };
	        // 關閉事件
	        socket.onclose = function() {
	            console.log("Socket has closed");
	        };
	        // 發生了錯誤事件
	        socket.onerror = function() {
	            alert("Socket happen an error !");
	        }
	    }
	</script>
	</body>
	</html>

效果展示

再次強調，圖檔很大很清晰。如果看不清楚，請單獨打開圖檔。

B端網頁初始化：

WebSocket的實作與應用

調用S端WarningWebsocketController下pushToWeb()接口，對sid=2的B端發送消息：

WebSocket的實作與應用

B端網頁接收到專門發給sid=2的消息後的效果：

WebSocket的實作與應用

調用S端WarningWebsocketController下pushToWeb()接口，所有連接配接該websocket的B端群發消息：

B端網頁接收到群發消息後的效果：

S端接收到消息後的日志列印：

S端在B端關閉連接配接後的日志列印：

總結

至此，websocket的應用就算入門了。至于實際的使用，其實就是服務端自己調用WebSocket的sendInfo接口。當然也可以自己擴充更為細緻的邏輯，方法等。

另外，需要注意的是，别忘了及時關閉webocket的連接配接。尤其在負載較大的情況下，更需要注意即使關閉不必要的連接配接。

架構的技術選型，需要的不是最好的，而是最适合的。

擴充：

如果想要了解更多概念上的細節，可以看看這篇文章：

websocket的了解&應用&場景

web websocket 輪詢 長連接配接 實時 java
上一篇: 基于Apache Hudi + Flink的億級資料入湖實踐
下一篇: vue報錯 Do not use built-in or reserved HTML elements as component id:header

繼續閱讀