物联网架构成长之路(58)-用户与应用服务器之WebSocket通信

一、前言

　　之前的物联网篇，有介绍过用户手机APP或在H5端，可以作为mqtt的客户端，通过tcp:1883或者Websocket:8083，连接到EMQ Broker上，订阅设备的主题，从而收到设备上报的数据。但是，在处理工业物联网的时候，会出现，物联网平台会对接客户自己的应用。设备上报与下发都是通过物联网平台与客户的应用服务器进行通信。客户应用服务器转发物联网平台，设备上报的数据到客户应用下自己的用户终端（手机）。这个时候，就需要建立起客户应用服务器与用户的通道了。传统上，可以使用http轮询。也可以使用mqtt类似的进行通信。但是这里就介绍通过SpringBoot 简单搭建一个WebSocket服务。建立起客户应用服务器与用户的通信。

二、简单流程

　　物联网平台只做设备接入、设备管理、客户接入、客户管理。客户自己的应用与终端客户，则又客户自行开发。从而进行解耦，这个流程方式与功能划分，与阿里的物联网平台类似。后续我们也会基于物联网平台开发一些标准化的应用。

三、后端代码

　　3.1 pom.xml

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

　　3.2 config配置

　　WebSocketConfig.java

package com.wunaozai.demo.websocket.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

@Configuration
public class WebSocketConfig {

    /**
     * 自动注入Websocket Bean
     * @return
     */
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

　　GetHttpHeaderConfig.java

package com.wunaozai.demo.websocket.config;

import java.util.List;
import java.util.Map;

import javax.websocket.HandshakeResponse;
import javax.websocket.server.HandshakeRequest;
import javax.websocket.server.ServerEndpointConfig;
import javax.websocket.server.ServerEndpointConfig.Configurator;

import org.springframework.context.annotation.Bean;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

/**
 * 将请求中的Header设置到属性中
 * @author wunaozai
 * @Date 2020-10-27
 */
public class GetHttpHeaderConfigurator extends Configurator {

    @Override
    public void modifyHandshake(ServerEndpointConfig config, HandshakeRequest request, HandshakeResponse response) {
        System.out.println("modifyHandshake 协议升级101");
        Map<String, List<String>> headers = request.getHeaders();
        config.getUserProperties().put("headers", headers);
        super.modifyHandshake(config, request, response);
    }
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

　　3.3 service 服务层

　　WebSocketService.java

package com.wunaozai.demo.websocket.service;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import javax.websocket.EndpointConfig;
import javax.websocket.OnClose;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;

import org.springframework.stereotype.Component;

import com.google.gson.Gson;
import com.wunaozai.demo.websocket.config.GetHttpHeaderConfigurator;

/**
 * 服务层
 * 注意: 这里的ServerEndpoint不会进入拦截器
 * @author wunaozai
 * @Date 2020-10-27
 */
@ServerEndpoint(value="/websocket/{token}", configurator=GetHttpHeaderConfigurator.class)
@Component
public class WebSocketService {

    //用来存放每个客户端对应的WebSocketServer对象
    private static ConcurrentHashMap<String, Session> sessionPools = new ConcurrentHashMap<>();

