Java通過WebSocket實現(xiàn)異步導(dǎo)出解決思路
更新時間：2024年01月12日 11:38:01 作者：夜の雨
這篇文章主要介紹了通過WebSocket實現(xiàn)異步導(dǎo)出,本篇文章記錄大批量數(shù)據(jù)導(dǎo)出時間過長,導(dǎo)致接口請求超時問題,本文給大家介紹的非常詳細(xì),對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友可以參考下
前言
本篇文章記錄大批量數(shù)據(jù)導(dǎo)出時間過長,導(dǎo)致接口請求超時問題。
解決思路與流程
- 前端點擊導(dǎo)出按鈕時開啟websocket連接
- 邏輯處理異步執(zhí)行
- 文件處理好后，得到文件的絕對路徑
- 后臺socket通知前端絕對路徑的地址
- 前端下載文件到瀏覽器
本地環(huán)境
Vue版本：2.6.12
Java版本：1.8
SpringBoot版本：2.2.13
項目使用若依前后分離版
一、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 項目啟動端口
// 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);
}本地項目是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)方法驗證是否成功建立連接
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存入的是下載的文件名稱
因為前端會再次調(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實現(xiàn)異步導(dǎo)出的文章就介紹到這了,更多相關(guān)WebSocket異步導(dǎo)出內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
這篇文章主要介紹了Javascript模塊導(dǎo)入導(dǎo)出詳解的相關(guān)資料,需要的朋友可以參考下2022-12-12
解決layui中onchange失效以及form動態(tài)渲染失效的問題
今天小編就為大家分享一篇解決layui中onchange失效以及form動態(tài)渲染失效的問題，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-09-09
這篇文章主要為大家詳細(xì)介紹了JavaScript定時器制作一個彈窗小廣告，具有一定的參考價值，感興趣的小伙伴們可以參考一下2017-02-02
這篇文章主要為大家詳細(xì)介紹了微信小程序圖片橫向左右滑動的案例，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價值，感興趣的小伙伴們可以參考一下2017-05-05
本文為大家詳細(xì)介紹下js中自動清除ie緩存的幾種方法，大家可以根據(jù)自己的需求自由選擇適合自己的，希望對大家學(xué)習(xí)js有所幫助2013-07-07
這篇文章主要介紹了JavaScript 聲明提升的相關(guān)資料，幫助大家更好的理解和學(xué)習(xí)JavaScript，感興趣的朋友可以了解下2020-09-09
這篇文章主要為大家詳細(xì)介紹了js canvas實現(xiàn)圓角圖片效果，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價值，感興趣的小伙伴們可以參考一下2021-09-09
原生js實現(xiàn)類似fullpage的單頁/全屏滾動
這篇文章主要介紹了利用原生js實現(xiàn)類似fullpage的全屏滾動的實現(xiàn)方法，文中給出了完整的實例代碼，相信對大家的理解和學(xué)習(xí)具有一定的參考價值，需要的朋友們可以參考借鑒，下面來一起看看吧。2017-01-01
Fetch?API提供了一個JavaScript接口，用于訪問和操縱HTTP管道的部分。它還提供了一個全局?fetch()方法，該方法提供了一種簡單，合理的方式來跨網(wǎng)絡(luò)異步獲取資源。本文將詳解JS如何使用Fetch，感興趣的可以學(xué)習(xí)一下2022-05-05
javascript中JSON對象與JSON字符串相互轉(zhuǎn)換實例
這篇文章主要介紹了javascript中JSON對象與JSON字符串相互轉(zhuǎn)換,實例分析了json對象與字符串常用的幾種轉(zhuǎn)換技巧,具有一定參考借鑒價值,需要的朋友可以參考下2015-07-07