WebSocket原理与实践

时间  2019-12-12
标签 websocket 原理 实践 繁體版
原文   原文链接

开题思考：如何实现客户端及时获取服务端数据？

Polling

指客户端每隔一段时间（周期性）请求服务端获取数据，可能有更新数据返回，也可能什么都没有，它并不在意服务端数据有无更新。（Web端通常采用ajax polling实现）javascript

Long Polling

阻塞型Polling，和Polling不一样的是假如服务端数据没有准备好，那么可能会hold住请求，直到服务端有相关数据，或者等待必定时间超时才会返回。html

WebSocket

HTML5 WebSocket规范定义了一种API，使Web页面可以使用WebSocket协议与远程主机进行双向通讯。与轮询和长轮询相比，巨大减小了没必要要的网络流量和等待时间。

Websocket体系结构

前端

Websocket协议

WebSocket协议被设计成与现有的Web基础结构很好地工做。该协议规范定义了HTTP链接做为WebSocket链接生命的开始，从Http协议转换成WebSocket，被称为WebSocket握手。

  • 浏览器向服务器发送请求，表示它但愿将协议从HTTP切换到WebSocket。客户端经过升级报头表达其愿望：java

    GET /chat HTTP/1.1 Host: server.example.com Upgrade: websocket Connection: Upgrade Sec-WebSocket-Key: x3JJHMbDL1EzLkh9GBhXDw== Sec-WebSocket-Protocol: chat, superchat Sec-WebSocket-Version: 13 Origin: http://example.com

  • 从上面的报文能够看到，和HTTP协议的请求中，多了几样东西，核心就是Upgrade和Connection两个参数，用来告诉服务器，我须要升级为Websocket：jquery

    Upgrade: websocket Connection: Upgrade Sec-WebSocket-Key: x3JJHMbDL1EzLkh9GBhXDw== Sec-WebSocket-Protocol: chat, superchat Sec-WebSocket-Version: 13

  • 若是服务端可以理解WebSocket协议，它赞成以Upgrade头字段来升级协议，会响应如下信息：nginx

    HTTP/1.1 101 Switching Protocols Upgrade: websocket Connection: Upgrade Sec-WebSocket-Accept:HSmrc0sMlYUkAGmm5OPpG2HaGWk= Sec-WebSocket-Protocol: chat

  • 此时，HTTP链接中断，并由同一底层TCP/IP链接上的WebSocket链接替换。 默认状况下，WebSocket链接使用与HTTP（80）和HTTPS（443）相同的端口。git

Spring-WebSocket实战

Spring框架提供了WebSocket支持，很容易实现相关功能，此处分享一下使用Spring集成WebSocket实现简单的多人会议系统。

服务端相关代码

  • MeetingController （很简单的一个入口，建立会议，并生成会议id和对应随机串）web

    @Controller
public class MeetingController {

    private static AtomicInteger id = new AtomicInteger(0);

    @RequestMapping(value = "/meeting", method = RequestMethod.POST)
    @ResponseBody
    public Map<String, Object> createMeeting() {
        int meetingId = id.incrementAndGet();
        String randStr = RandomStringUtils.random(6, true, true);
        SystemCache.idRandStrMap.put(meetingId, randStr);
        Map<String, Object> meetingVO = new HashMap<>();
        meetingVO.put("id", meetingId);
        meetingVO.put("randStr", randStr);
        return meetingVO;
    }
}

  • WebSocketConfig （经过WebSocketConfigurer来配置定义本身的Websocket处理器和拦截器）ajax

    @Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        /**
         * 注册websocket处理器以及拦截器
         */
        registry.addHandler(meetingWebSocketHandler(), "/websocket/spring/meeting").addInterceptors(myInterceptor());
    }

    @Bean
    public MeetingWebSocketHandler meetingWebSocketHandler() {
        return new MeetingWebSocketHandler();
    }

    @Bean
    public WebSocketHandshakeInterceptor myInterceptor() {
        return new WebSocketHandshakeInterceptor();
    }
}

  • WebSocketHandshakeInterceptor （握手拦截器，用于处理请求携带参数）redis

    public class WebSocketHandshakeInterceptor implements HandshakeInterceptor {

    @Override
    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler,
            Map<String, Object> attributes) throws Exception {
        if (request instanceof ServletServerHttpRequest) {
            ServletServerHttpRequest serverHttpRequest = (ServletServerHttpRequest) request;
            String randStr = serverHttpRequest.getServletRequest().getParameter("randStr");
            String role = serverHttpRequest.getServletRequest().getParameter("role");
            if (StringUtils.isNotBlank(randStr)) {
                attributes.put("randStr", randStr);
            }
            if (StringUtils.isNotBlank(role)) {
                attributes.put("role", role);
            }
        }
        return true;
    }

   @Override
   public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler,
        Exception exception) {
   }
}

  • MeetingWebSocketHandler（websocket处理器，用于接受客户端发送各类类型数据，主要分为数据帧和控制帧）

    @Service
