时间戳

首页 > 解决方案 > 带有 PubSub 的 Socket.io：多页面刷新后几乎没有实时数据显示

javascript - 带有 PubSub 的 Socket.io：多页面刷新后几乎没有实时数据显示

问题描述

我正在尝试使用 Google PubSub、Node js 和 Google App Engine 可视化流数据。我正在做的只是：

  1. 每 1 秒将消息流推送到 PubSub 主题（使用 python 脚本）
  2. 创建对该主题的订阅
  3. 创建一个简单的 web 应用程序，它监听订阅，解析每个传入的消息，并以实时条形图的形式显示到浏览器（使用 node js、socket.io 和 fusionchart）

我遵循了本教程：https : //www.fusioncharts.com/blog/visualize-real-time-data-socket-io-charts/，唯一的区别是我使用 PubSub 而不是 PubNub，同时遵循此文档：https://cloud.google.com/pubsub/docs/quickstart-client-libraries#pubsub-client-libraries-nodejs用于“接收消息”部分。

它实际上有效，当我每 1 秒发布 10 条消息时，图表将每 1 秒实时显示 10 条消息。这是我运行应用程序时图表的样子：

可视化示例

问题是它仅在我第一次运行应用程序并打开页面时才有效。当我刷新页面并推送另外 10 条新消息时，数据显示需要更长的时间。当它最终出现时，仅显示 10 条消息中的 5 条。如果我尝试刷新同一页面，显示的数据会越来越少，甚至在某些时候都不会出现。

我通过控制台跟踪每条传入和传出的消息，似乎一切正常。推送了 10 条消息，收到了 10 条消息。只是出于某种原因，每次刷新后显示的越来越少。

这是我的 app.js 代码：

var express = require('express');
var app = require('express')();
var http = require('http').Server(app);
//creates a new socket.io instance attached to the http server.
var io = require('socket.io')(http);

// Imports the Google Cloud client library
const PubSub = require('@google-cloud/pubsub');

// Your Google Cloud Platform project ID
const projectId = 'myprojecthere';




//Provide the absolute path to the dist directory.
app.use(express.static(__dirname + '/dist'));

//On get request send 'index.html' page as a response.
app.get('/', function(req, res) {
   res.sendFile(__dirname +'/index.html');
});

//Whenever someone connects this gets executed
//original : connection
io.on('connection', function (socket) {
  console.log(`Enter io connection`);
  console.log(' %s sockets connected', io.engine.clientsCount)

  // Instantiates a client
  const pubsub = new PubSub({
    projectId: projectId,
    key: """censored"""
  });

  var strData;
    /**
     * TODO(developer): Uncomment the following lines to run the sample.
     * https://cloud.google.com/pubsub/docs/pull#pubsub-pull-messages-async-nodejs
     */
    const subscriptionName = 'testing_subscription';
    const topicName = 'testing';
    const timeout = 50;

    // References an existing subscription
    //var topic = pubsub.topic(topicName)
    const subscription = pubsub.subscription(subscriptionName);
    

    //Function to format time and date
    function formatDatetime (TimeStamp){
    	var formatted =  (TimeStamp.getHours()) + ':' + (TimeStamp.getMinutes()) + ':' + (TimeStamp.getSeconds()) + ':' + (TimeStamp.getMilliseconds());
    	return formatted;
    }

    // Create an event handler to handle messages
    let messageCount = 0;
    const messageHandler = message => {
      console.log(`Received message: ${message.id}`);
      console.log(`\tData: ${message.data}`);
      console.log(`\tAttributes: ${message.attributes}`);
      var obj = JSON.parse(message.data);
	  console.log(`\tTimeStamp: ${obj.messages.timestamp}`);
	  console.log(`\tAmount: ${obj.messages.amount}`);
	  
      
      messageCount += 1;
      console.log(`Message count : ${messageCount}`);
      
      message.ack();
      console.log(`Message Acknowledged`);

      // This doesn't ack the message, but allows more messages to be retrieved
      // if your limit was hit or if you don't want to ack the message.
      // message.nack();
    


      // Get creation timestamp
      var x = new Date(obj.messages.timestamp);    
      // Time formatting for x-axis in chart
      var formatted = formatDatetime(x);
      var Count = obj.messages.amount;

      console.log(`Extracting Timestamp: ${formatted}`);
      console.log(`Counts : ${Count}`);
      strData = {"label": formatted,
                     "value": Count
                  }
      socket.emit('news', strData);
      console.log(``);
      };

    // Listen for new messages until timeout is hit
      subscription.on(`message`, messageHandler);
      
      setTimeout(() => {
      	console.log(`Enter timeout`);
      	//subscription.removeListener('message', messageHandler);
        console.log(`0 message(s) received.`);
        var x = new Date();
        var formatted =  formatDatetime(x);
      	var Count = 0;
      	console.log(`Extracting Timestamp: ${formatted}`)
      	strData = {"label": formatted,
                     "value": Count
                  }
        console.log(`strData : ${strData}`)
        console.log(``);
        socket.emit('news', strData);
        
      }, timeout);

    //other handling
    if ( typeof strData == 'undefined') {
    	console.log(`Something else happened`)
    	var x = new Date();
        var formatted =  formatDatetime(x);
        console.log(`Extracting Timestamp: ${formatted}`)
    	strData = {"label": formatted,
                     "value": 9
                  }
        socket.emit('news', strData);
              }

    console.log(`strData : ${strData}`);
    console.log(``);
    
    
    
});


//server listening on port 8080
http.listen(8080, function() {
   console.log('listening on *:8080');
});

