20190805发表springboot整合websocket原生版

Posted 2021-04-25 新华前后端开发

HTTP请求用于我们开发以及用户之间最为广泛的一种协议，在HTTP中我们可以简单的通过浏览器获取到我们需要的内容(页面)。但是他也有他的局限性。今天我们的主角websocket将为展现他的功能

HTTP缺点

• HTTP只能有client发起请求服务端做出响应返回结果。服务端是不能主动向客户端发送信息的。所以有的网站在解决实时性上采用的是页面的定时器功能。简而言之是客户端定时的向服务端发送请求
  这样多多少少的造成资源的浪费。

• HTTP是无记忆的。每次请求服务端是无法了解到客户端之前的行为的，但是我们平时浏览器网站的时候感觉浏览器是知道我们之前做的事情的。这是网站在请求是添加的cookie这些服务端提供的数据。对我们而言我们感觉是有记忆的。实则不然

• HTTP1.1之后采用了短连接、长连接两种方式。HTTP请求的发送每次也需要三次握手机制。所以每次的连接耗费资源。1.1后才一定时间内HTTP其实采用的是长连接，这样可以减少资源的开销

• 上述说道的长连接有人可能有疑问，其实HTTP协议是基于TCP协议开发的。所以自然有长连接的特性。

HTTP websocket区别

• HTTP因为短连接的特性所以是无记忆的。为了解决这个问题每个请求都是由General+Request Head+Request Paylaod+Response Headers组成的。其中Heads就是浏览器需要记住的东西，每次传递来传递去的很是耗费性能。

• websocket由于是长连接特性，一次连接就可以一直的双向通信。从载体来说websocket关注的更少，只需要通信当前需要的信息。历史信息双方都是有的。

websocket原理

使用场景

springboot整合websocket

环境准备

• 在springboot基础上引入websocket的jar

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

• 然后在项目中加入如下的配置

@Configuration
public class WebSocketConfig {
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

• 然后我们就可以编写websocket的接受和发送等事件了。我在次基础上封装了一下，先抽象除一个websocket

public abstract class BaseWebSocket extends  BaseController{
    /**
     * 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
     */
    private int onlineCount = 0;

    /**
     * concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
     */
    public CopyOnWriteArraySet<BaseWebSocket> webSocketSet = new CopyOnWriteArraySet<BaseWebSocket>();

    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    public Session session;

    private Logger log = LoggerFactory.getLogger("BaseWebSocket");


    /**
     * 连接建立成功调用的方法
     * @param session
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("sid") String sid) throws IOException {
        this.session = session;
        //加入set中
        webSocketSet.add(this);
        //在线数加1
        addOnlineCount();
        log.debug("有新连接加入！当前在线人数为" + getOnlineCount());
        //发送信息
        MultiMap multiMap = new MultiMap();
        if (null!=session.getQueryString()&&!"".equals(session.getQueryString())) {
            UrlEncoded.decodeTo(session.getQueryString(), multiMap, "UTF-8");
        }
        sendInfo(defaultMessage(multiMap));
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        //从set中删除
        webSocketSet.remove(this);
        //在线数减1
        subOnlineCount();
        log.info("有一连接关闭！当前在线人数为" + getOnlineCount());
    }

    /**
     * 收到客户端消息后调用的方法
     * @param message 客户端发送过来的消息
     * @param session 缓存
     * @throws IOException
     */
    @OnMessage
    public void onMessage(String message, Session session) throws IOException {
        this.session = session;
        try {
            Map paraMap = (Map) JSONObject.parse(message);
            handlerMessage(paraMap);
        } catch (JSONException e) {
            MultiMap multiMap = new MultiMap();
            UrlEncoded.decodeTo(message, multiMap, "UTF-8");
            handlerMessage(multiMap);
            //throw new BusinessException("传递消息格式错误(Json)");
        }
    }

