springboot集成websocket（SpringBoot+Vue+Websocket 实现服务器端向客户端主动发送消息）

本文目录

• SpringBoot+Vue+Websocket 实现服务器端向客户端主动发送消息
• SpringBoot - WebSocket
• springboot2.1.3整合websocket和websocket-security支持跨域连接
• Websocket使用
• SpringBoot整合Websocket实现即时聊天功能
• 实战spring-boot-starter-websocket之断网心跳续期实践
• Springboot整合Websocket遇到的坑
• springbootstart-websocket修改序列化

SpringBoot+Vue+Websocket 实现服务器端向客户端主动发送消息

本文通过一个实际的场景来介绍在前后端分离的项目中通过 WebSocket 来实现服务器端主动向客户端发送消息的应用。主要内容如下

Websocket 是一种在单个 TCP 连接上进行全双工通信的协议。WebSocket 连接成功后，服务端与客户端可以双向通信。在需要消息推送的场景，Websocket 相对于轮询能更好的节省服务器资源和带宽，并且能够更实时地进行通讯。

具体如下特点

在客户端的列表数据中有个 status 字段，服务器端需要花费较长的时间进行处理，处理完成后才会更新对应数据的 status 字段值，通过 Websocket 的处理流程如下：

通过注入 ServerEndpointExporter 类，用于在项目启动的时候自动将使用了 @ServerEndpoint 注解声明的 Websocket endpoint 注册到 WebSocketContainer 中。

为什么增加一个 ServerEndpointExporter Bean，并通过在一个类上增加 @ServerEndpoint 和 @Component 注解就可以实现服务器端 Websocket 功能，这里简单解析一下。

java 定义了一套 javax.servlet-api, 一个 HttpServlet 就是一个 HTTP 服务。java websocket 并非基于 servlet-api 简单扩展, 而是新定义了一套 javax.websocket-api。

一个 websocket 服务对应一个 Endpoint。与 ServletContext 对应, websocket-api 也定义了 WebSocketContainer, 而编程方式注册 websocket 的接口是继承自 WebSocketContainer 的 ServerContainer。

一个 websocket 可以接受并管理多个连接, 因此可被视作一个 server。主流 servlet 容器都支持 websocket, 如 tomcat, jetty 等。看 ServerContainer api 文档, 可从 ServletContext attribute 找到 ServerContainer。

SpringBoot - WebSocket

***隐藏网址***

Http 协议有一个缺陷：通信只能由客户端发起

***隐藏网址***

轮询的效率低下，有一定的延迟性，并且频繁的发送请求也会造成资源的浪费。

使用 WebSocket 协议可以 实现由服务器端主动向客户端推送消息 ，当然客户端也可以向服务器端发送消息。

这里仅介绍利用 Spring 框架使用 WebSocket 的方式，原因：Spring 使用 WebSocket 简便且易于扩展。

SpringBoot 使用 WebSocket 非常方便，依赖上仅需要添加相应的 Starter 即可。

先给出概要的开发步骤：

其实到这里，基础的 websocket 服务已经搭建好了，剩下的可以自己在 handler 与 interceptor 中写自己的业务逻辑了

前端页面：

先启动服务器端 SpringBoot 应用，再使用前端页面点击测试一下就 ok 了

springboot2.1.3整合websocket和websocket-security支持跨域连接

springboot整合websocket和websocket-security支持跨域连接

***隐藏网址***
完整配置参考com.example.websocket包

所以继承了StompSubProtocolHandler，WebMvcStompEndpointRegistry，DelegatingWebSocketMessageBrokerConfiguration这三个类，添加websocket自定义拦截器接口，可以在拦截器中自定义websocket授权决策检查

配置类可以重写这个方法，默认该方法返回false，看方法的名字是关闭同源策略，但是只重写这个方法不能解决跨域的问题，还需要在registerStompEndpoints方法中设置允许的域名，"*"代表所有

如果有误导人的地方，欢迎大神批评指正！

Websocket使用

springboot 整合 websocket的使用，Demo实例在码云里；

***隐藏网址***

2. WebSocket错误：java.lang.IllegalStateException: The WebSocket session has been closed and no method (apart from close()) may be called on a closed session ，出现这个问题的原因是：在建立连接时保存的对象id，在断开连接时移除失败，所以对应services中的onOpen()方法存储的id 以及 onClose()方法中移除的id ，即可。

