問題描述
我有一個 XMLHttpRequest 和一個 progress 事件處理程序，它正在請求一個連續發送添加消息塊的分塊頁面.如果我沒有設置 responseType，我可以在每個 progress 事件中訪問 XMLHttpRequest 的 response 屬性，并且處理額外的消息塊.這種方法的問題是瀏覽器必須將整個響應保存在內存中，最終瀏覽器會因為這種內存浪費而崩潰.
I have a XMLHttpRequest with a progress event handler that is requesting a chunked page which continuously sends adds message chunks. If I do not set a responseType, I can access the response property of the XMLHttpRequest in each progress event and handle the additional message chunk. The problem of this approach is that the browser must keep the entire response in memory, and eventually, the browser will crash due to this memory waste.
所以，我嘗試了一個arraybuffer的responseType，希望可以對緩沖區進行切片，防止之前的內存浪費過多.不幸的是，此時 progress 事件處理程序不再能夠讀取 XMLHttpRequest 的 response 屬性.progress 事件的事件參數也不包含緩沖區.這是我嘗試的一個簡短的、獨立的示例(這是為 node.js 編寫的):
So, I tried a responseType of arraybuffer in the hope that I can slice the buffer to prevent the previous excessive memory waste. Unfortunately, the progress event handler is no longer capable of reading the response property of the XMLHttpRequest at this point. The event parameter of the progress event does not contain the buffer, either. Here is a short, self-contained example of my attempt at this (this is written for node.js):
var http = require('http');
// -- The server.
http.createServer(function(req, res) {
if (req.url === '/stream') return serverStream(res);
serverMain(res);
}).listen(3000);
// -- The server functions to send a HTML page with the client code, or a stream.
function serverMain(res) {
res.writeHead(200, {'Content-Type': 'text/html'});
res.write('<html><body>Hello World</body><script>');
res.end(client.toString() + ';client();</script></html>');
}
function serverStream(res) {
res.writeHead(200, {'Content-Type': 'text/html'});
setInterval(function() {
res.write('Hello World<br />
');
}, 1000);
}
// -- The client code which runs in the browser.
function client() {
var xhr = new XMLHttpRequest();
xhr.addEventListener('progress', function() {
if (!xhr.response) return console.log('progress without response :-(');
console.log('progress: ' + xhr.response.size);
}, false);
xhr.open('GET', '/stream', true);
xhr.responseType = 'arraybuffer';
xhr.send();
}
progress 事件處理程序無法訪問我想要的 response.如何以節省內存的方式處理瀏覽器中的消息塊?請不要建議 WebSocket.我不希望只使用一個來處理消息塊的只讀流.
The progress event handler has no access to the response I wanted. How can I handle the message chunks in the browser in a memory-efficient way? Please do not suggest a WebSocket. I do not wish to use one just to process a read-only stream of message chunks.
推薦答案
XMLHttpRequest 似乎并不是真正為這種用法而設計的.顯而易見的解決方案是輪詢，這是 XMLHttpRequest 的一種流行用法，但我猜你不想錯過流中的數據，這些數據會在調用之間滑動.
XMLHttpRequest doesn't seem really designed for this kind of usage. The obvious solution is polling, which is a popular use of XMLHttpRequest but I'm guessing you don't want to miss data from your stream that would slip between the calls.
對于我的問題可以以某種方式識別真實"數據塊還是基本上是隨機數據?，您回答了 通過一些努力，可以通過向服務器端添加各種事件 ID 來識別塊
To my question Can the "real" data chunks be identified in some way or is it basically random data ?, you answered With some effort, the chunks could be identified by adding an event-id of sorts to the server-side
基于這個前提，我提出:
Based on this premise, I propose:
- 連接流并設置進度監聽器(簡稱
listenerA()). - 當一個塊到達時，處理它并輸出它.保留對
listenerA()收到的第一個和最后一個塊的 id 的引用.計算listenerA()收到了多少塊. listenerA()收到一定數量的塊后，生成另一個線程"(連接 + 偵聽器，listenerB())執行步驟 1 和 2與第一個并行，但將處理后的數據保存在緩沖區中而不是輸出.- 當
listenerA()接收到與listenerB()接收到的第一個chunk id相同的chunk時，發送一個信號給listenerB()，斷開第一個連接并終止listenerA(). - 當
listenerB()接收到來自listenerA()的終止信號時，將緩沖區轉儲到輸出，繼續正常處理. - 讓
listenerB()在與之前相同的條件下生成listenerC(). - 根據需要不斷重復使用盡可能多的連接和偵聽器.
- Connect to the stream and set up the progress listener (referred to as
listenerA()). - When a chunk arrives, process it and output it. Keep a reference to the ids of both the first and last chunk received by
listenerA(). Count how many chunkslistenerA()has received. - After
listenerA()has received a certain amount of chunks, spawn another "thread" (connection + listener,listenerB()) doing the steps 1 and 2 in parallel to the first one but keep the processed data in a buffer instead of outputting it. - When
listenerA()receives the chunk with the same id as the first chunk received bylistenerB(), send a signal tolistenerB(), drop the first connection and killlistenerA(). - When
listenerB()receives the termination signal from thelistenerA(), dump the buffer to the output and keep processing normally. - Have
listenerB()spawnlistenerC()on the same conditions as before. - Keep repeating with as many connections + listeners as necessary.
通過使用兩個重疊的連接，您可以防止由于斷開單個連接然后重新連接而可能導致的塊丟失.
By using two overlapping connections, you can prevent the possible loss of chunks that would result from dropping a single connection and then reconnecting.
- 這假設所有連接的數據流都是相同的，并且不引入一些個性化設置.
- 根據流的輸出速率和連接延遲，從一個連接轉換到另一個連接期間的緩沖區轉儲可能會很明顯.
- 您還可以衡量總響應大小而不是塊數來決定何時切換到新連接.
- 可能需要保留一個完整的塊 id 列表以進行比較，而不僅僅是第一個和最后一個，因為我們無法保證重疊的時間.
XMLHttpRequest的responseType必須設置為其默認值""或"text"，才能返回文本.其他數據類型不會返回部分response.請參閱 https://xhr.spec.whatwg.org/#the-response-attribute
- This assumes the data stream is the same for all connections and doesn't introduce some individualized settings.
- Depending on the output rate of the stream and the connection delay, the buffer dump during the transition from one connection to another might be noticeable.
- You could also measure the total response size rather than the chunks count to decide when to switch to a new connection.
- It might be necessary to keep a complete list of chunks ids to compare against rather than just the first and last one because we can't guarantee the timing of the overlap.
- The
responseTypeofXMLHttpRequestmust be set to its default value of""or "text", to return text. Other datatypes will not return a partialresponse. See https://xhr.spec.whatwg.org/#the-response-attribute
以下代碼是一個 node.js 服務器，它輸出一致的元素流以用于測試目的.您可以打開到它的多個連接，輸出將是相同的 accross 會話，減去可能的服務器延遲.
The following code is a node.js server that outputs a consistent stream of elements for testing purposes. You can open multiple connections to it, the output will be the same accross sessions, minus possible server lag.
http://localhost:5500/stream
將返回 id 為遞增數字的數據
will return data where id is an incremented number
http://localhost:5500/streamRandom
將返回數據，其中 id 是一個隨機的 40 個字符長的字符串.這是為了測試一個不能依賴 id 來排序數據的場景.
will return data where id is a random 40 characters long string. This is meant to test a scenario where the id can not be relied upon for ordering the data.
var crypto = require('crypto');
// init + update nodeId
var nodeId = 0;
var nodeIdRand = '0000000000000000000000000000000000000000';
setInterval(function() {
// regular id
++nodeId;
//random id
nodeIdRand = crypto.createHash('sha1').update(nodeId.toString()).digest('hex');
}, 1000);
// create server (port 5500)
var http = require('http');
http.createServer(function(req, res) {
if(req.url === '/stream') {
return serverStream(res);
}
else if(req.url === '/streamRandom') {
return serverStream(res, true);
}
}).listen(5500);
// serve nodeId
function serverStream(res, rand) {
// headers
res.writeHead(200, {
'Content-Type' : 'text/plain',
'Access-Control-Allow-Origin' : '*',
});
// remember last served id
var last = null;
// output interval
setInterval(function() {
// output on new node
if(last != nodeId) {
res.write('[node id="'+(rand ? nodeIdRand : nodeId)+'"]');
last = nodeId;
}
}, 250);
}
概念證明，使用上述 node.js 服務器代碼
<!DOCTYPE html>
<html>
<head>
<meta http-equiv="Content-Type" content="text/html; charset=utf-8"/>
</head>
<body>
<button id="stop">stop</button>
<div id="output"></div>
<script>
/*
Listening to a never ending page load (http stream) without running out of
memory by using concurrent overlapping connections to prevent loss of data,
using only xmlHttpRequest, under the condition that the data can be identified.
listen arguments
url url of the http stream
chunkMax number of chunks to receive before switching to new connection
listen properties
output a reference to a DOM element with id "output"
queue an array filled with non-duplicate received chunks and metadata
lastFetcherId an incrementing number used to assign an id to new fetchers
fetchers an array listing all active fetchers
listen methods
fire internal use fire an event
stop external use stop all connections
fetch internal use starts a new connection
fetchRun internal use initialize a new fetcher object
Usage
var myListen = new listen('http://localhost:5500/streamRandom', 20);
will listen to url "http://localhost:5500/streamRandom"
will switch connections every 20 chunks
myListen.stop()
will stop all connections in myListen
*/
function listen(url, chunkMax) {
// main ref
var that = this;
// output element
that.output = document.getElementById('output');
// main queue
that.queue = [];
// last fetcher id
that.lastFetcherId = 0;
// list of fetchers
that.fetchers = [];
//********************************************************* event dispatcher
that.fire = function(name, data) {
document.dispatchEvent(new CustomEvent(name, {'detail':data}));
}
//******************************************************** kill all fetchers
that.stop = function() {
that.fire('fetch-kill', -1);
}
//************************************************************** url fetcher
that.fetch = function(fetchId, url, fetchRef) {
//console.log('start fetcher #'+fetchId);
var len = 0;
var xhr = new XMLHttpRequest();
var cb_progress;
var cb_kill;
// progress listener
xhr.addEventListener('progress', cb_progress = function(e) {
// extract chunk data
var chunkData = xhr.response.substr(len);
// chunk id
var chunkId = chunkData.match(/id="([a-z0-9]+)"/)[1];
// update response end point
len = xhr.response.length;
// signal end of chunk processing
that.fire('chunk-ready', {
'fetchId' : fetchId,
'fetchRef' : fetchRef,
'chunkId' : chunkId,
'chunkData' : chunkData,
});
}, false);
// kill switch
document.addEventListener('fetch-kill', cb_kill = function(e) {
// kill this fetcher or all fetchers (-1)
if(e.detail == fetchId || e.detail == -1) {
//console.log('kill fetcher #'+fetchId);
xhr.removeEventListener('progress', cb_progress);
document.removeEventListener('fetch-kill', cb_kill);
xhr.abort();
that.fetchers.shift(); // remove oldest fetcher from list
xhr = null;
delete xhr;
}
}, false);
// go
xhr.open('GET', url, true);
xhr.responseType = 'text';
xhr.send();
};
//****************************************************** start a new fetcher
that.fetchRun = function() {
// new id
var id = ++that.lastFetcherId;
//console.log('create fetcher #'+id);
// create fetcher with new id
var fetchRef = {
'id' : id, // self id
'queue' : [], // internal queue
'chunksIds' : [], // retrieved ids, also used to count
'hasSuccessor' : false, // keep track of next fetcher spawn
'ignoreId' : null, // when set, ignore chunks until this id is received (this id included)
};
that.fetchers.push(fetchRef);
// run fetcher
that.fetch(id, url, fetchRef);
};
//************************************************ a fetcher returns a chunk
document.addEventListener('chunk-ready', function(e) {
// shorthand
var f = e.detail;
// ignore flag is not set, process chunk
if(f.fetchRef.ignoreId == null) {
// store chunk id
f.fetchRef.chunksIds.push(f.chunkId);
// create queue item
var queueItem = {'id':f.chunkId, 'data':f.chunkData};
// chunk is received from oldest fetcher
if(f.fetchId == that.fetchers[0].id) {
// send to main queue
that.queue.push(queueItem);
// signal queue insertion
that.fire('queue-new');
}
// not oldest fetcher
else {
// use fetcher internal queue
f.fetchRef.queue.push(queueItem);
}
}
// ignore flag is set, current chunk id the one to ignore
else if(f.fetchRef.ignoreId == f.chunkId) {
// disable ignore flag
f.fetchRef.ignoreId = null;
}
//******************** check chunks count for fetcher, threshold reached
if(f.fetchRef.chunksIds.length >= chunkMax && !f.fetchRef.hasSuccessor) {
// remember the spawn
f.fetchRef.hasSuccessor = true;
// spawn new fetcher
that.fetchRun();
}
/***********************************************************************
check if the first chunk of the second oldest fetcher exists in the
oldest fetcher.
If true, then they overlap and we can kill the oldest fetcher
***********************************************************************/
if(
// is this the oldest fetcher ?
f.fetchId == that.fetchers[0].id
// is there a successor ?
&& that.fetchers[1]
// has oldest fetcher received the first chunk of its successor ?
&& that.fetchers[0].chunksIds.indexOf(
that.fetchers[1].chunksIds[0]
) > -1
) {
// get index of last chunk of the oldest fetcher within successor queue
var lastChunkId = that.fetchers[0].chunksIds[that.fetchers[0].chunksIds.length-1]
var lastChunkIndex = that.fetchers[1].chunksIds.indexOf(lastChunkId);
// successor has not reached its parent last chunk
if(lastChunkIndex < 0) {
// discard whole queue
that.fetchers[1].queue = [];
that.fetchers[1].chunksIds = [];
// set ignore id in successor to future discard duplicates
that.fetchers[1].ignoreId = lastChunkId;
}
// there is overlap
else {
/**
console.log('triming queue start: '+that.fetchers[1].queue.length
+" "+(lastChunkIndex+1)
+" "+(that.fetchers[1].queue.length-1)
);
/**/
var trimStart = lastChunkIndex+1;
var trimEnd = that.fetchers[1].queue.length-1;
// trim queue
that.fetchers[1].queue = that.fetchers[1].queue.splice(trimStart, trimEnd);
that.fetchers[1].chunksIds = that.fetchers[1].chunksIds.splice(trimStart, trimEnd);
//console.log('triming queue end: '+that.fetchers[1].queue.length);
}
// kill oldest fetcher
that.fire('fetch-kill', that.fetchers[0].id);
}
}, false);
//***************************************************** main queue processor
document.addEventListener('queue-new', function(e) {
// process chunks in queue
while(that.queue.length > 0) {
// get chunk and remove from queue
var chunk = that.queue.shift();
// output item to document
if(that.output) {
that.output.innerHTML += "<br />"+chunk.data;
}
}
}, false);
//****************************************************** start first fetcher
that.fetchRun();
};
// run
var process = new listen('http://localhost:5500/streamRandom', 20);
// bind global kill switch to button
document.getElementById('stop').addEventListener('click', process.stop, false);
</script>
</body>
</html>
這篇關于使用 XMLHttpRequest 的內存高效消息塊處理的文章就介紹到這了，希望我們推薦的答案對大家有所幫助，也希望大家多多支持html5模板網！