    /**
     * 处理消息接受
     * @param paraMap 接受到map类型的参数
     */
    public void handlerMessage(Map paraMap) {
        try {
            sendInfo(defaultMessage(paraMap));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public Object defaultMessage(Map<String, Object> paraMap) {
        Object obj = new Object();
        try {
            obj = defauleMessage(paraMap);
        } catch (BusinessException e) {
            return formatReturnAppData(e.getMessage());
        }
        return obj;
    }
    /**
     * 默认的发送数据
     * @param paraMap 连接时传递的参数
     * @return
     */
    public abstract Object defauleMessage(Map<String, Object> paraMap);

    public static boolean isJson(String content) {
        try {
            JSONObject.parse(content);
            return true;
        } catch (Exception e) {
            return false;
        }
    }
    /**
     * 发生错误时调用
     @OnError
     **/
    public void onError(Session session, Throwable error) {
        log.error("onMessage方法异常"+error.toString());
        error.printStackTrace();
    }


    /**
     * 发送消息需注意方法加锁synchronized，避免阻塞报错
     * 注意session.getBasicRemote()与session.getAsyncRemote()的区别
     * @param message
     * @throws IOException
     */
    public synchronized void sendMessage(Object message) throws IOException {
//         this.session.getBasicRemote().sendText(message);
        this.session.getAsyncRemote().sendText(JSONObject.toJSONString(message));
    }


    /**
     * 群发自定义消息
     * */
    public void sendInfo(Object message) throws IOException {
        for (BaseWebSocket item : webSocketSet) {
            try {
                item.sendMessage(message);
            } catch (IOException e) {
                continue;
            }
        }
    }

    public  synchronized int getOnlineCount() {
        return onlineCount;
    }

    public  synchronized void addOnlineCount() {
        onlineCount++;
    }

    public  synchronized void subOnlineCount() {
        onlineCount--;
    }
}

• 然后我们新增一个websocket的时候我们只需要继承一下这个抽象的websocket的类然后重现里面的defauleMessage方法就行了。在类上需要加上如下注解

@ServerEndpoint(value = "/accident/getAccident")
@Component
public class AccidentGetAccident extends BaseWebSocket {

    AccidentController accidentController;


    @Override
    public Object defauleMessage(Map<String, Object> paraMap) {
        accidentController = ContextUtil.getApplicationContext().getBean(AccidentController.class);
        return accidentController.getAccident(paraMap);
    }
}

客户端连接

ws = new WebSocket(wsUrl);

• 然后我们就可以重写ws里面的一些事件了

ws.onclose = function () {
    console.log('链接关闭');
};
ws.onerror = function() {
    console.log('发生异常了');
};
ws.onopen = function () {
    console.log('新建连接');
};
ws.onmessage = function (event) {
    console.log("接收到服务端反馈的信息了");
}

• 但是我们得考虑一种情况就是，我们的服务端因为某种因素造成服务宕机，这个时候客户端捕获到onclose事件，这次的连接就会结束，但是服务端可能在短时间内抢修好了。这个时候我们要客户端进行宠幸刷新才会进行重连。websocket正常都是用在大屏的时候，有时认为刷新并不是很方便，所以这个时候需要我们的客户端有重连机制。

var lockReconnect = false;//避免重复连接
    var wsUrl = "ws://127.0.0.1:8088/accident/getAccident?entId=zhonghuaxingzhong";
    var ws;
    var tt;
    function createWebSocket() {
        try {
            ws = new WebSocket(wsUrl);
            init();
        } catch(e) {
            console.log(e+'catch');
            reconnect(wsUrl);
        }
    }
    function init() {
        ws.onclose = function () {
            console.log('链接关闭');
            reconnect(wsUrl);
        };
        ws.onerror = function() {
            console.log('发生异常了');
            reconnect(wsUrl);
        };
        ws.onopen = function () {
            //心跳检测重置
            heartCheck.start();
        };
        ws.onmessage = function (event) {
            setMessageInnerhtml(event.data);
            //拿到任何消息都说明当前连接是正常的
            console.log('接收到消息');
            heartCheck.start();
        }
    }
    function reconnect(url) {
        if(lockReconnect) {
            return;
        };
        lockReconnect = true;
        //没连接上会一直重连，设置延迟避免请求过多
        tt && clearTimeout(tt);
        tt = setTimeout(function () {
            createWebSocket(url);
            lockReconnect = false;
        }, 4000);
    }
    //心跳检测
    var heartCheck = {
        timeout: 3000,
        timeoutObj: null,
        serverTimeoutObj: null,
        start: function(){

        }
    }
    //将消息显示在网页上
    function setMessageInnerHTML(innerHTML) {
        document.getElementById('message').innerHTML += innerHTML + '<br/>';
    }
    createWebSocket(wsUrl);

加入战队

加入战队

微信公众号

微信公众号