Lesson 21-Node.js Advanced Features

Event-Driven Architecture and Asynchronous I/O

Core Mechanism of the Event Loop

Event Loop Working Principle:
Node.js’s event loop is implemented based on libuv, a continuously running process responsible for checking the event queue and executing corresponding callback functions. The event loop is divided into multiple phases, each handling specific types of events.

// Simplified representation of the event loop
while (true) {
  // 1. Execute timer callbacks (setTimeout/setInterval)
  processTimers();

  // 2. Execute I/O callbacks (system-level callbacks)
  processIOCallbacks();

  // 3. Execute idle/prepare phase (internal use)
  idlePrepare();

  // 4. Poll phase (retrieve new I/O events)
  poll();

  // 5. Check phase (setImmediate callbacks)
  check();

  // 6. Close callbacks (e.g., socket.on('close'))
  closeCallbacks();
}

Event Loop Flowchart:

  1. Timers Phase: Executes setTimeout and setInterval callbacks
  2. Pending Callbacks Phase: Executes system-level operation callbacks (e.g., TCP errors)
  3. Idle/Prepare Phase: Internal use
  4. Poll Phase: Retrieves new I/O events and executes I/O-related callbacks
  5. Check Phase: Executes setImmediate callbacks
  6. Close Callbacks Phase: Executes callbacks for close events (e.g., socket.on(‘close’))

libuv Library Working Principle

libuv Architecture:
libuv is the core library of Node.js, providing cross-platform asynchronous I/O capabilities. It includes two main components: a thread pool and an event-driven system.

// Simplified pseudocode for libuv event loop
int uv_run(uv_loop_t* loop, uv_run_mode mode) {
  while (r != 0 && loop->stop_flag == 0) {
    uv__update_time(loop);
    uv__run_timers(loop);          // Timers phase
    uv__run_pending(loop);         // Pending Callbacks phase
    uv__run_idle(loop);            // Idle/Prepare phase
    uv__run_prepare(loop);         // Idle/Prepare phase

    blocked = uv__io_poll(loop, timeout); // Poll phase

    uv__run_check(loop);           // Check phase
    uv__run_closing_handles(loop); // Close Callbacks phase

    if (mode == UV_RUN_ONCE) {
      uv__update_time(loop);
      uv__run_timers(loop);
    }
  }
  return r;
}

Thread Pool Implementation:
libuv uses a default thread pool of 4 threads to handle file I/O, DNS, and other operations.

// Thread pool workflow
void uv__work_submit(uv_loop_t* loop, uv__work* w) {
  // 1. Add task to queue
  QUEUE_INSERT_TAIL(&loop->wq, &w->wq);

  // 2. Notify worker thread
  uv_async_send(&loop->wq_async);
}

// Worker thread execution function
static void uv__worker(void* arg) {
  while (1) {
    // 1. Wait for tasks
    QUEUE_WAIT(&loop->wq);

    // 2. Execute task
    w->work(w);

    // 3. Complete callback
    w->done(w);
  }
}

Asynchronous I/O Implementation

Non-Blocking I/O Operations:
Node.js implements non-blocking I/O through libuv, primarily using an event notification mechanism.

// Non-blocking file read example
const fs = require('fs');

fs.readFile('example.txt', (err, data) => {
  if (err) throw err;
  console.log(data.toString());
});

console.log('Continue executing other operations');

Underlying Implementation Process:

  1. The application calls fs.readFile
  2. Node.js initiates an asynchronous file read request via libuv
  3. libuv adds the request to the thread pool or uses system-level asynchronous I/O
  4. The main thread continues executing subsequent code
  5. Upon I/O completion, libuv adds the callback to the event queue
  6. The event loop executes the callback during the Poll phase

Event Loop Phases

Detailed Phase Analysis:

  1. Timers Phase:
    • Executes setTimeout and setInterval callbacks
    • Accuracy depends on the operating system’s scheduling
setTimeout(() => {
  console.log('Timeout 1');
}, 0);

setTimeout(() => {
  console.log('Timeout 2');
}, 10);
  1. Pending Callbacks Phase:
    • Executes callbacks for system-level operations (e.g., TCP errors)
    • Typically used internally by libuv
  2. Idle/Prepare Phase:
    • Internal use, not relevant to developers
  3. Poll Phase:
    • Retrieves new I/O events
    • Executes I/O-related callbacks
    • May block if no callbacks are pending
const fs = require('fs');