    /**
     * 成功建立连接
     * @param session
     * @param token
     */
    @OnOpen
    public void onOpen(Session session, @PathParam(value="token") String token, EndpointConfig config) {
        try {
            Map<String, List<String>> headers = 
                    (Map<String, List<String>>) config.getUserProperties().get("headers");
            System.out.println(new Gson().toJson(headers)); //这里可以通过Headers或者token进行认证
            if(!checkToken(token)) {
                //校验失败，禁止建立连接
                session.close();    
            }
            System.out.println("建立连接: " + token);
            sessionPools.put(token, session);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    @OnClose
    public void onClose(@PathParam(value="token") String token) {
        sessionPools.remove(token);
        System.out.println("断开连接: " + token);
    }
    
    @OnMessage
    public void onMessage(@PathParam(value="token")String token, String message) throws IOException {
        System.out.println("收到客户端发来信息: " + message);
    }
    
    //发送广播信息
    public void sendBroadCast(String message) throws IOException {
        //这里从第三方服务或调用该接口向所有客户端进行广播信息
        for(Session session: sessionPools.values()) {
            sendMessage(session, "{"msg":""+message+""}");
        }
    }
    //指定单个发送数据
    public void sendMessage(Session session, String message) throws IOException {
        if(session != null && session.isOpen()) {
            synchronized (session) {
                session.getBasicRemote().sendText(message);
            }
        }
    }
    
    public String getToken() {
        //这里随机生成Token，并缓存到Redis，设置10分钟过期策略
        return "access_token_random_qcb0a6S";
    }
    public boolean checkToken(String token) {
        if("access_token_random_qcb0a6S".equals(token)) {
            return true;
        }
        return false;
    }
}

　　3.4 controller 控制器

　　WebSocketController.java

package com.wunaozai.demo.websocket.controller;

import java.io.IOException;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import com.wunaozai.demo.websocket.service.WebSocketService;

@Controller
@RequestMapping(value="/websocket")
public class WebSocketController {

    @Autowired
    private WebSocketService websocketService;
    
    @RequestMapping(value="index")
    public String index() {
        return "websocket/index";
    }
    
    @ResponseBody
    @RequestMapping(value="send")
    public String sendBroadCast(String msg) throws IOException {
        websocketService.sendBroadCast(msg);
        return "ok.";
    }
    @ResponseBody
    @RequestMapping(value="token")
    public String getToken() {
        //这里可以判断请求是否合法，查询数据库
        //如果在拦截器已经做处理的，就不需要进行判断
        return websocketService.getToken();
    }
    
}

　　3.5 目录结构

　　

四、前端代码

<!DOCTYPE html>
<html lang="en">
  <head>
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no">
    <meta name="description" content="">
    <meta name="author" content="">
    <title>Websocket 例子</title>
  </head>
  <body>
  <button id="btn-token">获取Token</button>
  <input id="token" placeholder="请填写Token"/>
  <br>
  <button id="btn-connect">连接WebSocket</button>
  <input id="message" value="Hello" placeholder="请填写要发送的信息"  />
  <button id="btn-send">发送信息</button>
  </body>
  
  <script type="text/javascript" src="http://libs.baidu.com/jquery/2.0.0/jquery.min.js"></script>
  <script type="text/javascript">
    jQuery(function(){
        initEvent();
    });
    
    function initEvent(){
        $("#btn-token").bind('click', function(){
            $.post('/websocket/token', function(ret){
                console.log(ret);
                $("#token").val(ret);
            });
        });
        $("#btn-connect").bind('click', function(){
            openSocket();
        });
        $("#btn-send").bind('click', function(){
            sendMessage();
        });
    }
    var socket;
    function openSocket() {
        if(typeof(WebSocket) == "undefined") {
            console.log("您的浏览器不支持WebSocket");
        }else{
            console.log("您的浏览器支持WebSocket");
            //实现化WebSocket对象，指定要连接的服务器地址与端口  建立连接
            var token = $("#token").val();
            var socketUrl="ws://127.0.0.1:8080/websocket/"+token;
            console.log(socketUrl);
            if(socket!=null){
                socket.close();
                socket=null;
            }
            socket = new WebSocket(socketUrl);
            //打开事件
            socket.onopen = function() {
                console.log("websocket已打开");
                //socket.send("这是来自客户端的消息" + location.href + new Date());
            };
            //获得消息事件
            socket.onmessage = function(msg) {
                var serverMsg = "收到服务端信息：" + msg.data;
                console.log(serverMsg);
                //发现消息进入    开始处理前端触发逻辑
            };
            //关闭事件
            socket.onclose = function() {
                console.log("websocket已关闭");
            };
            //发生了错误事件
            socket.onerror = function() {
                console.log("websocket发生了错误");
            }
        }
    }
    function sendMessage() {
        if(typeof(WebSocket) == "undefined") {
            console.log("您的浏览器不支持WebSocket");
        }else {
            // console.log("您的浏览器支持WebSocket");
            var message = $("#message").val();
            var msg = '{"contentText":"'+message+'"}';
            console.log(msg);
            socket.send(msg);
        }
    }
  </script>
</html>

五、运行效果

1. 访问页面，然后请求获取Token，获取到【access_token_random_qcb0a6S】,然后拼接EndPoint，ws://127.0.0.1:8080/websocket/access_token_random_qcb0a6S

2. 服务器执行协议升级握手协议101，并将请求里面的Headers设置到当前请求的Session中

3. 完成协议升级。由Http协议转WebSocket协议。进入WebSocketService的OnOpen方法。

4. OnOpen方法，判断当前ws连接的token或者headers里面是否包含认证信息（比如jwt或oauth、自定义token等方式）

5. 完成连接建立，将ws连接session放到session池里面，以供下一步广播信息给ws客户端

6. ws客户发送Hello到服务器，服务器可以做数据处理应答等业务操作

7. 通过请求 http://127.0.0.1:8080/websocket/send?msg=Hi 模拟【物联网平台】通过rpc或者amqp等远程调用协议，调用发送设备上报的信息到ws客户端

8.1 假如广播设备上报信息给所有客户端，那么将从session池中遍历所有ws客户端，然后依次发送数据。

8.2 假如是单个发送数据。那么需要根据设备的ID信息，查询设备与用户的绑定信息，再转发数据给ws客户端（用户）

9. 客户端（手机APP、H5等）通过弹框等方式，提醒用户

参考资料:

          https://www.cnblogs.com/JohanChan/p/12522001.html

本文地址：https://www.cnblogs.com/wunaozai/p/13889216.html
本系列目录： https://www.cnblogs.com/wunaozai/p/8067577.html
个人主页：https://www.wunaozai.com/