使用webSocket实现对数据的实时推送详解

1.什么是webSocket?

相对于 HTTP 这种非持久的协议来说,websocket是 HTML5 出的一个持久化的协议。

2.实时推送数据的实现方式以及应用场景

实现方式
1.轮询:客户端通过代码定时向服务器发送AJAX请求,服务器接收请求并返回响应信息。
优点:代码相对简单,适用于小型应用。
缺点:在服务器数据没有更新时,会造成请求重复数据,请求无用,浪费带宽和服务器资源。

2.长连接:在页面中嵌入一个隐藏的iframe,将这个隐藏的iframe的属性设置为一个长连接的请求或者xrh请求,服务器通过这种方式往客户端输入数据。
优点:数据实时刷新,请求不会浪费,管理较简洁。
缺点:长时间维护保持一个长连接会增加服务器开销。

3.webSocket:websocket是HTML5开始提供的一种客户端与服务器之间进行通讯的网络技术,通过这种方式可以实现客户端和服务器的长连接,双向实时通讯。
优点:减少资源消耗;实时推送不用等待客户端的请求;减少通信量;
缺点:少部分浏览器不支持，不同浏览器支持的程度和方式都不同

应用场景:聊天室、智慧大屏、消息提醒、股票k线图监控等。

3.代码详解

(1)使用Visual Studio开发工具创建控制台项目,(这里只做数据实时推送演示效果,中间的实际业务代码可根据需求场景来开发)

(2)使用NuGet包管理器搜索Fleck包,并安装到我们上一步所创建的项目。

(3)在Main方法下面贴上如下代码,有注释

Console.WriteLine("(1)输入1更新图表数据;");
Console.WriteLine("(2)输入exit退出系统");
//先使用Nuget导入Fleck包 版本要求.net Framework 4.5及以上
FleckLog.Level = LogLevel.Debug;
var allSockets = new List<IWebSocketConnection>();
//监听所有的的地址
var server = new WebSocketServer("ws://0.0.0.0:30000");
server.Start(socket =>
{
    // onopen：当WebSocket建立网络连接的时候触发该事件
    socket.OnOpen = () =>
    {
        Console.WriteLine("连接已打开!");
        allSockets.Add(socket);
    };
    // onclose：当WebSocket被关闭的时候触发该事件
    socket.OnClose = () =>
    {
        Console.WriteLine("连接已关闭!");
        allSockets.Remove(socket);
    };
    // onmessage：当WebSocket接受到远程服务器的数据的时候触发该事件
    socket.OnMessage = message =>
    {
        Console.WriteLine(message);
        allSockets.ToList().ForEach(s => s.Send("路人甲: " + message));
    };
});
var input = Console.ReadLine();
while (input != "exit")
{
    foreach (var socket in allSockets.ToList())
    {
        //send()：向远程服务器发送数据
        //socket.Send("路人乙:"+input);
        if (input == "1") {
            string jsonData = @"{""teamData"":[25,33,15,0,5,10,20],""salesData"":[35,25,18,0,32,22,13]}";
            socket.Send(jsonData);
            Console.WriteLine("发送成功");
        }
    }
    input = Console.ReadLine();
}

(4)构建前端页面,在前端JS配置代码如下,可以在ws.onmessage事件中获取服务器实时推送的数据,可根据业务场景设计页面样式,我这里模拟的是一个智慧大屏的实时推送。

<script type="text/javascript">
    var start = function () {
        var wsImpl = window.WebSocket || window.MozWebSocket;
        window.ws = new wsImpl('ws://localhost:30000/');
		//接收到消息的回调方法
        ws.onmessage = function (evt) {
        //evt.data为后台推送数据
            console.log(evt.data);
        };
        //连接成功建立的回调方法
        ws.onopen = function () {
            //连接已打开
        };
        //连接关闭的回调方法
        ws.onclose = function () {
            //连接已关闭
        }
    }
    window.onload = start;
</script>

4.小结

从上面的案例可以看到,websocket采取的方式是让客户端连接服务器,只要两端进行连接之后,就可以避免多次请求直接发送端对端的数据,不需要经过第三方的转发,只需要客户端和浏览器通过http协议进行一个握手的动作,然后单独建立一条TCP的通信通道进行数据的传送。

文章案例由于前端页面代码太多,就没有全部贴到文章里,有需要的可以在下面留言

业务场景是这样的，公司要将业务员的业绩实时公布在大屏上，于是我就做了一套大屏页面，现在只剩实时更新数据了；vue前端也可以不断轮询请求数据，但是这样你打开F12会发现很多网络请求，很不好看，也不安全，于是我选择了websocket来实时推送

引入websocket的依赖

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

向spring注入websocket的bean

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
//注意注解
@EnableWebSocket
@Configuration
public class WebSocketConfig {
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

开始写websocket的接收推送消息的方法就行了

import com.alibaba.fastjson.JSONArray;
import com.zbkj.crmeb.tool.service.DateScricpService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
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 java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * @author zihao
 *
 * @date 2022-02-20
 */
@Controller
@ServerEndpoint("/websocket/{tableId}")
public class WebSocket {