fs.readFile('file1.txt', () => {
  console.log('File 1 read');
});

fs.readFile('file2.txt', () => {
  console.log('File 2 read');
});
  1. Check Phase:
    • Executes setImmediate callbacks
    • Runs immediately after the Poll phase
setImmediate(() => {
  console.log('Immediate 1');
});

setImmediate(() => {
  console.log('Immediate 2');
});
  1. Close Callbacks Phase:
    • Executes callbacks for close events
    • e.g., socket.on(‘close’)

Best Practices for Asynchronous Programming

Avoiding Callback Hell:

// Callback hell example
fs.readFile('file1.txt', (err, data1) => {
  if (err) return console.error(err);
  fs.readFile('file2.txt', (err, data2) => {
    if (err) return console.error(err);
    fs.writeFile('output.txt', data1 + data2, err => {
      if (err) return console.error(err);
      console.log('File merge completed');
    });
  });
});

// Improved with Promises
function readFilePromise(path) {
  return new Promise((resolve, reject) => {
    fs.readFile(path, (err, data) => {
      if (err) reject(err);
      else resolve(data);
    });
  });
}

readFilePromise('file1.txt')
  .then(data1 => readFilePromise('file2.txt')
    .then(data2 => writeFilePromise('output.txt', data1 + data2))
    .then(() => console.log('File merge completed'))
    .catch(console.error)
  );

// Improved with async/await
async function mergeFiles() {
  try {
    const data1 = await readFilePromise('file1.txt');
    const data2 = await readFilePromise('file2.txt');
    await writeFilePromise('output.txt', data1 + data2);
    console.log('File merge completed');
  } catch (err) {
    console.error(err);
  }
}

Event-Driven Architecture Design:

// Custom event emitter example
const EventEmitter = require('events');

class MyEmitter extends EventEmitter {}

const myEmitter = new MyEmitter();

// Register event listener
myEmitter.on('event', (a, b) => {
  console.log('Event triggered', a, b);
});

// Trigger event
myEmitter.emit('event', 'Parameter 1', 'Parameter 2');

// One-time listener
myEmitter.once('onceEvent', () => {
  console.log('Event triggered once');
});

// Error handling
myEmitter.on('error', (err) => {
  console.error('Error occurred:', err);
});

Processes and Threads

Child Process Module

Spawn Method:

const { spawn } = require('child_process');

const ls = spawn('ls', ['-lh', '/usr']);

ls.stdout.on('data', (data) => {
  console.log(`stdout: ${data}`);
});

ls.stderr.on('data', (data) => {
  console.error(`stderr: ${data}`);
});

ls.on('close', (code) => {
  console.log(`Child process exited with code ${code}`);
});

Exec Method:

const { exec } = require('child_process');

exec('ls -lh /usr', (error, stdout, stderr) => {
  if (error) {
    console.error(`Execution error: ${error}`);
    return;
  }
  console.log(`stdout: ${stdout}`);
  if (stderr) {
    console.error(`stderr: ${stderr}`);
  }
});

Fork Method:

// parent.js
const { fork } = require('child_process');
const child = fork('child.js');

child.on('message', (msg) => {
  console.log('Message from child:', msg);
});

child.send({ hello: 'world' });

// child.js
process.on('message', (msg) => {
  console.log('Message from parent:', msg);
  process.send({ response: 'Message received' });
});

Cluster Mode

Basic Cluster Example:

const cluster = require('cluster');
const http = require('http');
const numCPUs = require('os').cpus().length;

if (cluster.isMaster) {
  console.log(`Master process ${process.pid} is running`);

  // Fork worker processes
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  cluster.on('exit', (worker, code, signal) => {
    console.log(`Worker process ${worker.process.pid} exited`);
  });
} else {
  // Workers can share any TCP connection
  http.createServer((req, res) => {
    res.writeHead(200);
    res.end('Hello World\n');
  }).listen(8000);

  console.log(`Worker process ${process.pid} started`);
}

Load Balancing Strategy:

cluster.schedulingPolicy = cluster.SCHED_RR; // Default round-robin scheduling
// or
cluster.schedulingPolicy = cluster.SCHED_NONE; // OS scheduling

Worker Threads Pool

Basic Thread Example:

const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');

