I'm working with an API to parse stock market data in NodeJS.
The program works fine up to around 30K transactions per second, which is fine for the slower parts of the day. During market open though, it can exceed 100K transactions per second and the heap explodes to 30GB+, lagging and eventually crashing, despite being on an extremely fast machine.
I'm relatively new to NodeJS, but the bottleneck appears to be the below section of code, I obtained from a sample client that is in an extension of the EventEmitter class.
onMessage( data ){
data = JSON.parse( data )
data.map(( msg ) => {
if( msg.ev === 'status' ){
console.log('Status Update:', msg.message)
}
this.emit(msg.ev, msg)
})
Is there a more efficient way to code this that could give a 2x+ speed improvement? The individual bits of JSON are quite small such as the following:
{
"ev": "T",
"sym": "MSFT",
"x": 4,
"i": "12345",
"z": 3,
"p": 114.125,
"s": 100,
"c": [
0,
12
],
"t": 1536036818784
}
In my comments, I suggested you take the data.map() loop out and send all that at once and then let the client iterate through the individual pieces:
onMessage( data ){
data = JSON.parse( data )
this.emit("multiMsg", data);
}
Then, the client would have this:
socket.on("multiMsg", (data) => {
for (const item of data) {
// process each item here
console.log(item);
}
});
If you're still getting lots of onMessage() calls in rapid fire, you can accumulate them for a short period of time (the time to choose depends upon the acceptable time delay for your app). That also allows you to drastically reduce the number of separate messages you send to the client during high traffic times which can be many, many times more efficient for the server. So, if you would have received 100 calls to onMessage() within 50ms, then this modification will send one accumulated message to the client, instead of 100 separate messages to the client.
Here's an idea how that could work:
const msgQueueTime = 500; // pick an appropriate time delay here
let msgQueue = [];
let msgQueueTimer = null;
onMessage( data ){
data = JSON.parse( data )
if (!msgQueueTimer) {
msgQueueTimer = setTimeout(() => {
// send data we have accumulated
this.emit("multiMsg", msgQueue);
msgQueue = [];
msgQueueTimer = null;
}, msgQueueTime);
}
// add data onto our queue array
msgQueue.push(...data);
}