public class MeetingWebSocketHandler extends TextWebSocketHandler {

    private static final Log LOG = LogFactory.getLog(MeetingWebSocketHandler.class);
    // 会议id和wsSession列表
    private static final ConcurrentHashMap<Integer, CopyOnWriteArraySet<WebSocketSession>> meetingWsSeesionMap = new ConcurrentHashMap<>();

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        LOG.info("spring websocket成功创建链接...");
        int meetingId = getMeetingId(session);
        if (meetingId <= 0) {
            singleMessage(session, new TextMessage("会议不存在!"));
            session.close();
        }
        // 若是该会议已存在，则直接加入
        if (meetingWsSeesionMap.containsKey(meetingId)) {
            CopyOnWriteArraySet<WebSocketSession> webSocketSessions = meetingWsSeesionMap.get(meetingId);
            webSocketSessions.add(session);
        }
        // 若是不存在，则新建
        else {
            CopyOnWriteArraySet<WebSocketSession> webSocketSessions = new CopyOnWriteArraySet<>();
            webSocketSessions.add(session);
            meetingWsSeesionMap.put(meetingId, webSocketSessions);
        }
    }

    @Override
    public void handleTextMessage(WebSocketSession session, TextMessage message) {
        if (!session.isOpen())
            return;
        LOG.info(message.getPayload());
        int meetingId = getMeetingId(session);
        TextMessage wsMessage = new TextMessage(message.getPayload());
        broadcastMessage(meetingId, wsMessage);
    }

    /**
     * 发送信息给指定用户
     * @param clientId
     * @param message
     * @return
     */
    public void singleMessage(WebSocketSession session, TextMessage message) {
        if (!session.isOpen())
            return;
        try {
            session.sendMessage(message);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 广播信息
     * @param message
     * @return
     */
    public void broadcastMessage(int meetingId, TextMessage message) {
        // 获取会议全部的wsSession
        CopyOnWriteArraySet<WebSocketSession> webSocketSessions = meetingWsSeesionMap.get(meetingId);
        for (WebSocketSession session : webSocketSessions) {
            try {
                if (session.isOpen()) {
                    session.sendMessage(message);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        if (session.isOpen()) {
            session.close();
        }
        LOG.info("链接出错");
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        LOG.info("链接已关闭：" + status);
        int meetingId = getMeetingId(session);
        // role 1为主持人
        String role = String.valueOf(session.getAttributes().get("role"));
        // 若是是主持人，则关闭全部该会议链接
        CopyOnWriteArraySet<WebSocketSession> webSocketSessions = meetingWsSeesionMap.get(meetingId);
        if (StringUtils.equals("1", role)) {
            SystemCache.idRandStrMap.remove(meetingId);
            for (WebSocketSession webSocketSession : webSocketSessions) {
                webSocketSession.close();
            }
            webSocketSessions.remove(meetingId);
        } else {
            webSocketSessions.remove(session);
        }
    }

    @Override
    public boolean supportsPartialMessages() {
        return false;
    }

    private int getMeetingId(WebSocketSession session) {
        String randStr = String.valueOf(session.getAttributes().get("randStr"));
        int meetingId = SystemCache.getMeetingIdByRandStr(randStr);
        return meetingId;
    }
 }

  • SystemCache（系统缓存，集群部署的状况下，可改成redis实现分布式缓存，单机则不须要）

    public class SystemCache {

    // 会议id和随机字符串的映射关系
    public static ConcurrentHashMap<Integer, String> idRandStrMap = new ConcurrentHashMap<>();

    public static int getMeetingIdByRandStr(String randStr) {
        int meetingId = 0;
        for (Map.Entry<Integer, String> entry : idRandStrMap.entrySet()) {
            if (randStr.equals(entry.getValue())) {
                meetingId = entry.getKey();
            }
        }
        return meetingId;
    }
}

前端相关代码

  • meeting-create.html（主持人页面，用于建立会议而且能够发送消息）

    <!DOCTYPE html>
<html>
<head>
<meta http-equiv="Content-Type" content="text/html;charset=utf-8" />
<title>在线会议系统</title>
</head>
<body>
    <h2>欢迎使用会议系统</h2>
    <button id="create" onclick="createMeeting()">建立会议</button>
    <hr />
    <div id="meeting"></div>
    消息内容：
    <input id="text" type="text" />
    <button id="send" disabled="disabled" onclick="send()">发送消息</button>
    <hr />
    <button id="close" onclick="closeWebSocket()">结束会议</button>
    <hr />
    <div id="message"></div>
</body>

<script type="text/javascript" src="js/jquery-1.12.0.js"></script>
<script type="text/javascript">
    var websocket = null;
    var randStr;
    var remote = window.location.host;
    function openWebsocket() {
        //判断当前浏览器是否支持WebSocket
        if ('WebSocket' in window) {
            websocket = new WebSocket("ws://" + window.location.host
                    + "/websocket/spring/meeting?role=1&randStr=" + randStr);

            //链接发生错误的回调方法
            websocket.onerror = function() {
                setMessageInnerHTML("会议链接发生错误!");
            };

            //链接成功创建的回调方法
            websocket.onopen = function() {
                setMessageInnerHTML("会议链接成功...");
                document.getElementById("send").disabled = false;
            }

            //接收到消息的回调方法
            websocket.onmessage = function(event) {
                setMessageInnerHTML(event.data);
            }

            //链接关闭的回调方法
            websocket.onclose = function() {
                setMessageInnerHTML("会议结束，链接关闭!");
                document.getElementById("create").disabled = false;
                document.getElementById("send").disabled = true;
            }

            //监听窗口关闭事件，当窗口关闭时，主动去关闭websocket链接，防止链接还没断开就关闭窗口，server端会抛异常
            window.onbeforeunload = function() {
                closeWebSocket();
            }
        } else {
            alert('当前浏览器 Not support websocket');
        }
    }

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

    //关闭WebSocket链接
    function closeWebSocket() {
        websocket.close();
    }

    //发送消息
    function send() {
        var content = document.getElementById('text').value;
        websocket.send(content);
    }

    function createMeeting() {
        $.post("/meeting", function(data, status) {
            randStr = data.randStr;
            $("#create").after("<p>会议邀请码：" + randStr + "</p>");
            $("#create").attr("disabled", true);
            openWebsocket();
        });
}
</script>
</html>

  • meeting-join.html（观众页面，用于加入会议而且也能够发送消息）

    <!DOCTYPE html>
<html>
<head>
<meta http-equiv="Content-Type" content="text/html;charset=utf-8" />
<title>在线会议系统</title>
</head>
<body>
    <h2>欢迎使用会议系统</h2>
    会议邀请码：
    <input id="randStr" type="text" />
    <button id="open" onclick="openWebsocket()">加入会议</button>
    <hr />
    消息内容：
    <input id="text" type="text" />
    <button id="send" disabled="disabled" onclick="send()">发送消息</button>
    <hr />
    <button id="close" disabled="disabled" onclick="closeWebSocket()">离开会议</button>
    <hr />
    <div id="message"></div>
</body>

<script type="text/javascript">
    var websocket = null;
    var remote = window.location.host;
    function openWebsocket() {
        var randStr = document.getElementById('randStr').value;
        //判断当前浏览器是否支持WebSocket
        if ('WebSocket' in window) {
            websocket = new WebSocket("ws://" + window.location.host
                    + "/websocket/spring/meeting?randStr=" + randStr);

            //链接发生错误的回调方法
            websocket.onerror = function() {
                setMessageInnerHTML("会议链接发生错误!");
            };

            //链接成功创建的回调方法
            websocket.onopen = function() {
                setMessageInnerHTML("会议链接成功...");
                document.getElementById("open").disabled = true;
                document.getElementById("randStr").disabled = true;
                document.getElementById("send").disabled = false;
                document.getElementById("close").disabled = false;
            }

            //接收到消息的回调方法
            websocket.onmessage = function(event) {
                setMessageInnerHTML(event.data);
            }

            //链接关闭的回调方法
            websocket.onclose = function() {
                setMessageInnerHTML("会议结束，链接关闭!");
                document.getElementById("randStr").disabled = false;
                document.getElementById("open").disabled = false;
                document.getElementById("send").disabled = true;
                document.getElementById("close").disabled = true;
            }

            //监听窗口关闭事件，当窗口关闭时，主动去关闭websocket链接，防止链接还没断开就关闭窗口，server端会抛异常
            window.onbeforeunload = function() {
                closeWebSocket();
            }
        } else {
            alert('当前浏览器 Not support websocket');
        }
    }

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

    //关闭WebSocket链接
    function closeWebSocket() {
        websocket.close();
    }

    //发送消息
    function send() {
        var content = document.getElementById('text').value;
        websocket.send(content);
}
</script>
</html>

     

项目演示

  • 访问meeting-create.html进入主持人界面，点击建立会议，生成会议邀请码，并显示会议链接成功，界面以下：
    

  • 访问meeting-join.html进入观众界面，并经过上面的邀请码加入会议，界面以下：
    

  • 此时双方就能够互相发送消息，主持人离开会议，则全部人退出，观众离开，不影响会议进行。

  • 具体代码地址：https://gitee.com/yehx/websocket-meeting

总结

WebSocket做为一个双通道的协议，颠覆了传统的Client请求Server这种单向通道的模式。因为WebSocket的兴起，Web领域的实时推送技术也被普遍使用，能够简单实现让用户不须要刷新浏览器就能够得到实时更新。它有着普遍的应用场景，好比在线聊天室、在线客服系统、评论系统、WebIM等。
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程