实战spring-boot-starter-websocket之

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

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

1. 小程序进程杀死
2. 主动退出
3. 后台挂起【会认为退出，重新唤醒则连接】
4. 横跨页面断线重连
5. 直接断网

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

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

如果有知道的小伙伴可以留言告知下。

如何解决?

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

1. 让客户端定时通过管道发送心跳.
2. 服务端会记录每个用户最后一次的心跳时间,并且配置一个心跳最大间隔时长。
3. 开启一个定时调度的任务，根据每个用户的最后一次心跳间隔时间和配置的心跳间隔时长来判断用户是否存在极端离线无法续约心跳的场景。
4. 遇到超过间隔时长的用户直接剔除。

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

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

先说下如何实现:

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

@Override
protected void configureMessageBroker(MessageBrokerRegistry registry) {
   // prefix for subscribe
   long[] serverHeartbeat = new long[2];
   serverHeartbeat[0] = 10000; // 对应的是写间隔
   serverHeartbeat[1] = 5000; // 对应的是读间隔时间，如果客户端长时间没有访问的话，那么会认为连接失效
   DefaultManagedTaskScheduler defaultManagedTaskScheduler = new DefaultManagedTaskScheduler();
    registry.enableSimpleBroker("/topic").setTaskScheduler(defaultManagedTaskScheduler).setHeartbeatValue(serverHeartbeat);
   /**
    * 通过消息队列传输，利用stomp协议
    */
//        registry
//                .enableStompBrokerRelay("/exchange", "/topic", "/queue", "/amq/queue")
//                .setRelayHost("127.0.0.1")
//                .setClientLogin("guest").setClientPasscode("guest")
//                .setSystemLogin("guest")
//                .setSystemPasscode("guest")
//                .setUserRegistryBroadcast("/topic/refreshLocalUser")
//                .setSystemHeartbeatSendInterval(5000)
//                .setSystemHeartbeatReceiveInterval(4000);
   // prefix for send
   registry.setApplicationDestinationPrefixes("/ws");
}

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

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

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

流程梳理

核心任务类和启动逻辑

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

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

org.springframework.messaging.simp.broker.SimpleBrokerMessageHandler

• org.springframework.context.Lifecycle : 该接口会在容器的生命周期执行完成之后被回调start方法
  • org.springframework.messaging.simp.broker.AbstractBrokerMessageHandler.start
    • org.springframework.messaging.simp.broker.SimpleBrokerMessageHandler.startInternal # 判断getTaskScheduler是否为空，不为空则进行心跳调度

    最终调度的类就是 HeartbeatTask.

剔除逻辑

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

private class HeartbeatTask implements Runnable {

    @Override
    public void run() {
        long now = System.currentTimeMillis();
        // 遍历所有在线的用户
        for (SessionInfo info : sessions.values()) {
            // 如果最后访问时间间隔超过间隔时间
            if (info.getReadInterval() > 0 && (now - info.getLastReadTime()) > info.getReadInterval()) {
                // 触发DISCONNECT_ACK下线事件
                handleDisconnect(info.getSessiondId(), info.getUser(), null);
            }
            // 定期向客户端发送心跳
            if (info.getWriteInterval() > 0 && (now - info.getLastWriteTime()) > info.getWriteInterval()) {
                SimpMessageHeaderAccessor accessor = SimpMessageHeaderAccessor.create(SimpMessageType.HEARTBEAT);
                accessor.setSessionId(info.getSessiondId());
                accessor.setUser(info.getUser());
                initHeaders(accessor);
                MessageHeaders headers = accessor.getMessageHeaders();
                getClientOutboundChannel().send(MessageBuilder.createMessage(EMPTY_PAYLOAD, headers));
            }
        }
    }
}

private void handleDisconnect(String sessionId, Principal user, Message<?> origMessage) {
    this.sessions.remove(sessionId); // 删除用户管道
    this.subscriptionRegistry.unregisterAllSubscriptions(sessionId); // 取消订阅
    SimpMessageHeaderAccessor accessor = SimpMessageHeaderAccessor.create(SimpMessageType.DISCONNECT_ACK);
    accessor.setSessionId(sessionId);
    accessor.setUser(user);
    if (origMessage != null) {
        accessor.setHeader(SimpMessageHeaderAccessor.DISCONNECT_MESSAGE_HEADER, origMessage);
    }
    initHeaders(accessor);
    Message<byte[]> message = MessageBuilder.createMessage(EMPTY_PAYLOAD, accessor.getMessageHeaders());
    // 发送离开事件
    getClientOutboundChannel().send(message);
}

关于过期时间

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

比如你配置的是5秒间隔，那么当超过15秒还没续上心跳时间的时候才会默认剔除。

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

/* STOMP spec: receiver SHOULD take into account an error margin */
private static final long HEARTBEAT_MULTIPLIER = 3;

public SessionInfo(String sessiondId, Principal user, long[] clientHeartbeat, long[] serverHeartbeat) {
    this.sessiondId = sessiondId;
    this.user = user;
    if (clientHeartbeat != null && serverHeartbeat != null) {
        // 这里取了间隔时间最后还相乘了一下.
        this.readInterval = (clientHeartbeat[0] > 0 && serverHeartbeat[1] > 0 ?
                Math.max(clientHeartbeat[0], serverHeartbeat[1]) * HEARTBEAT_MULTIPLIER : 0);
        this.writeInterval = (clientHeartbeat[1] > 0 && serverHeartbeat[0] > 0 ?
                Math.max(clientHeartbeat[1], serverHeartbeat[0]) : 0);
    }
    else {
        this.readInterval = 0;
        this.writeInterval = 0;
    }
    this.lastReadTime = this.lastWriteTime = System.currentTimeMillis();
}

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

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

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

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