if (isMainThread) {
  // Main thread
  const worker = new Worker(__filename, { workerData: { start: 1, end: 100 } });
  worker.on('message', (result) => {
    console.log('Calculation result:', result);
  });
  worker.on('error', (err) => {
    console.error('Worker thread error:', err);
  });
  worker.on('exit', (code) => {
    if (code !== 0) {
      console.error(`Worker thread exited abnormally with code ${code}`);
    }
  });
} else {
  // Worker thread
  let sum = 0;
  for (let i = workerData.start; i <= workerData.end; i++) {
    sum += i;
  }
  parentPort.postMessage(sum);
}

Thread Pool Implementation:

const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');
const os = require('os');

if (isMainThread) {
  class ThreadPool {
    constructor(size = os.cpus().length) {
      this.workers = [];
      this.taskQueue = [];

      for (let i = 0; i < size; i++) {
        this.addWorker();
      }
    }

    addWorker() {
      const worker = new Worker(__filename);
      worker.on('message', (result) => {
        const { resolve } = this.taskQueue.shift();
        resolve(result);
        this.processNext();
      });
      worker.on('error', (err) => {
        const { reject } = this.taskQueue.shift();
        reject(err);
        this.addWorker(); // Replace crashed worker
      });
      this.workers.push(worker);
    }

    processNext() {
      if (this.taskQueue.length > 0) {
        const { task } = this.taskQueue[0];
        const worker = this.workers.find(w => !w.isBusy);
        if (worker) {
          worker.isBusy = true;
          worker.postMessage(task);
        }
      }
    }

    execute(task) {
      return new Promise((resolve, reject) => {
        this.taskQueue.push({ task, resolve, reject });
        this.processNext();
      });
    }
  }

  // Use thread pool
  const pool = new ThreadPool();
  pool.execute({ start: 1, end: 1000000 }).then(console.log);
} else {
  // Worker thread logic
  parentPort.on('message', (task) => {
    let sum = 0;
    for (let i = task.start; i <= task.end; i++) {
      sum += i;
    }
    parentPort.postMessage(sum);
  });
}

Inter-Process Communication

IPC Communication Example:

// parent.js
const { fork } = require('child_process');
const child = fork('child.js');

// Send message to child process
child.send({ message: 'Hello, child process' });

// Receive message from child process
child.on('message', (msg) => {
  console.log('Message from child:', msg);
});

// child.js
process.on('message', (msg) => {
  console.log('Message from parent:', msg);
  // Reply to parent process
  process.send({ response: 'Hello, parent process' });
});

Shared Memory Communication:

// Using SharedArrayBuffer for shared memory
// Note: Requires --experimental-worker flag
const { Worker } = require('worker_threads');

const sharedBuffer = new SharedArrayBuffer(1024);
const arr = new Int32Array(sharedBuffer);

const worker = new Worker('./worker.js', { workerData: sharedBuffer });

worker.on('message', (msg) => {
  console.log('Main thread received message:', msg);
  console.log('Shared array value:', arr[0]);
});

// worker.js
const { parentPort, workerData } = require('worker_threads');
const arr = new Int32Array(workerData);

// Modify shared memory
arr[0] = 42;

// Notify main thread
parentPort.postMessage('Shared memory updated');

Process Management and Resource Control

Process Management Example:

const { spawn } = require('child_process');

const child = spawn('node', ['long-running-task.js'], {
  stdio: 'pipe',
  detached: true // Make child process independent of parent
});

// Set process group ID for batch termination
child.unref(); // Allow parent to exit without waiting for child

// Resource limit example
const { exec } = require('child_process');
const maxMemory = '500MB'; // Maximum memory limit

exec(`node --max-old-space-size=${parseSize(maxMemory)} memory-intensive.js`, 
  (error, stdout, stderr) => {
    if (error) {
      console.error('Process execution failed:', error);
    }
  }
);

function parseSize(size) {
  const units = { KB: 1024, MB: 1024 * 1024, GB: 1024 * 1024 * 1024 };
  const match = size.match(/^(\d+)(KB|MB|GB)$/);
  if (!match) throw new Error('Invalid size format');
  return parseInt(match[1]) * units[match[2]];
}

Process Monitoring:

const { spawn } = require('child_process');

const child = spawn('node', ['app.js']);

// Monitor resource usage
setInterval(() => {
  const usage = process.memoryUsage();
  console.log(`Memory usage: ${Math.round(usage.rss / 1024 / 1024)}MB`);
}, 1000);

// Handle child process exit
child.on('exit', (code, signal) => {
  console.log(`Child process exited with code: ${code}, signal: ${signal}`);
  // Implement auto-restart logic here if needed
});

