javascript - Socket.io 与 PubSub : Less to no real-time data displayed after multiple page refresh

Question

我正在尝试使用 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 条消息。这是我运行应用程序时图表的样子:

Example of Visualization

问题是它仅在我第一次运行应用程序并打开页面时有效。当我刷新页面并推送另外 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 page refresh multiple connections

有人可以帮我解决这个问题吗？

Answer 1

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

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

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

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

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

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

因此，每当一个套接字断开连接时，它也会停止监听订阅，新的套接字将收到一组完整的消息。