一、WebSocket配置类
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
/**
* @author HFL
* @date 2022/5/16 14:49
* 配置类
*/
@Configuration
public class WebSocketConfiguration {
@Bean
public ServerEndpointExporter serverEndpointExporter(){
return new ServerEndpointExporter();
}
}
二、WebSocket服务端类
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
/**
* @author HFL
* @date 2022/5/16 15:17
* Websocket应用实现:
* 1.建立连接，连接放入连接池
* 2.关闭连接，连接移出连接池
* 3.接收客户端发送的消息，并做出相应处理
* 4.注入业务层的service
* [注意：Spring管理的Bean是单例模式的，而WebSocket不是单例，注入时需要处理一下]
* 5.异常处理，连接移除连接池
*/
@Slf4j
@Component
@ServerEndpoint("/endPoint/{screen}")
public class WebSocketServer {
/**
* 建立连接成功调用 (Session + 场景)
*/
@OnOpen
public void onOpen(Session session,@PathParam("screen") String screen) throws IOException {
log.info("[onOpen][session({}) 接入, [screen: {}]", session, screen);
WebSocketServerPool.addDataConnect(session, screen);
//WebSocketServerPool.sendMessage(session, configurationScreenService.queryAllJsonById(screen));
}
/**
* 关闭连接时调用
* @param session 连接
*/
@OnClose
public void onClose(Session session, CloseReason closeReason) {
log.info("[onClose][session({}) 连接关闭。关闭原因是({})}]", session, closeReason);
WebSocketServerPool.removeConnect(session);
}
/**
* 错误时调用
* @param session 连接
* @param throwable 异常
*/
@OnError
public void onError(Session session, Throwable throwable) {
log.info("[onClose][session({}) 发生异常]", session, throwable);
WebSocketServerPool.removeConnect(session);
}
/**
* 收到客户端信息后，根据接收到的信息进行处理
* @param session 连接
* @param message 数据消息
*/
@OnMessage
public void onMessage(Session session, String message) {
log.info("[onOpen][session({}) 接收到一条消息({})]", session, message);
// TODO: 2022/5/18 对于客户端发送的指令信息，解析后进行对应的逻辑处理
}
}
三、WebSocket的连接池类
import javax.websocket.Session;
import java.io.IOException;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
/**
* @author HFL
* @date 2022/5/16 9:39
* Websocket连接池、对连接池内连接操作 和数据推送方法
*/
public class WebSocketServerPool {
/**
* WebSocket连接池
*/
private static ConcurrentMap<Session, String> dataConnect = new ConcurrentHashMap<>();
/**
* 将websocket连接，放入连接池
* @param session websocket连接
* @param screen 场景ID
*/
public static void addDataConnect(Session session, String screen){
dataConnect.put(session, screen);
Iterator<Map.Entry<Session, String>> iterator = dataConnect.entrySet().iterator();
synchronized (iterator){
//移除失效连接
while(iterator.hasNext()){
Map.Entry<Session, String> entry = iterator.next();
Session sessionNew = entry.getKey();
Map<String, Object> userProperties = sessionNew.getUserProperties();
if(null != userProperties && null != userProperties.get("ReadyState") && "0" != String.valueOf(userProperties.get("ReadyState"))){
iterator.remove();
}
}
}
}
/**
* 将websocket连接从连接池中移除
* @param session websocket连接
*/
public static void removeConnect(Session session){
Iterator<Map.Entry<Session, String>> iterator = dataConnect.entrySet().iterator();
synchronized (iterator){
//主动移除连接
while (iterator.hasNext()){
if(session.equals(iterator.next().getKey())){
iterator.remove();
}
}
}
}
/**
* 获取连接池中所有连接
* @return 连接池所有数据
*/
public static ConcurrentMap<Session, String> getDataConnect(){
return dataConnect;
}
/**
* Websocket消息推送
* @param session 连接
* @param message 消息主体
* @throws IOException I/O异常
*/
public static void sendMessage(Session session, String message) throws IOException {
session.getBasicRemote().sendText(message);
}
}
四、启动Spring Boot服务
import com.ljgk.ems.maitreya.user.annotation.EnableLoginArgResolver;
import com.ljgk.ems.maitreya.validator.config.EnableFormValidator;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.cloud.openfeign.EnableFeignClients;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.EnableAspectJAutoProxy;
import org.springframework.core.env.Environment;
import org.springframework.scheduling.annotation.EnableScheduling;
import java.net.InetAddress;
import java.net.UnknownHostException;
/**
* @author HFL
* @date 2022/5/13 8:40
* 仅做测试，只需要@SpringBootApplication注释即可，
* 其他注释，用于其他的功能，例如：
* @EnableScheduling用于扫描开启定时任务即@Scheduled注解的方法
*/
@SpringBootApplication
@EnableDiscoveryClient
@Configuration
@EnableFeignClients(value = {"com.ljgk.ems.maitreya"})
@EnableAspectJAutoProxy(proxyTargetClass = true, exposeProxy = true)
@Slf4j
@EnableLoginArgResolver
@EnableFormValidator
@EnableScheduling
public class ServerApplication {
public static void main(String[] args) throws UnknownHostException {
ConfigurableApplicationContext application = SpringApplication.run(ConfigurationServerApplication.class, args);
}
}
五、测试WebSocket连接
- WebSocket在线测试工具：
http://www.easyswoole.com/wstool.html - 测试连接
服务地址：ws://172.18.42.29:14785/endPoint/1
服务启动的IP：172.18.42.29
服务端口：14785
WS的URl：/endPoint
入参：1
六、遇到的问题
-
服务启动报错：
2022-06-09 10:31:27.616:[ERROR] [main:18941] [org.springframework.boot.SpringApplication.reportFailure:826] --> Application run failed java.lang.IllegalStateException: Failed to register @ServerEndpoint class: class com.ljgk.ems.maitreya.configuration.websocket.WebSocketServer$$EnhancerBySpringCGLIB$$8a624780 at org.springframework.web.socket.server.standard.ServerEndpointExporter.registerEndpoint(ServerEndpointExporter.java:159) at org.springframework.web.socket.server.standard.ServerEndpointExporter.registerEndpoints(ServerEndpointExporter.java:134) at org.springframework.web.socket.server.standard.ServerEndpointExporter.afterSingletonsInstantiated(ServerEndpointExporter.java:112) at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:896) at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:878) at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:550) at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:141) at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:747) at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:397) at org.springframework.boot.SpringApplication.run(SpringApplication.java:315) at org.springframework.boot.SpringApplication.run(SpringApplication.java:1226) at org.springframework.boot.SpringApplication.run(SpringApplication.java:1215) at com.ljgk.ems.maitreya.ConfigurationServerApplication.main(ConfigurationServerApplication.java:36) Caused by: javax.websocket.DeploymentException: UT003027: Class class com.ljgk.ems.maitreya.configuration.websocket.WebSocketServer$$EnhancerBySpringCGLIB$$8a624780 was not annotated with @ClientEndpoint or @ServerEndpoint at io.undertow.websockets.jsr.ServerWebSocketContainer.addEndpointInternal(ServerWebSocketContainer.java:735) at io.undertow.websockets.jsr.ServerWebSocketContainer.addEndpoint(ServerWebSocketContainer.java:628) at org.springframework.web.socket.server.standard.ServerEndpointExporter.registerEndpoint(ServerEndpointExporter.java:156) ... 12 common frames omitted原因：
WebSocketServer类被代理了，我出现这个问题的原因是做了整个项目的接口拦截，然后做了接口的日志切面处理，导致这个类被代理了，而@ServerEndpoint注解在处理WebSocketServer时，取到了代理的类，无法处理导致的异常。解决办法：
将WebSocketServer类从日志处理的切面中排除掉即可。
-
加入业务处理时报错：
java.lang.NullPointerException: null at com.ljgk.ems.maitreya.configuration.websocket.WebSocketServer.onOpen(WebSocketServer.java:47) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:497) at io.undertow.websockets.jsr.annotated.BoundMethod.invoke(BoundMethod.java:87) at io.undertow.websockets.jsr.annotated.AnnotatedEndpoint$3.run(AnnotatedEndpoint.java:158) at io.undertow.websockets.jsr.ServerWebSocketContainer$1.call(ServerWebSocketContainer.java:170) at io.undertow.websockets.jsr.ServerWebSocketContainer$1.call(ServerWebSocketContainer.java:167)原因：
通过注解@Resource自动注入的ConfigurationScreenService的Bean为空，即Spring容器中，ConfigurationScreenService没有注入进去，因为Spring管理的Bean的作用域和WebSocket的作用域不同，Spring管理的Bean都是单例，WebSocket不是。
解决办法： 将截图中通过@Resource注解注入ConfigurationScreenService的方式换成下面方式：
/** * @author HFL * @date 2022/5/16 15:17 * Websocket应用实现: * 1.建立连接，连接放入连接池 * 2.关闭连接，连接移出连接池 * 3.接收客户端发送的消息，并做出相应处理 * 4.注入业务层的service * [注意：Spring管理的Bean是单例模式的，而WebSocket不是单例，注入时需要处理一下] * 5.异常处理，连接移除连接池 */ @Slf4j @Component @ServerEndpoint("/endPoint/{screen}") public class WebSocketServer { private static ConfigurationScreenService configurationScreenService; @Resource public void setConfigurationScreenService(ConfigurationScreenService configurationScreenService){ WebSocketServer.configurationScreenService = configurationScreenService; } /** * 建立连接成功调用 (Session + 场景ID) */ @OnOpen public void onOpen(Session session,@PathParam("screen") String screen) throws IOException { log.info("[onOpen][session({}) 接入, [screen: {}]", session, screen); WebSocketServerPool.addDataConnect(session, screen); WebSocketServerPool.sendMessage(session, configurationScreenService.queryAllJsonById(screen)); } ... }
七、扩展（实现实时推送数据）
-
定时任务，轮询连接池中的连接，并取到对于的场景、绑定的设备，即可查询最新数据，最后推送至客户端。（最简单实现）
import cn.hutool.json.JSONUtil; import com.ljgk.ems.maitreya.configuration.entity.Content; import com.ljgk.ems.maitreya.configuration.service.ConfigurationScreenDataService; import com.ljgk.ems.maitreya.configuration.vm.ScreenDataJsonVm; import com.ljgk.ems.maitreya.configuration.websocket.WebSocketServerPool; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import javax.annotation.Resource; import javax.websocket.Session; import java.io.IOException; import java.util.List; import java.util.concurrent.ConcurrentMap; /** * @author HFL * @date 2022/6/88:35 */ @Slf4j @Component public class ScreenPushScheduledTask { @Resource private ConfigurationScreenDataService configurationScreenDataService; /** * 5秒一次 * 定时场景数据推送 */ @Scheduled(cron = "0/5 * * * * ? ") public void executeScreenDataPush(){ try { ConcurrentMap<Session, String> dataConnect = WebSocketServerPool.getDataConnect(); //查询待推送场景 dataConnect.keySet().forEach(session -> { try { String screen = dataConnect.get(session); //查询需要的场景对应的元件需要的值，并按规则组装成JSON List<ScreenDataJsonVm> screenDataJsonVms = configurationScreenDataService.queryDataJson(screen); WebSocketServerPool.sendMessage(session, JSONUtil.toJsonStr(screenDataJsonVms)); } catch (IOException e) { log.error("WebSocket SendMessage Error, Session:[{}], Exception : [{}]", session, e.getMessage()); } }); }catch (Exception e){ log.error("WebSocket Scheduler Executor Error : [{}]", e.getMessage()); } } } -
监听Binlog日志，将MySql中变化数据取出，推送至客户端。
-
RocketMq实现，将变化数据写入队列，WS服务端消费队列中数据时，推送至客户端。文章来源：https://www.toymoban.com/news/detail-424830.html
八、参考
1. WebSocket原理及技术简介
2. Java实现WebSocket服务端
3. Java Websocket——服务器端
4.【WebSocket】SpringBoot整合WebSocket 注入Bean的方式文章来源地址https://www.toymoban.com/news/detail-424830.html
到了这里，关于【Spring Boot 实现 WebSocket实时数据推送-服务端】的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
