【Spring Boot 实现 WebSocket实时数据推送-服务端】

二、WebSocket服务端类三、WebSocket的连接池类四、启动Spring Boot服务五、测试WebSocket连接WebSocket在线测试工具：http://www.easyswoole.com/wstool.html测试连接服务启动的IP：172.18.42.29服务端口：14785WS的URl：/endPoint入参：1六、遇到的问题服务启动报错：原因：WebSocketServ

秋水挽春风

5876人浏览 · 2022-06-09 11:57:14

秋水挽春风 · 2022-06-09 11:57:14 发布

一、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
$[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-tI6kkgAw-1654745595807)(C:\Users\coding\AppData\Roaming\Typora\typora-user-images\image-20220609100123036.png)]$

六、遇到的问题

服务启动报错：
$[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-hsVNiQmt-1654745595808)(C:\Users\coding\AppData\Roaming\Typora\typora-user-images\image-20220609103318123.png)]$

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类从日志处理的切面中排除掉即可。

加入业务处理时报错：
$[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-FyZWgzr1-1654745595809)(C:\Users\coding\AppData\Roaming\Typora\typora-user-images\image-20220609110343177.png)]$

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());
        }
    }

}