欢迎来到尧图网

客户服务 关于我们

您的位置:首页 > 房产 > 家装 > 【SOLUTION】Spring Boot 集成 WebSocket

【SOLUTION】Spring Boot 集成 WebSocket

2025/1/19 13:04:46 来源:https://blog.csdn.net/qq_40890233/article/details/140661156  浏览:    关键词:【SOLUTION】Spring Boot 集成 WebSocket

1. Maven 依赖

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

2. 注册ServerEndpointExporter、WebSocketConfigProperties

WebSocketConfigProperties:

  1. 建议代码中的第三方插件配置属性都通过@ConfigurationProperties注解类实现配置属性的定义,方便Spring管理和维护
  2. 建议第三方插件资源都设置一个enabel属性控制资源的加载,可以节省内存开销和代码安全
package xx.xx.xx.xx.xx;import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;@Data
@ConfigurationProperties("spring.socket")
public class WebSocketConfigProperties {//是否开启WebSocketprivate boolean enabel;
}

ServerEndpointExporter:

  1. WebSocket 需要创建一个ServerEndpointExporter的实例,ServerEndpointExporter负责注册ServerEndpoint和管理ServerContainer
  2. 配置类继承BeanFactoryAware 是因为WebSocket在每次连接时都会重新创建ServerEndpoint类,所以导致在ServerEndpoint类中无法通过@Autowrite自动注入IOC中的Bean,需要借助BeanFactory来获取对应的类
package xx.xx.xx.xx.xx;import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;@ConditionalOnProperty(prefix = "spring.socket", name = "enabel", havingValue = "true")
@Configuration
@EnableConfigurationProperties(WebSocketConfigProperties.class)
public class WebSocketAutoConfig implements BeanFactoryAware {@Beanpublic ServerEndpointExporter serverEndpointExporter() {return new ServerEndpointExporter();}@Overridepublic void setBeanFactory(BeanFactory beanFactory) throws BeansException {SocketBaseService.setBeanFactory(beanFactory);}
}

配置SocketBaseService

这里配置SocketBaseService类的原因有以下几点:

  1. 每个Socket端点都需要创建一个ServerEndpoint类,所以在代码上可以节省时间成本且规范ServerEndpoint
  2. ServerEndpoint类有4个声明周期@OnOpen@OnMessage@OnError@OnClose,而每次连接在四个周期方法内都需要统一处理一些逻辑,例如在@OnOpen时需要储存Session,在@OnClose时需要删除Session@OnError时需要记录错误日志等;所以如果不写SocketBaseService类,则需要在每个ServerEndpoint中写重复的逻辑,导致代码可读性低,维护难等问题
  3. 子级只需要通过继承SocketBaseService 抽象类,实现内部4个生命周期的抽象方法,即可实现对应的业务逻辑,由于子级的4个生命周期方法是父级抽象出来的,所以子级方法上就无需在标注@OnOpen@OnMessage@OnError@OnClose注解
package xx.xx.xx.xx.xx;import cn.hutool.core.util.*;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Autowired;import javax.websocket.*;
import java.lang.reflect.Field;
import java.util.Arrays;
import java.util.concurrent.ConcurrentHashMap;public abstract class SocketBaseService {protected String randomId;protected Session session;protected static BeanFactory beanFactory;/*** 会话池*/private static ConcurrentHashMap<String, Session> sessionPool = new ConcurrentHashMap<>();/*** 解析自动装配属性*/private void processAutowriteFields() {Field[] fields = this.getClass().getDeclaredFields();Arrays.stream(fields).forEach(field -> {Autowired annotation = field.getAnnotation(Autowired.class);if (ObjectUtil.isEmpty(annotation)) {return;}Object bean = SocketBaseService.beanFactory.getBean(field.getType());if (ObjectUtil.isNotEmpty(bean)) {try {field.setAccessible(true);field.set(this, bean);} catch (IllegalAccessException e) {}}});}@OnOpenpublic void onConnect(Session session) {this.processAutowriteFields();this.randomId = IdUtil.nanoId();this.session = session;sessionPool.put(randomId, session);this.afterConnect(session);}/*** 连接回调** @param session*/public abstract void afterConnect(Session session);/*** 消息** @param message*/@OnMessagepublic void onMessage(String message) {this.afterMessage(message);}/*** 接受消息回调** @param message*/public abstract void afterMessage(String message);/*** 连接错误** @param session* @param error*/@OnErrorpublic void onError(Session session, Throwable error) {this.afterError(error);}/*** 连接错误回调** @param error*/public abstract void afterError(Throwable error);/*** 关闭连接*/@OnClosepublic void onClose() {this.beforeClose();sessionPool.remove(randomId);}/*** 关联连接回调*/public abstract void beforeClose();
}