3.本地开发的时候，需要手写WebsocketConfiguration类，交由spring管理 ServerEndpointExporter 类，但是当打包war包然后运行在tomcat等中间件时，此类将影响项目启动websocket，所以打包时需注意这个文件

SpringBoot整合Websocket实现即时聊天功能

近期，公司需要新增即时聊天的业务，于是用websocket 整合到Springboot完成业务的实现。

一、我们来简单的介绍下websocket的交互原理：

1.客户端先服务端发起websocket请求；

2.服务端接收到请求之后，把请求响应返回给客户端；

3.客户端与服务端只需要一次握手即可完成交互通道；
***隐藏网址***

***隐藏网址***

***隐藏网址***

  三、开始我们的实现java后端的实现

  1.添加依赖

  《!-- websocket --》

        《dependency》

            《groupId》org.springframework.boot《/groupId》

            《artifactId》spring-boot-starter-websocket《/artifactId》

        《/dependency》

        《dependency》

            《groupId》org.springframework《/groupId》

            《artifactId》spring-websocket《/artifactId》

            《version》${spring.version}《/version》

        《/dependency》

        《dependency》

            《groupId》org.springframework《/groupId》

            《artifactId》spring-messaging《/artifactId》

            《version》${spring.version}《/version》

        《/dependency》

        《!-- WebSocket --》

  2.配置config

@ConditionalOnWebApplication

@Configuration

@EnableWebSocketMessageBroker

public class WebSocketConfig extends AbstractSessionWebSocketMessageBrokerConfigurer {

    @Bean

    public ServerEndpointExporter serverEndpointExporter(){

        return  new ServerEndpointExporter();

    }

    @Bean

    public CustomSpringConfigurator customSpringConfigurator() {

        return new CustomSpringConfigurator();

    }

    @Override

    protected void configureStompEndpoints(StompEndpointRegistry registry) {

        registry.addEndpoint("/websocket").setAllowedOrigins("*")

                .addInterceptors(new HttpSessionHandshakeInterceptor()).withSockJS();

    }
public class CustomSpringConfigurator extends ServerEndpointConfig.Configurator implements ApplicationContextAware {

    private static volatile BeanFactory context;

    @Override

    public 《T》 T getEndpointInstance(Class《T》 clazz) throws InstantiationException {

        return context.getBean(clazz);

    }

    @Override

    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {

        CustomSpringConfigurator.context = applicationContext;

    }

    @Override

    public void modifyHandshake(ServerEndpointConfig sec,

                                HandshakeRequest request, HandshakeResponse response) {

        super.modifyHandshake(sec,request,response);

***隐藏网址***

***隐藏网址***

***隐藏网址***

        }

    }

}

@SpringBootApplication

@EnableCaching

@ComponentScan("com")

@EnableWebSocket

public class Application extends SpringBootServletInitializer {

static final Logger logger = LoggerFactory.getLogger(Application.class);

    @Override

    protected SpringApplicationBuilder configure(SpringApplicationBuilder application) {

        return application.sources(Application.class);

    }

需要注意的是: @EnableWebSocket  一定要加在启动类上，不然springboot无法对其扫描进行管理;

@SeverEndpoint --将目标类定义成一个websocket服务端，注解对应的值将用于监听用户连接的终端访问地址，客户端可以通过URL来连接到websocket服务端。

四、设计思路：用map《房间id， 用户set》来保存房间对应的用户连接列表，当有用户进入一个房间的时候，就会先检测房间是否存在，如果不存在那就新建一个空的用户set，再加入本身到这个set中，确保不同房间号里的用户session不串通！

/**

* Create by wushuyu

* on 2020/4/30 13:24

*

*/

@ServerEndpoint(value = "/websocket/{roomName}", configurator = CustomSpringConfigurator.class)

@Component

public class WebSocketRoom {

    //连接超时--一天

    private static final long MAX_TIME_OUT = 24*60*60*1000;

    // key为房间号，value为该房间号的用户session

    private static final Map《String, Set《Session》》 rooms = new ConcurrentHashMap《》();

    //将用户的信息存储在一个map集合里

    private static final Map《String, Object》 users = new ConcurrentHashMap《》();

/**

*{roomName} 使用通用跳转，实现动态获取房间号和用户信息  格式：roomId|xx|xx

*/