    @Autowired
    private DateScricpService dateScricpService;

    private Session session;
    private String uid;
    public static CopyOnWriteArraySet<WebSocket> webSockets =new CopyOnWriteArraySet<>();
    private static Map<String,Session> sessionPool = new HashMap<String,Session>();
    //session集合
    private static List<Map<String,Object>> sessionList = new ArrayList<>();
    //信息集合
    private static List<Map<String,String>> MessageList = new ArrayList<>();

    @OnOpen
    public void onOpen(Session session, @PathParam(value="tableId")String code) {
        Map<String,Object> map = new HashMap<String,Object>();
        this.session = session;
        this.uid = code;
        webSockets.add(this);
        sessionPool.put(code, session);
        map.put("uid", code);
        map.put("session", session);
        sessionList.add(map);
        // Constants.WEBSOCKET = true;//定义常量  是否开启websocket连接
        System.out.println("【websocket消息】有新的连接，总数为:"+webSockets.size());
    }

    @OnClose
    public void onClose() {
        webSockets.remove(this);
        //Constants.WEBSOCKET = false;
        System.out.println("【websocket消息】连接断开，总数为:"+webSockets.size());
        for (int i = 0; i < MessageList.size(); i++) {
            if(this.uid.equals(MessageList.get(i).get("uid"))){
                MessageList.remove(i);
                break;
            }
        }
        for (int i = 0; i < sessionList.size(); i++) {
            if(this.uid.equals(sessionList.get(i).get("uid"))){
                sessionList.remove(i);
                break;
            }
        }

    }

    @OnMessage
    public void onMessage(String message) {
        System.out.println("【websocket消息】收到客户端消息:"+message);
        if(MessageList != null && MessageList.size() > 0){
            for (Map<String, String> messageMap : MessageList) {
                if(this.uid.equals(messageMap.get("uid"))){
                    messageMap.put("message",message);
                    return;
                }
            }
        }
        Map<String,String> map = new HashMap<String,String>();
        map.put("message",message);
        map.put("uid",this.uid);
        MessageList.add(map);
    }