案例:文件上传进度监听

在面对大文件上传时通常遇到无法即时获取上传进度的问题,在没有WebSocket之前都是异步上传,然后通过轮询的方式实时获取上传进度
而在有了WebSocket后就更方便了,可以实现持续连接,实现实时获取上传进度,步骤如下:

第一步:异步上传

由于userFileService.asynceUpload是异步方法,所以这里不会等待userFileService.asynceUpload方法执行,而是立即返回任务id

Controller
    @PostMapping("/asynceUpload")@Transactionalpublic Object asynceUpload(FileInfo info) {//获取一个异步上传任务id,方便后续跟踪文件上传进度String taskId = userFileService.getAsynceUploadTaskId(info);userFileService.asynceUpload(info, taskId);return taskId;}
Service

Spring如果想实现异步方法,需要在方法上添加@Async注解,并且在启动类或配置类上添加@EnabelAsync注解开启异步任务

@Async
@Override
public String asynceUpload(FileInfo info, String taskId) {
//处理文件
//更新进度
}

第二步:监听上传进度

监听步骤:

  1. 继承上文的SocketBaseService 抽象类,实现4个生命周期抽象方法
  2. afterConnect方法中注册定时任务,实现定时获取任务进度
  3. afterMessage中接受接口传的任务id
  4. beforeClose中关闭定时任务
  5. exce方法中编写获取任务进度的逻辑,并通过Session返回获取到的进度信息

注意:

  1. 子类中的teamDataService属性通过@Autowite注入成功是因为,在SocketBaseService onConnect方法中执行了processAutowriteFields方法,而processAutowriteFields通过获取当前类中标注了@Autowite注解的字段,再根据字段的类型通过BeanFactoryIOC获取对应的实例的方式注册
package xx.xx.xx.xx.xx.xx;import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import xx.common.web.socket.SocketBaseService;
import xx.xx.client.server.domain.TeamData;
import xx.xx.client.server.service.TeamDataService;import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;@Service
@ServerEndpoint("/socket/getSyncFileTask")
public class UserFileSocketController extends SocketBaseService {private String scheduleId;private List<String> tasks = new ArrayList<>();@Autowiredprivate TeamDataService teamDataService;@Overridepublic void afterConnect(Session session) {scheduleId = CronUtil.schedule("*/2 * * * * *", new Task() {@Overridepublic void execute() {exce();}});}@Overridepublic void afterMessage(String message) {this.tasks = JSONUtil.toList(message, String.class);}@Overridepublic void afterError(Throwable error) {}@Overridepublic void beforeClose() {if (ObjectUtil.isNotEmpty(scheduleId)) {CronUtil.remove(scheduleId);}}private void exce() {if (session.isOpen()) {//获取任务进度List<TeamData> taskList = teamDataService.findByIds(tasks);//返回进度信息session.getAsyncRemote().sendText(JSONUtil.toJsonStr(taskList.stream().filter(el -> StrUtil.startWith(el.getKey(), "SYNCE_UPLOAD_")).collect(Collectors.toList())));}}
}

版权声明:

本网仅为发布的内容提供存储空间,不对发表、转载的内容提供任何形式的保证。凡本网注明“来源:XXX网络”的作品,均转载自其它媒体,著作权归作者所有,商业转载请联系作者获得授权,非商业转载请注明出处。

我们尊重并感谢每一位作者,均已注明文章来源和作者。如因作品内容、版权或其它问题,请及时与我们联系,联系邮箱:809451989@qq.com,投稿邮箱:809451989@qq.com