    @OnOpen 

    public void connect(@PathParam("roomName") String roomName, Session session) {

        String roomId = roomName.split(";

        String nickname = roomName.split(";

        String loginId = roomName.split(";

        //设置连接超时时间

            session.setMaxIdleTimeout(MAX_TIME_OUT);

        try {
          //可实现业务逻辑

            }

            // 将session按照房间名来存储，将各个房间的用户隔离

            if (!rooms.containsKey(roomId)) {

                // 创建房间不存在时，创建房间

                Set《Session》 room = new HashSet《》();

                // 添加用户

                room.add(session);

                rooms.put(roomId, room);

            } else { // 房间已存在，直接添加用户到相应的房间             

                if (rooms.values().contains(session)) {//如果房间里有此session直接不做操作

                } else {//不存在则添加

                    rooms.get(roomId).add(session);

                }

            }

            JSONObject jsonObject = new JSONObject();

            -----

            //根据自身业务情况实现业务

            -----

            users.put(session.getId(), jsonObject);

            //向在线的人发送当前在线的人的列表    -------------可有可无，看业务需求

            List《ChatMessage》 userList = new LinkedList《》();

            rooms.get(roomId)

                    .stream()

                    .map(Session::getId)

                    .forEach(s -》 {

                        ChatMessage chatMessage = new ChatMessage();

                        chatMessage.setDate(new Date());

                        chatMessage.setStatus(1);

                        chatMessage.setChatContent(users.get(s));

                        chatMessage.setMessage("");

                        userList.add(chatMessage);

                    });

//        session.getBasicRemote().sendText(JSON.toJSONString(userList));

            //向房间的所有人群发谁上线了

            ChatMessage chatMessage = new ChatMessage();  ----将聊天信息封装起来。

            chatMessage.setDate(new Date());

            chatMessage.setStatus(1);

            chatMessage.setChatContent(users.get(session.getId()));

            chatMessage.setMessage("");

            broadcast(roomId, JSON.toJSONString(chatMessage));

            broadcast(roomId, JSON.toJSONString(userList));

        } catch (Exception e) {

            e.printStackTrace();

        }

    }

    @OnClose

    public void disConnect(@PathParam("roomName") String roomName, Session session) {

        String roomId = roomName.split(";

        String loginId = roomName.split(";

        try {

            rooms.get(roomId).remove(session);

            ChatMessage chatMessage = new ChatMessage();

            chatMessage.setDate(new Date());

            chatMessage.setUserName(user.getRealname());

            chatMessage.setStatus(0);

            chatMessage.setChatContent(users.get(session.getId()));

            chatMessage.setMessage("");

            users.remove(session.getId());

            //向在线的人发送当前在线的人的列表  ----可有可无，根据业务要求

            List《ChatMessage》 userList = new LinkedList《》();

            rooms.get(roomId)

                    .stream()

                    .map(Session::getId)

                    .forEach(s -》 {

                        ChatMessage chatMessage1 = new ChatMessage();

                        chatMessage1.setDate(new Date());

                        chatMessage1.setUserName(user.getRealname());

                        chatMessage1.setStatus(1);

                        chatMessage1.setChatContent(users.get(s));

                        chatMessage1.setMessage("");

                        userList.add(chatMessage1);

                    });

            broadcast(roomId, JSON.toJSONString(chatMessage));

            broadcast(roomId, JSON.toJSONString(userList));

        } catch (Exception e) {

            e.printStackTrace();

        }

    }

    @OnMessage

    public void receiveMsg( String msg, Session session) {

        try {

                ChatMessage chatMessage = new ChatMessage();

                chatMessage.setUserName(user.getRealname());

                chatMessage.setStatus(2);

                chatMessage.setChatContent(users.get(session.getId()));

                chatMessage.setMessage(msg);

                // 按房间群发消息

                broadcast(roomId, JSON.toJSONString(chatMessage));

            }

        } catch (IOException e) {

            e.printStackTrace();

        }

    }

    // 按照房间名进行群发消息

    private void broadcast(String roomId, String msg) {

        rooms.get(roomId).forEach(s -》 {

            try {

                s.getBasicRemote().sendText(msg);  -----此还有一个getAsyncRemote() 

            } catch (IOException e) {

                e.printStackTrace();

            }

        });

    }

    @OnError

    public void onError(Throwable error) {

        error.printStackTrace();

    }

}

***隐藏网址***

实战spring-boot-starter-websocket之断网心跳续期实践

业务中需要应用到Websocket长连接进行数据传输，由于服务使用的是Zuul1.0版本，对ws协议支持较弱，后续尝试使用了 spring-boot-starter-websocket 来完成的。关于怎么集成的话网上有非常多的文章了，我就不多费口舌了。

我们目前实现的功能是可以通过WebSocket调用接口发送埋点，另外还需要监听用户离开的事件为这个埋点画上一个终止访问时间。目前测试下场景有：

前4点触发了任意操作，服务端都会监听到 DISCONNECT 离开事件。但是第5点直接断网， 服务端竟然是无感知的，这个时候产生的问题就是客户断网了，服务端是认为在线的，如果不重新联网登录的话，那么这个用户将会一直一直在线，埋点会一直计算。完了个蛋~

至于为什么断网之后，ws会认为他是在线的， 可能管道打开了之后由于断网导致断开时间发送不出去吧。

我目前能够想到比较简单的办法就是： 心跳续约

捋清了思路，大概就知道如何做了。

然后特意看了下 spring-boot-starter-websocket 的源码，发现其实他有提供此功能。

先说下如何实现:

我们是在在实现了 DelegatingWebSocketMessageBrokerConfiguration 的配置类中重写 configureMessageBroker 方法。

比较关键就是 setTaskScheduler 和 setHeartbeatValue 一个负责调度、一个负责配置间隔。
这俩要么都填要么都不填。

配置了这俩参数之后，服务启动的时候会触发一个 HeartbeatTask 线程来专门维持心跳。

我们可以看看他的流程是如何运转的。

核心任务调度类: org.springframework.messaging.simp.broker.SimpleBrokerMessageHandler.HeartbeatTask

服务端在启动的时候 SimpleBrokerMessageHandler 在初始化完成之后会回调 start 的方法，然而他会触发一个 startInternal 开始调度任务，判断依据就是有没有配置 TaskScheduler ( 对应的就是配置类中的setTaskScheduler )，一旦启动之后，会根据你给的心跳数组 serverHeartbeat ，来选择调度时间。

org.springframework.messaging.simp.broker.SimpleBrokerMessageHandler

检测心跳、超过间隔则剔除、并且定期回写心跳给客户端。

还有一点需要注意的是读的间隔时间：
假设用户网络不好，心跳漏发了一次，这个时候如果按照本次的逻辑而言，该用户的最后心跳时间肯定会超时。而它的做法是， 将设定的读超时时间 3 ，就相当于有3次机会*。

这个在创建session的时候 SessionInfo 里面就已经做好了处理:
org.springframework.messaging.simp.broker.SimpleBrokerMessageHandler.SessionInfo#SessionInfo

所以这个时候，你可能需要根据业务能够接受的时长去配置，也别忘了这个事。

还有很关键的一点就是让客户端的心跳发送间隔和服务端尽可能保持一致，不然有可能出现莫名其妙的下线情况，尽可能还是在这种地方加好日志。

好了，希望在遇到断网问题的时候，能够帮助到你。

如果有疑问请留言，我会尽快答复。

Springboot整合Websocket遇到的坑

如果客户端关闭了websocket，但服务端没有监听到关闭事件，即onClose方法没有调用，这是会发生的情况

此时如果服务端向客户端推送消息，会出现异常告诉开发者：关闭了一个连接，并重新调用onClose方法

springbootstart-websocket修改序列化

在springboot中使用websocket时，默认情况下使用的是jackson库进行消息的序列化和反序列化。如果需要修改websocket消息的序列化方式，可以按照以下步骤进行操作：
1、创建自定义的消息序列化器：可以实现org.springframework.messaging.simp.stomp.StompEncoder接口来创建自定义的消息序列化器。该接口定义了两个方法encode和supports,分别用于将消息编码为字节数组和判断是否支持某种消息类型。
2、配置websocket消息转换器：在SpringBoot的配置类中，通过重写configureMessageConverters方法或添加@Bean注解的方式配置WebSocket消息转换器。在配置中，将自定义的消息序列化器添加到WebSocketMessageBrokerConfigurer中的messageConverter中。
3、使用自定义的消息序列化器：在编写WebSocket控制器时，可以直接使用自定义的消息序列化器。