// Handle errors
child.on('error', (err) => {
  console.error('Child process error:', err);
});

Memory Management and Performance Optimization

V8 Engine Memory Management

V8 Memory Structure:

  • Young Generation: Stores newly created objects, reclaimed using the Scavenge algorithm
  • Old Generation: Stores long-lived objects, reclaimed using Mark-Sweep and Mark-Compact algorithms
  • Large Object Space: Stores objects larger than 1.5MB

Memory Allocation Example:

// Small objects allocated in young generation
function createSmallObjects() {
  const objects = [];
  for (let i = 0; i < 10000; i++) {
    objects.push({ id: i }); // Small objects
  }
  return objects;
}

// Large objects allocated directly in old generation
function createLargeObject() {
  // Allocate Buffer larger than 1.5MB
  return Buffer.alloc(2 * 1024 * 1024); // 2MB
}

Garbage Collection Triggers:

  1. Scavenge collection triggered when young generation space is full
  2. Mark-Sweep or Mark-Compact collection triggered when old generation space is full
  3. More aggressive collection triggered on allocation failure

Memory Leak Detection and Analysis

heapdump Usage Example:

const heapdump = require('heapdump');

// Manually trigger heap snapshot
process.on('SIGUSR2', () => {
  const filename = `/tmp/heapdump-${Date.now()}.heapsnapshot`;
  heapdump.writeSnapshot(filename, (err, filename) => {
    if (err) console.error(err);
    else console.log('Heap snapshot saved to', filename);
  });
});

// Periodically trigger heap snapshot (for debugging)
setInterval(() => {
  if (process.memoryUsage().heapUsed > 500 * 1024 * 1024) {
    heapdump.writeSnapshot(`/tmp/heapdump-auto-${Date.now()}.heapsnapshot`);
  }
}, 60000);

clinic.js Usage Example:

# Install clinic.js
npm install -g clinic

# Generate performance report
clinic doctor -- node app.js

# Generate flame graph
clinic flame -- node app.js

# Generate bubble graph
clinic bubbleprof -- node app.js

Common Memory Leak Patterns:

  1. Unintended global variables
  2. Uncleaned timers
  3. Unreleased closures
  4. Unremoved event listeners
  5. Unbounded cache growth

Performance Monitoring Tools

Node.js Inspector:

# Start Inspector
node --inspect app.js

# Or specify port
node --inspect=9229 app.js

Chrome DevTools Debugging:

  1. Open Chrome browser
  2. Navigate to chrome://inspect
  3. Click “Open dedicated DevTools for Node”
  4. Use Performance and Memory panels for analysis

pm2 Monitoring:

# Install pm2
npm install -g pm2

# Start application with monitoring
pm2 start app.js --name "my-app"

# View real-time monitoring
pm2 monit

# Generate performance report
pm2 report

# Memory usage limit
pm2 start app.js --max-memory-restart 300M

Code Optimization Strategies

Reducing Callback Nesting:

// Bad practice: Callback hell
fs.readFile('file1.txt', (err, data1) => {
  if (err) return console.error(err);
  fs.readFile('file2.txt', (err, data2) => {
    if (err) return console.error(err);
    // More nesting...
  });
});

// Good practice: Promise chain
function readFilePromise(path) {
  return new Promise((resolve, reject) => {
    fs.readFile(path, (err, data) => {
      if (err) reject(err);
      else resolve(data);
    });
  });
}

readFilePromise('file1.txt')
  .then(data1 => readFilePromise('file2.txt'))
  .then(data2 => {
    // Process data
  })
  .catch(console.error);

// Best practice: async/await
async function readFiles() {
  try {
    const data1 = await readFilePromise('file1.txt');
    const data2 = await readFilePromise('file2.txt');
    // Process data
  } catch (err) {
    console.error(err);
  }
}

Avoiding Blocking Operations:

// Blocking operation example (avoid this)
function calculateSync() {
  let sum = 0;
  for (let i = 0; i < 1e9; i++) {
    sum += i;
  }
  return sum;
}

// Non-blocking operation example
function calculateAsync() {
  return new Promise(resolve => {
    setImmediate(() => {
      let sum = 0;
      for (let i = 0; i < 1e9; i++) {
        sum += i;
      }
      resolve(sum);
    });
  });
}

// Better approach: Use Worker threads
const { Worker } = require('worker_threads');