这是我用来显示图表的代码：

/*globals io */
var FusionCharts = require("fusioncharts");
require("fusioncharts/fusioncharts.charts")(FusionCharts);
require("fusioncharts/fusioncharts.widgets")(FusionCharts);

  var socket = io();
    	var transactionChart = new FusionCharts({
    		id: "mychart",
	        type: 'realtimecolumn',
	        width: '700',
	        height: '350',
	        dataFormat: 'json',
	        dataSource: {
	            "chart": {
    	         "caption": "Streaming Data Visualization",
                    "subCaption": "Testing",
                    "yaxismaxvalue": "10",
                    "numdisplaysets": "10",
                    "yAxisName":"Quantity",
                    "labeldisplay": "rotate",
                    "showLegend":"0",
                    "showValues": "0",
                    "numbersuffix": "Kg",
                    "showlabels": "1",
/*This parameter lets you set whether you want the latest value (received from server) to be displayed on the chart or not*/
                    "showRealTimeValue": "0",
/*For this parameter, you can specify the number of seconds after which the chart will look for new data. This process will happen continuously - i.e., if you specify 5 seconds here, the chart will look for new data every 5 seconds*/
                     "refreshInterval":".1",
/*If you want the chart to keep polling for new data every x seconds and queue it, you can specify that x seconds as updateInterval. This helps you poll at different intervals and then draw at another interval (specified as refreshInterval)*/
                    "updateInterval":".1",
                    "yAxisNamePadding":"10",
                    //Cosmetics
                    "paletteColors" : "#0075c2,#1aaf5d",
                    "baseFontColor" : "#333333",
                    "baseFont" : "Helvetica Neue,Arial",
                    "captionFontSize" : "14",
                    "subcaptionFontSize" : "14",
                    "subcaptionFontBold" : "0",
                    "showBorder" : "0",
                    "bgColor" : "#ffffff",
                    "showShadow" : "0",
                    "canvasBgColor" : "#ffffff",
                    "canvasBorderAlpha" : "0",
                    "divlineAlpha" : "100",
                    "divlineColor" : "#999999",
                    "divlineThickness" : "1",
                    "divLineIsDashed" : "1",
                    "divLineDashLen" : "1",
                    "divLineGapLen" : "1",
                    "usePlotGradientColor" : "0",
                    "showplotborder" : "0",
                    "valueFontColor" : "#ffffff",
                    "placeValuesInside" : "1",
                    "rotateValues" : "1",
                    "showXAxisLine" : "1",
                    "xAxisLineThickness" : "1",
                    "xAxisLineColor" : "#999999",
                    "showAlternateHGridColor" : "0",
                    "legendBgAlpha" : "0",
                    "legendBorderAlpha" : "0",
                    "legendShadow" : "0",
                    "legendItemFontSize" : "10",
                    "legendItemFontColor" : "#666666"
    	            },
	            "categories": [
	                {
	                    "category": [
	                        { "label": "Start" }
	                    ]
	                }
	            ],
	            "dataset": [ 
	                {
	                    "seriesname": "",
	                    "alpha": "100",
	                    "data": [
	                        { "value": "3" }
	                    ]
	                }
	            ]      
	        }
    	}).render("chart-container");
//On connection with socket, will start receiving the data
	  socket.connect('http://localhost:8080/');
	  socket.on('news', function (data) {
	    function updateData() {
                         //Converting the fetched data in FusionCharts format
	    	var strData = "&label=" + data.label + "&value=" + data.value;
                        //feeding the data to the real time chart
	    	FusionCharts.items.mychart.feedData(strData);
	    }
	    //calling the update method
	    updateData();

	 });

这是我的 index.html 代码：

<!DOCTYPE html>
<html>
   <head>
      <title>Hello world</title>
      <script src="/socket.io/socket.io.js"></script>
   </head>
 
   <body>
   <div id="chart-container">FusionCharts will render here</div>
        <script src="bundle.js"></script>
   </body>
</html>

我对 Javascript 还是很陌生，以前从未在网络应用程序上工作过。我可能错过了关于事情如何运作的重要知识。但我确实有些怀疑，虽然我不确定。

可能是因为每次我刷新页面时都会建立一个新的套接字连接，并且丢失的消息实际上是由以前的连接接收的（因此它不显示）？

我尝试过的几个解决方案仍然无效： Node.js Socket.io 页面刷新多个连接

任何人都可以帮助我吗？

标签： javascriptnode.jssocket.iogoogle-cloud-pubsubfusioncharts

解决方案

我碰巧找到了解决我自己问题的有效方法。

不完全确定原因，但发生了以下情况：每当我刷新页面时，旧的套接字连接将被断开，并会创建一个新的套接字连接。这个新的套接字将监听相同的订阅。

虽然旧套接字的状态在页面刷新期间似乎断开了连接，但由于某种原因它仍然侦听相同的订阅。这会导致 10 条消息在两个或多个连接之间拆分（取决于页面刷新次数）。

但是，浏览器上显示的只是最新的连接。这可能看起来像是消息丢失了，而实际上它们分布在许多（看不见的）连接上。当我尝试打印每条消息结束的“套接字 ID”时，这一点很明显。

所以我所做的基本上是在套接字断开连接期间添加一个小处理：

//on Disconnect
socket.on('disconnect', function () {
console.log("LOG: just disconnected: " + socket.id);
subscription.removeListener('message', messageHandler);

所以每当一个socket断开连接时，它也会停止监听订阅，新的socket会收到一整套消息。

推荐阅读