Java通過WebSocket實(shí)現(xiàn)異步導(dǎo)出解決思路
前言
本篇文章記錄大批量數(shù)據(jù)導(dǎo)出時間過長,導(dǎo)致接口請求超時問題。
解決思路與流程
- 前端點(diǎn)擊導(dǎo)出按鈕時開啟websocket連接
- 邏輯處理異步執(zhí)行
- 文件處理好后,得到文件的絕對路徑
- 后臺socket通知前端絕對路徑的地址
- 前端下載文件到瀏覽器
本地環(huán)境
Vue版本:2.6.12
Java版本:1.8
SpringBoot版本:2.2.13
項(xiàng)目使用若依前后分離版
一、WebSocket配置
1.pom文件配置
<!-- SpringBoot Websocket --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-websocket</artifactId> <version>2.2.13.RELEASE</version> </dependency>
2.信號量相關(guān)處理
import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.Semaphore; /** * 信號量相關(guān)處理 * * @author */ public class SemaphoreUtils { /** * SemaphoreUtils 日志控制器 */ private static final Logger LOGGER = LoggerFactory.getLogger(SemaphoreUtils.class); /** * 獲取信號量 * * @param semaphore * @return */ public static boolean tryAcquire(Semaphore semaphore) { boolean flag = false; try { flag = semaphore.tryAcquire(); } catch (Exception e) { LOGGER.error("獲取信號量異常", e); } return flag; } /** * 釋放信號量 * * @param semaphore */ public static void release(Semaphore semaphore) { try { semaphore.release(); } catch (Exception e) { LOGGER.error("釋放信號量異常", e); } } }
3.websocket 配置
import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.web.socket.server.standard.ServerEndpointExporter; /** * websocket 配置 * * @author */ @Configuration public class WebSocketConfig { @Bean public ServerEndpointExporter serverEndpointExporter() { return new ServerEndpointExporter(); } }
4.websocket 消息處理
import com.hnxr.scada.utils.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; import javax.websocket.*; import javax.websocket.server.ServerEndpoint; import java.util.concurrent.Semaphore; /** * websocket 消息處理 * * @author */ @Component @ServerEndpoint("/websocket/message") public class WebSocketServer { /** * WebSocketServer 日志控制器 */ private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketServer.class); /** * 默認(rèn)最多允許同時在線人數(shù)100 */ public static int socketMaxOnlineCount = 100; public static Semaphore socketSemaphore = new Semaphore(socketMaxOnlineCount); /** * 連接建立成功調(diào)用的方法 */ @OnOpen public void onOpen(Session session) throws Exception { boolean semaphoreFlag = false; // 嘗試獲取信號量 semaphoreFlag = SemaphoreUtils.tryAcquire(socketSemaphore); if (!semaphoreFlag) { // 未獲取到信號量 LOGGER.error("\n 當(dāng)前在線人數(shù)超過限制數(shù)- {}", socketMaxOnlineCount); WebSocketUsers.sendMessageToUserByText(session, "當(dāng)前在線人數(shù)超過限制數(shù):" + socketMaxOnlineCount); session.close(); } else { //截取前端傳入的id String sessionId = StringUtils.substringAfterLast(session.getQueryString(), "socketId="); System.out.println("---------------------解析socketId" + sessionId); // 添加用戶 WebSocketUsers.put(sessionId, session); System.out.println("-----------------當(dāng)前用戶" + WebSocketUsers.getUsers()); LOGGER.info("\n 建立連接 - {}", session); LOGGER.info("\n 當(dāng)前人數(shù) - {}", WebSocketUsers.getUsers().size()); WebSocketUsers.sendMessageToUserByText(session, "連接成功"); } } /** * 連接關(guān)閉時處理 */ @OnClose public void onClose(Session session) { LOGGER.info("\n 關(guān)閉連接 - {}", session); String sessionId = StringUtils.substringAfterLast(session.getQueryString(), "socketId="); // 移除用戶 WebSocketUsers.remove(sessionId); // 獲取到信號量則需釋放 SemaphoreUtils.release(socketSemaphore); } /** * 拋出異常時處理 */ @OnError public void onError(Session session, Throwable exception) throws Exception { if (session.isOpen()) { // 關(guān)閉連接 session.close(); } String sessionId = StringUtils.substringAfterLast(session.getQueryString(), "socketId="); LOGGER.info("\n 連接異常 - {}", sessionId); LOGGER.info("\n 異常信息 - {}", exception); // 移出用戶 WebSocketUsers.remove(sessionId); // 獲取到信號量則需釋放 SemaphoreUtils.release(socketSemaphore); } /** * 服務(wù)器接收到客戶端消息時調(diào)用的方法 */ @OnMessage public void onMessage(String message, Session session) { WebSocketUsers.sendMessageToUserByText(session, message); } }
白名單放開socket配置接口
.antMatchers("/websocket/**").permitAll()
5.websocket 用戶集
import org.slf4j.Logger; import org.slf4j.LoggerFactory; import javax.websocket.Session; import java.io.IOException; import java.util.Collection; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; /** * websocket 客戶端用戶集 * * @author */ public class WebSocketUsers { /** * WebSocketUsers 日志控制器 */ private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketUsers.class); /** * 用戶集 */ private static Map<String, Session> USERS = new ConcurrentHashMap<String, Session>(); /** * 存儲用戶 * * @param key 唯一鍵 * @param session 用戶信息 */ public static void put(String key, Session session) { USERS.put(key, session); } /** * 移除用戶 * * @param session 用戶信息 * @return 移除結(jié)果 */ public static boolean remove(Session session) { String key = null; boolean flag = USERS.containsValue(session); if (flag) { Set<Map.Entry<String, Session>> entries = USERS.entrySet(); for (Map.Entry<String, Session> entry : entries) { Session value = entry.getValue(); if (value.equals(session)) { key = entry.getKey(); break; } } } else { return true; } return remove(key); } /** * 移出用戶 * * @param key 鍵 */ public static boolean remove(String key) { LOGGER.info("\n 正在移出用戶 - {}", key); Session remove = USERS.remove(key); if (remove != null) { boolean containsValue = USERS.containsValue(remove); LOGGER.info("\n 移出結(jié)果 - {}", containsValue ? "失敗" : "成功"); return containsValue; } else { return true; } } /** * 獲取在線用戶列表 * * @return 返回用戶集合 */ public static Map<String, Session> getUsers() { return USERS; } /** * 群發(fā)消息文本消息 * * @param message 消息內(nèi)容 */ public static void sendMessageToUsersByText(String message) { Collection<Session> values = USERS.values(); for (Session value : values) { sendMessageToUserByText(value, message); } } /** * 發(fā)送文本消息 * * @param session 緩存 * @param message 消息內(nèi)容 */ public static void sendMessageToUserByText(Session session, String message) { if (session != null) { try { session.getBasicRemote().sendText(message); } catch (IOException e) { LOGGER.error("\n[發(fā)送消息異常]", e); } } else { LOGGER.info("\n[你已離線]"); } } }
二、建立WebSocket連接
1.導(dǎo)出按鈕
<el-button type="warning" plain icon="el-icon-download" size="mini" :loading="exportLoading" @click="handleExport">導(dǎo)出 </el-button>
2.導(dǎo)出方法
/** 導(dǎo)出按鈕操作 */ handleExport() { this.$confirm('是否確認(rèn)導(dǎo)出所有?', '警告', { confirmButtonText: '確定', cancelButtonText: '取消', type: 'warning' }) .then(() => { const ip = window.location.host; //TODO 服務(wù)器端口號 nginx代理為4430 const hostIp = ip.split(":")[0]; console.log('ip:', hostIp) const url = "wss://" + hostIp + ":4430/websocket/message"; //TODO 本地端口號為9996 項(xiàng)目啟動端口 // let url = "wss://localhost:9996/websocket/message"; console.log('url', url) const socketId = this.getRandomId(); //將socketId當(dāng)做參數(shù)傳入后端 this.queryParams.sessionId = socketId const socket = new WebSocket(url + '?socketId=' + socketId); socket.onopen = function (event) { console.log("已開啟websocket連接!" + "sessionId:" + socketId) }; socket.onmessage = function (event) { const val = event.data console.log("收到消息", val) if (val.endsWith("asyExport")) { const path = val.slice(0, val.indexOf("asyExport")); console.log("文件名:" + path) const baseURL = process.env.VUE_APP_BASE_API window.location.href = baseURL + "/common/download?fileName=" + encodeURI(path) + "&delete=" + true; //延時關(guān)閉連接 確保WebSocket處于打開狀態(tài) setTimeout(function () { if (socket.readyState === WebSocket.OPEN) { socket.close(); } }, 5000); } }; // 處理連接關(guān)閉 socket.onclose = function (event) { console.log("已經(jīng)關(guān)閉連接!"); }; // 處理連接錯誤 socket.onerror = function (error) { console.error("WebSocket 出現(xiàn)錯誤:", error); if (socket.readyState === WebSocket.OPEN) { socket.close(); } }; exportChineseParsing(this.queryParams) }) }, // 生成唯一標(biāo)識符 getRandomId() { return Math.random().toString(36).substring(2) + Date.now().toString(36); }
本地項(xiàng)目是http請求,前綴設(shè)置為ws
后端application.yml 的ssl配置將server下的enabled改為false
前端代理地址保持http
https請求,前綴設(shè)置為wss
ssl配置文件將server下的enabled改為true
前端代理地址改為https
3.導(dǎo)出接口地址
//導(dǎo)出中文解析 export function exportChineseParsing(query) { return request({ url: '/basic/chineseParsing/export', method: 'get', params: query, responseType: 'blob' }) }
瀏覽器控制臺查看連接時是否正常
前后端socket配置完成后在瀏覽器控制臺輸入
new WebSocket("ws://localhost:9996/websocket/message")
測試是否正常 如下圖 拋出異常需要檢查上方配置是否正確
4.流程說明
首先獲取當(dāng)前ip,本地使用localhost即可,服務(wù)器部署需要配置nginx代理
ip后跟端口號,端口后是后端websocket接口地址
socket.onopen回調(diào)方法驗(yàn)證是否成功建立連接
socket.onmessage是接收后端接口異步導(dǎo)出后的消息
后端文件處理好后,約定發(fā)送的消息以asyExport字符為后綴,方便前端識別和截取文件絕對路徑
調(diào)用下載接口,關(guān)閉socket連接
如不是若依前后分離版,替換為自己的下載文件的接口就行
三、異步導(dǎo)出接口
1.導(dǎo)出接口示例
@GetMapping("/export") public void export(BasicChineseParsing basicChineseParsing) throws Exception { //存入文件名稱 AtomicReference<String> path = new AtomicReference<>(""); //邏輯異步處理 CompletableFuture<Void> runAsync = CompletableFuture.runAsync(() -> { try { //模擬邏輯處理時長 System.out.println("-----------------------睡眠5秒"); Thread.sleep(5000); ....... //創(chuàng)建臨時文件 絕對路徑 File tempFile = PoiUtils.createTempNameFile(rows, title, "_中文解析表"); String result = tempFile.getCanonicalPath().substring(tempFile.getCanonicalPath().indexOf("vehicle")); System.out.println("-----------------------處理后文件名:" + result); path.set(result); } catch (InterruptedException | IOException e) { e.printStackTrace(); } }); //異步處理完成后 runAsync.thenRun(() -> { // 發(fā)送消息 拿到前端傳入的socketId String sessionId = basicChineseParsing.getSessionId(); System.out.println("-----------------------sessionId:" + sessionId); Session session = WebSocketUsers.getUsers().get(sessionId); webSocketServer.onMessage(path.toString() + "asyExport", session); System.out.println("-----------------------導(dǎo)出成功"); //前端關(guān)閉socket連接 }); }
2.debug調(diào)試流程截圖
下載方法大家的和我的會不一樣
但是File tempFile只要獲取到下載好后的絕對路徑就可以
我這里path存入的是下載的文件名稱
因?yàn)榍岸藭俅握{(diào)用下載接口,有文件名稱,就可以將文件下載至瀏覽器了
在webSocketServer.onMessage方法中發(fā)送的消息就是文件名稱+asyExport一個特殊約定的后綴
前端在收到消息后會截取asyExport前的文件名稱
完成流程日志打印如下
因使用若依前后分離版,大家調(diào)整后的下載方法可能和我的不一樣
比如socket需要發(fā)送的信息是文件的絕對路徑,前端調(diào)用下載方法傳入絕對路徑即可
四、服務(wù)器部署nginx配置
這個配置看大家是否需要部署至服務(wù)器
server { listen 4430 ssl; server_name localhost; ..... location /websocket/message { proxy_pass https://服務(wù)器ip:4430; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; } }
配置需要放在當(dāng)前代理的端口配置下
如果是ssl認(rèn)證的proxy_pass后地址需要用https,沒有認(rèn)證就用http
ip后的端口是你的程序服務(wù)使用端口,不是代理映射的端口
這里配置的地址是和前端創(chuàng)建連接時const url = "wss://" + hostIp + ":4430/websocket/message";
ip和端口是一致的
總結(jié)
第一步WebSocket配置大家是可以直接復(fù)制使用
第二步我的處理是只有導(dǎo)出時才啟用socket連接,下載文件后關(guān)閉,也可以設(shè)置為全局性的,方便處理其他通知
socket通知前端的下載方式也有很多種,需要考慮的是會有其他用戶或其他導(dǎo)出接口,其他socket通知,所以通知下載的socket消息是需要具有唯一性的,上方建立socket連接時就加了保障唯一性的socketId,并當(dāng)為參數(shù)傳入后端。
到此這篇關(guān)于通過WebSocket實(shí)現(xiàn)異步導(dǎo)出的文章就介紹到這了,更多相關(guān)WebSocket異步導(dǎo)出內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
解決layui中onchange失效以及form動態(tài)渲染失效的問題
今天小編就為大家分享一篇解決layui中onchange失效以及form動態(tài)渲染失效的問題,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧2019-09-09原生js實(shí)現(xiàn)類似fullpage的單頁/全屏滾動
這篇文章主要介紹了利用原生js實(shí)現(xiàn)類似fullpage的全屏滾動的實(shí)現(xiàn)方法,文中給出了完整的實(shí)例代碼,相信對大家的理解和學(xué)習(xí)具有一定的參考價值,需要的朋友們可以參考借鑒,下面來一起看看吧。2017-01-01javascript中JSON對象與JSON字符串相互轉(zhuǎn)換實(shí)例
這篇文章主要介紹了javascript中JSON對象與JSON字符串相互轉(zhuǎn)換,實(shí)例分析了json對象與字符串常用的幾種轉(zhuǎn)換技巧,具有一定參考借鑒價值,需要的朋友可以參考下2015-07-07