function calculateWithWorker() {
  return new Promise((resolve, reject) => {
    const worker = new Worker(`
      let sum = 0;
      for (let i = 0; i < 1e9; i++) {
        sum += i;
      }
      parentPort.postMessage(sum);
    `, { eval: true });

    worker.on('message', resolve);
    worker.on('error', reject);
    worker.on('exit', code => {
      if (code !== 0) reject(new Error(`Worker stopped with exit code ${code}`));
    });
  });
}

High Concurrency Scenario Tuning

Cluster Mode Optimization:

const cluster = require('cluster');
const http = require('http');
const numCPUs = require('os').cpus().length;

if (cluster.isMaster) {
  // Master process
  console.log(`Master process ${process.pid} is running`);

  // Create worker processes based on CPU core count
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  // Handle process exit
  cluster.on('exit', (worker, code, signal) => {
    console.log(`Worker process ${worker.process.pid} exited`);
    // Auto-restart crashed worker
    cluster.fork();
  });

  // Adjust load balancing strategy
  cluster.schedulingPolicy = cluster.SCHED_RR; // Round-robin scheduling
} else {
  // Worker process
  // Create HTTP server
  const server = http.createServer((req, res) => {
    // Simulate request processing
    setTimeout(() => {
      res.writeHead(200);
      res.end('Hello World\n');
    }, Math.random() * 100); // Random delay to simulate processing time
  });

  server.listen(8000);
  console.log(`Worker process ${process.pid} started`);
}

Connection Pool Management:

const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');
const { GenericPool } = require('generic-pool');

if (isMainThread) {
  // Main thread creates connection pool
  const factory = {
    create: () => {
      // Create database connection
      return new Promise(resolve => {
        const connection = { id: Math.random() };
        console.log('Created new connection:', connection.id);
        setTimeout(() => resolve(connection), 100); // Simulate connection creation delay
      });
    },
    destroy: (connection) => {
      // Destroy connection
      console.log('Destroyed connection:', connection.id);
    },
    validate: (connection) => {
      // Validate connection
      return Promise.resolve(true);
    }
  };

  const pool = GenericPool.createPool(factory, {
    max: 10, // Maximum connections
    min: 2,  // Minimum connections
    acquireTimeoutMillis: 3000, // Connection acquisition timeout
    idleTimeoutMillis: 30000    // Idle connection timeout
  });

  // Use connection pool
  (async () => {
    const connection = await pool.acquire();
    try {
      // Process request with connection
      console.log('Using connection:', connection.id);
      await new Promise(resolve => setTimeout(resolve, 100)); // Simulate work
    } finally {
      pool.release(connection);
    }
  })();
} else {
  // Worker thread logic
}

Memory Optimization Techniques:

// 1. Use Buffer instead of strings for binary data
function processFile() {
  // Bad practice: String concatenation for binary data
  // let data = '';
  // fs.createReadStream('file.bin').on('data', chunk => {
  //   data += chunk; // Inefficient string concatenation
  // });

  // Good practice: Use Buffer
  const chunks = [];
  fs.createReadStream('file.bin')
    .on('data', chunk => {
      chunks.push(chunk); // Efficient Buffer concatenation
    })
    .on('end', () => {
      const data = Buffer.concat(chunks);
      // Process data
    });
}

// 2. Avoid global variables
// Bad practice:
let globalCache = {}; // Global variables are not garbage collected

// Good practice:
function createCache() {
  const cache = {}; // Local variables are collected after function ends
  return {
    get(key) { return cache[key]; },
    set(key, value) { cache[key] = value; }
  };
}

// 3. Explicitly release large objects no longer needed
function processLargeData() {
  let largeData = new Array(1e6).fill('data');
  // Process data...

  // Explicitly release reference when no longer needed
  largeData.length = 0; // Clear array
  // or
  largeData = null; // Dereference
}

Through this in-depth analysis, we gain a comprehensive understanding of Node.js’s core advanced features, including event-driven architecture, process and thread management, memory management, and performance optimization. Mastering these concepts is crucial for building high-performance, scalable Node.js applications.

Course: Node.js Asynchronous Patterns, Architecture & Frameworks
Share your love

Trending now

Exploring Web 3D Core Technologies Development
Assembly Language Programming From Low-Level Logic to System Kernel
WebRTC Advanced Guide：Media, Transport & Architecture
WebSocket Deep Guide：Building High-Performance Realtime Systems