前言
最近在接触的一个项目，将PDF上传到项目里，通过调用OCR云服务把PDF里的表格数据识别出来。在此过程中，前后端需要实时通信，对识别数据进行“进度跟踪”。因此我们采用SocketIO的通讯方式，识别中前端和后端服务建立SocketIO连接，根据事件进行数据的实时更新百分比进度，并且显示在页面中。
简介
SocketIO是在客户端和服务端之间建立的双向通信数据交换技术,它可以在客户端和服务器之间实现低延迟、双向和基于事件的通信。
SocketIO建立在WebSocket协议之上，WebSocket仍然可能纯在浏览器不兼容的情况，但SocketIO无需担心兼容问题，底层会自动选用最佳的通信方式。例如回退到 HTTP 长轮询或自动重新连接。
工作原理
当一个浏览器尝试建立SocketIO时，SocketIO首先使用xhr-polling创建一个长轮询。长轮询一旦建立，它将升级为WebSocket连接。
客户端
SocketIO通过一个XHR（XMLHttprequest）握手，前端发送一个XHR，告诉服务器我要开始长轮询你了~
握手是通过一个HTTP Upgrade request开始的，一个请求头部分示例如下：
###### Request Headers ######
Connection: Upgrade
Host: ip:SocketIO监听的端口
Origin: http://sockert.io.demo.com
Sec-WebSocket-Extensions: permessage-deflate;client_max_window_bits
Sec-WebSocket-Key: bz69CkdvaPHtOTEshtmWvw==
Sec-WebSocket-Version: 13
Upgrade: websocket
1. Upgrade 是HTTP/1.1中规定的用于转换当前连接的应用层协议的头部，表示客户端希望用现有的连接转换到新的应用层协议WebSocket协议
2. Origin 用于防止跨站攻击，浏览器一般会使用这个来标识原始域，对于非浏览器的客户端应用可以根据需要使用。
3. 请求头中的 Sec-WebSocket-Version 是WebSocket版本号，Sec-WebSocket-Key 是用于握手的密钥
前端发送完一个XHR后，后端返回的数据如下：
0{"pingInterval":25000,
"pingTimeout":60000,
"upgrades":["websocket"],
"sid":"9c54f9c1759c4dbab8f3ce20c1fe43a4"
}40
1. sid是本次SocketIO的会话ID，因为一次SocketIO包含了多个请求，而后端又会同时连接多个SocketIO，sid的作用就相当于SESSIONID。
2. 另一个字段upgrades，正常情况下是 [‘websocket’]，表示可以把连接方式从长轮询升级到WebSocket。
前端收到后端返回的upgrades后，SocketIO会检测浏览器是否支持WebSocket，如果支持，就会启动一个**WebSocket**连接，然后通过这个WebSocket往服务器发一条内容为 probe, 类型为ping的数据。如果这时服务器返回了内容为probe,类型为pong的数据，前端就会把前面建立的HTTP长轮询停掉，后面只使用WebSocket通道进行收发数据。
绿色发送，白色接收。前面的数字表示数据包类型，2是ping，3是pong。
其中“42”，4是message类型标识，2是socket.io的EVENT类型标识，而后面则是事件名称和数据，数据可以是字符串，字典，列表等类型。
SocketIO生命周期内，会间隔一段时间ping-pong一次，检测连接是否中断。
SpringBoot + vue-socket.io 实现双向数据通信
常用的方式是前端使用SocketIO，后端使用node.js实现SocketIO的接口。但由于目前服务端使用JAVA，所以我使用的是Netty-SocketIO开源库，基于Netty网络库编写的WebSocket实现。
服务端
pom
<dependency>
<groupId>com.corundumstudio.socketio</groupId>
<artifactId>netty-socketio</artifactId>
<version>1.7.19</version>
</dependency>SocketIO配置示例
@Configuration
public class SocketIOConfig {
@Bean
public SocketIOServer socketIoServer(PdfConfigurationProperties properties) {
SocketConfig socketConfig = new SocketConfig();
//socketConfig.setTcpNoDelay(false);
//socketConfig.setSoLinger(0);
com.corundumstudio.socketio.Configuration config = new com.corundumstudio.socketio.Configuration();
config.setSocketConfig(socketConfig);
config.setHostname(properties.getSocketIo().getHost());
config.setPort(properties.getSocketIo().getPort());
config.setContext(properties.getSocketIo().getContext());
// socket连接数大小（如只监听一个端口boss线程组为1即可）
config.setBossThreads(properties.getSocketIo().getBossCount());
config.setAllowCustomRequests(properties.getSocketIo()
.isAllowCustomRequests());
// 协议升级超时时间（毫秒），默认10秒。
//config.setUpgradeTimeout(properties.getSocketIo()
//.getUpgradeTimeout());
// Ping消息超时时间（毫秒），默认60秒
//config.setPingTimeout(properties
//.getSocketIo().getPingTimeout());
// Ping消息间隔（毫秒），默认25秒。
//config.setPingInterval(properties
//.getSocketIo().getPingInterval());
return new SocketIOServer(config);
}
@Bean
public SpringAnnotationScanner springAnnotationScanner(SocketIOServer socketServer) {
return new SpringAnnotationScanner(socketServer);
}
}消息处理代码示例（用于前后的消息交互）
/**
* 接受事件消息
* 识别进度
*
* @param client 客户端
* @param request 请求
* @param ids 数据
*/
@OnEvent(PUSH_DATA_EVENT)
public void pushDataEvent(SocketIOClient client, AckRequest request, String... ids) {
removeSession(sessionId(client), IDENTIFY_MAP);
for (String id : ids) {
updateMap(IDENTIFY_MAP, id, client);
}
publishEvent(client, PUSH_DATA_EVENT, request, ids);
}
/**
* 推送统计信息
*/
@OnEvent(Constants.PUSH_IDENTIFY_MESSAGE)
public void statistics(SocketIOClient client, AckRequest request, String bh) {
if (StringUtils.isNotBlank(bh)) {
publishEvent(client, Constants.PUSH_IDENTIFY_MESSAGE, request, bh);
}
publishEvent(client, Constants.PUSH_IDENTIFY_MESSAGE, request);
}
private void publishEvent(SocketIOClient client, String event, AckRequest request, String... ids) {
SocketInteract interact = new SocketInteract(sessionId(client), event, ids);
context.publishEvent(interact);
}
/**
* 向前端发送消息
*/
public void sendMessage(String sessionId) {
SocketIOClient client = CLIENT_MAP.get(sessionId);
if (client != null) {
client.sendEvent(Constants.PUSH_IDENTIFY_MESSAGE);
}
}
通过SpringBoot的事务监听器@TransactionalEventListener处理接收到数据后的相关业务逻辑。
@Component
@Slf4j
@RequiredArgsConstructor
public class SocketListener {
private final SocketService socketService;
@Async
@TransactionalEventListener(fallbackExecution = true)
public void handlerUploadFinished(SocketInteract interact) {
// 调用scoketService查询数据
socketService.process(interact);
}
/**
* 识别进度
* @param event 事件
*/
@Async
@TransactionalEventListener(fallbackExecution = true)
public void handlerPictureOcrFinished(PictureOcrFinishedEvent event) {
// 调用scoketService查询数据
PictureOcrTask task = (PictureOcrTask) event.getSource();
log.debug("文件编号：{}识别进度，总数：{}，进度:{}", task.getPdfBh(), task.getPdfPageSize(), task.getCurrentPage());
socketService.process(new SocketInteract(Constants.PUSH_DATA_EVENT, task.getPdfBh()));
}
}客户端（浏览器）
客户端建立连接，发送消息示例代码
1. 引入vue-socket.io插件
2. 进入vue项目src文件夹下main.js添加下方代码
import VueSocketIO from 'vue-socket.io'
Vue.use(new VueSocketIO({
debug: true,
connection: 'http://socketAdress', //
vuex: {
store,
actionPrefix: 'SOCKET_',
mutationPrefix: 'SOCKET_'
},
}));3. 前端实现数据传输文章来源：https://www.toymoban.com/news/detail-463706.html
//将json发送给服务端，message事件 实现数据交互
this.$socket.$emit('getUser', this.currentUser);
// 建立连接
this.$socket.emit('push_dataItem_event', fileItem);总结
使用SocketIO长连接的特性，无需短轮询频繁建立连接，做进度变动并且实现消息及时通知还是很nice的！
文章来源地址https://www.toymoban.com/news/detail-463706.html
到了这里，关于进度变动实时通知-使用SocketIO实现前后端的通信（基于WebSocket的实时通信库）的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