    // 此为广播消息
    public void sendAllMessage(String message) {
        for(WebSocket webSocket : webSockets) {
            try {
                webSocket.session.getAsyncRemote().sendText(message);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    // 此为单点消息
    public void sendOneMessage(String code, String message) {
        Session session = sessionPool.get(code);
        /*在发送数据之前先确认 session是否已经打开 使用session.isOpen() 为true 则发送消息
         * 不然会报错:The WebSocket session [0] has been closed and no method (apart from close()) may be called on a closed session */
        if (session != null && session.isOpen()) {
            try {
                session.getAsyncRemote().sendText(message);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    // 此为对应的广播消息
    public void sendAllMessageBySelf() {
        //查询大屏的基础数据
        Map<String, String> TotalDate = dateScricpService.selectTotalDate();
        //查询客户概况的基础数据
        Map<String, Object> customerDate = dateScricpService.selectCustomerDate();
        //查询收入概况的基础数据
        Map<String, Object> receiveDate = dateScricpService.selectReceiveDate();
        //查询月收入概况的基础数据
        Map<String, Object> monthReceiveMoneyY = dateScricpService.selectMonthReceiveMoneyY();
        //查询业务分配数据
        Map<String, Object> businessDistribute = dateScricpService.selectBusinessDistribute();
        if(MessageList != null && MessageList.size() > 0){
            for (Map<String, String> messageMap : MessageList) {
                if(sessionList != null && sessionList.size() > 0){
                    for (Map<String, Object> sessionMap : sessionList) {
                        if(String.valueOf(sessionMap.get("uid")).equals(messageMap.get("uid"))){
                            Map<String,Object>  message = (Map<String, Object>)JSONArray.parse(messageMap.get("message"));
                            int employeeNum = Integer.parseInt(String.valueOf(message.get("employeeNum")));
                            //查询员工概况的基础数据
                            List<Map<String, String>> employeeDateList = dateScricpService.selectEmployeeDate(String.valueOf(employeeNum));
                            //查询月平均贡献数据
                            Map<String, Object> avageMonth = dateScricpService.selectAverageMonthReceiveMoneyY(String.valueOf(message.get("avageMonth")));
                            //查询业务员分配数据
                            Map<String, Object> EmployeeInfo = dateScricpService.selectEmployeeInfo(String.valueOf(message.get("employeeDate")));
                            //返回的数据
                            Map<String,Object> returnDate = new HashMap<>();
                            returnDate.put("TotalDate",TotalDate);
                            returnDate.put("customerDate",customerDate);
                            returnDate.put("receiveDate",receiveDate);
                            returnDate.put("monthReceiveMoneyY",monthReceiveMoneyY);
                            returnDate.put("businessDistribute",businessDistribute);
                            returnDate.put("employeeDateList",employeeDateList);
                            returnDate.put("avageMonth",avageMonth);
                            returnDate.put("EmployeeInfo",EmployeeInfo);
                            String returnStr = JSONArray.toJSONString(returnDate);
                            //向指定客户端推送信息
                            Session session = (Session)sessionMap.get("session");
                            session.getAsyncRemote().sendText(returnStr);
                        }
                    }
                }
            }
        }
    }
}

这里的sendAllMessageBySelf方法是我混合业务写的特定推送消息，借鉴即可。具体思路是每个客户端通过onOpen方法进来，这时候我同uid和session来存取对应客户端的标志和session对象，然后将这两个参数放到map对象里再放到客户端集合sessionList里，方便群发针对性的消息；MessageList是存放对应客户端的消息集合，具体看onMessage方法即可，写完之后我们就可以测试一下是否连通。让后端多久推送一次，自己写个定时任务就行。

import com.utils.DateUtil;
import com.zbkj.crmeb.tool.service.DateScricpService;
import com.zbkj.crmeb.tool.utils.WebSocket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

/**
 * websocket的定时推送
 *
 * @author zihao
 *
 * @date 2022-02-20
 */
@Component
@Configuration //读取配置
@EnableScheduling // 2.开启定时任务
public class WebsocketTask {
    //日志
    private static final Logger logger = LoggerFactory.getLogger(WebsocketTask.class);

    @Autowired
    private DateScricpService dateScricpService;
    @Autowired
    private WebSocket webSocket;

    @Scheduled(fixedDelay = 1000 * 5L) //5秒钟同步一次数据
    public void init(){
        logger.info("---开始推送数据 - {}", DateUtil.nowDateTime());
        try {
            //调用websocket对应客户端推送数据推送消息
            webSocket.sendAllMessageBySelf();
        }catch (Exception e){
            e.printStackTrace();
            logger.error("websocket.task" + " | msg : " + e.getMessage());
        }

    }
}

测试本地websocket是否连通

用http://ws.douqq.com/地址测一下就行
没问题了我们就可以配置Nginx了，毕竟还是要发到服务器上的

map $http_upgrade $connection_upgrade { 
	 default upgrade;
	 '' close;
	}

server {
        listen       888 ssl;
        server_name  域名;
		
		ssl_certificate      证书路径;
        ssl_certificate_key  证书路径;

        ssl_session_cache    shared:SSL:1m;
        ssl_session_timeout  5m;

        ssl_ciphers  HIGH:!aNULL:!MD5;
        ssl_prefer_server_ciphers  on;
 
		location ~/websocket/ {
			proxy_pass https://127.0.0.1:8081;
			proxy_set_header Host $host;
			proxy_set_header X-Real-IP $remote_addr;
			proxy_set_header X-Forwarded-For     
			$proxy_add_x_forwarded_for;
			proxy_http_version 1.1;
			proxy_set_header Upgrade $http_upgrade;
			proxy_set_header Connection $connection_upgrade;
       }
 
        error_page   500 502 503 504  /50x.html;
        location = /50x.html {
            root   html;
        }
    }

友情提示不要忘了在服务器上把888端口放开，配完了再用刚才那个地址测一下，测通了我们就开始写前端了

前端vue

created(){
    this.initWebSocket();
},
destroyed() {
    this.websocketclose();
}, 
methods: {
	//初始化websocket
    initWebSocket() {
      var uid = (((1+Math.random())*0x10000)|0).toString(16);
      var uid1 = (((1+Math.random())*0x10000)|0).toString(16);
      var uid2 = (((1+Math.random())*0x10000)|0).toString(16);
      this.uid = uid+uid1+uid2;
      //debugger;
      // WebSocket与普通的请求所用协议有所不同，ws等同于http，wss等同于https   key你自定义的key
      //var host = window.location.host
      // this.websock = new WebSocket("wss://ip:888/websocket/"+this.uid);
      this.websock = new WebSocket("ws://127.0.0.1:8081/websocket/"+this.uid);
      this.websock.onopen = this.websocketonopen;
      this.websock.onerror = this.websocketonerror;
      this.websock.onmessage = this.websocketonmessage;
      this.websock.onclose = this.websocketclose;
    },
    websocketonopen() {
      this.websocketsend();
    },
    websocketonerror(e) {
      console.log("error")
      console.log(e)
    },
    websocketonmessage(e) {
      //JSON.parse(e.data); //这个是收到后端主动推送的值
      var receiveMessage = JSON.parse(e.data);
      console.log("message")
      console.log(receiveMessage)
    },
    websocketsend(){
      let data = {employeeNum: this.subTabIndex,employeeDate: this.monthDate,avageMonth: this.subTabMonthIndex};
      //数据发送
      this.websock.send(JSON.stringify(data));
    },
    websocketclose(e) {
      console.log("close")
      console.log(e)
    },
} 

如果是https就用wss来链接websocket，我服务器是https，所以配的证书用的wss。到这已经基本搞完了，看看成果吧

从图上可以看出来，那个50，我后台录入了200的订单，已经实时更新成250了，成功了