进程间通信

进程间通信的基本概念

进程间通信（IPC）是操作系统提供的机制，允许不同进程之间交换数据和信息。在Node.js中，IPC尤为重要，因为Node.js采用单线程事件循环模型，需要借助子进程来处理CPU密集型任务。IPC机制使得主进程和子进程能够协同工作，共享数据和状态。

Node.js提供了多种IPC方式，包括但不限于：管道（pipe）、消息队列、共享内存、信号量和套接字等。每种方式都有其适用场景和优缺点，开发者需要根据具体需求选择最合适的通信方式。

Node.js中的子进程模块

Node.js内置的child_process模块是处理进程间通信的核心。该模块提供了几种创建子进程的方法：

const { spawn, exec, execFile, fork } = require('child_process');

其中fork()方法专门用于创建Node.js子进程，并且会自动建立IPC通道：

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

child.on('message', (msg) => {
  console.log('来自子进程的消息:', msg);
});

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

// child.js
process.on('message', (msg) => {
  console.log('来自父进程的消息:', msg);
  process.send({ foo: 'bar' });
});

进程间通信的常用模式

消息传递模式

这是Node.js中最常用的IPC方式，通过process.send()和process.on('message')实现双向通信。消息会被序列化为JSON格式传输，因此可以传递复杂对象：

// 父进程发送复杂对象
child.send({
  type: 'config',
  data: {
    port: 3000,
    env: 'production',
    features: ['logging', 'monitoring']
  }
});

// 子进程接收并处理
process.on('message', (msg) => {
  if (msg.type === 'config') {
    console.log('收到配置:', msg.data);
  }
});

流式通信

对于大量数据传输，可以使用标准输入输出流（stdio）进行通信：

// 使用spawn创建子进程并通过流通信
const { spawn } = require('child_process');
const child = spawn('node', ['child.js'], {
  stdio: ['pipe', 'pipe', 'pipe', 'ipc'] // 显式启用IPC
});

child.stdout.on('data', (data) => {
  console.log(`子进程输出: ${data}`);
});

child.stdin.write('父进程通过stdin发送数据\n');

高级IPC技术

共享内存

虽然Node.js不直接支持共享内存，但可以通过第三方模块如shared-memory实现：

const SharedMemory = require('shared-memory');
const memory = new SharedMemory('my-shared-memory', 1024);

// 进程A写入数据
memory.write(0, Buffer.from('Hello from Process A'));

// 进程B读取数据
const data = memory.read(0, 20);
console.log(data.toString()); // 输出: Hello from Process A

使用消息代理

对于分布式系统，可以使用消息代理如Redis或RabbitMQ实现跨机器的进程通信：

// 使用Redis发布/订阅
const redis = require('redis');
const subscriber = redis.createClient();
const publisher = redis.createClient();

subscriber.on('message', (channel, message) => {
  console.log(`收到消息: ${message} 来自频道: ${channel}`);
});

subscriber.subscribe('my-channel');

// 另一个进程
publisher.publish('my-channel', '这是一条跨进程消息');

进程间通信的错误处理

可靠的IPC需要完善的错误处理机制：

child.on('error', (err) => {
  console.error('子进程错误:', err);
});

child.on('exit', (code, signal) => {
  if (code !== 0) {
    console.warn(`子进程异常退出，代码: ${code}, 信号: ${signal}`);
  }
});

// 超时处理
const timeout = setTimeout(() => {
  child.kill('SIGTERM');
}, 5000);

child.on('exit', () => clearTimeout(timeout));

性能优化与最佳实践

消息批处理

频繁的小消息会降低性能，可以考虑批量发送：

// 不推荐
for (let i = 0; i < 1000; i++) {
  child.send({ index: i });
}

// 推荐
const batch = [];
for (let i = 0; i < 1000; i++) {
  batch.push({ index: i });
}
child.send({ type: 'batch', data: batch });

序列化优化

大型对象的序列化/反序列化会消耗CPU资源：

// 使用更高效的序列化方式
const msgpack = require('msgpack-lite');
child.send(msgpack.encode(largeObject));

process.on('message', (msg) => {
  const data = msgpack.decode(msg);
});

实际应用场景

微服务架构中的IPC

在微服务架构中，不同服务可以通过IPC进行通信：

// service-a.js
const { fork } = require('child_process');
const serviceB = fork('service-b.js');

serviceB.send({ 
  action: 'getUser',
  params: { id: 123 }
});

// service-b.js
process.on('message', async ({ action, params }) => {
  if (action === 'getUser') {
    const user = await db.getUser(params.id);
    process.send({ 
      status: 'success',
      data: user 
    });
  }
});

工作进程池

创建进程池处理高CPU负载任务：

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

if (isMainThread) {
  // 主线程代码
  const workerPool = Array(4).fill().map(() => new Worker(__filename));
  
  workerPool.forEach(worker => {
    worker.on('message', result => {
      console.log('收到结果:', result);
    });
  });
  
  // 分发任务
  workerPool[0].postMessage({ task: 'heavyCalculation', data: 1000 });
} else {
  // 工作线程代码
  parentPort.on('message', ({ task, data }) => {
    if (task === 'heavyCalculation') {
      const result = performHeavyCalculation(data);
      parentPort.postMessage(result);
    }
  });
  
  function performHeavyCalculation(n) {
    // 模拟CPU密集型计算
    let result = 0;
    for (let i = 0; i < n; i++) {
      for (let j = 0; j < n; j++) {
        result += i * j;
      }
    }
    return result;
  }
}

安全注意事项

进程间通信需要考虑安全性问题：

// 验证消息来源
process.on('message', (msg, handle) => {
  if (typeof msg !== 'object' || !msg.type) {
    return; // 忽略无效消息
  }
  
  // 检查发送者是否可信
  if (handle && !isTrustedSender(handle)) {
    console.warn('收到不可信来源的消息');
    return;
  }
  
  // 处理消息...
});

// 限制消息大小
const MAX_MESSAGE_SIZE = 1024 * 1024; // 1MB
child.on('message', (msg) => {
  if (JSON.stringify(msg).length > MAX_MESSAGE_SIZE) {
    child.kill('SIGTERM');
    throw new Error('消息过大');
  }
});

调试与监控

调试IPC通信需要特殊工具和技术：

// 记录所有IPC消息
const debug = require('debug')('ipc');
child.on('message', (msg) => {
  debug('收到消息 %O', msg);
});

// 监控IPC性能
const start = process.hrtime();
child.send('ping');
child.once('message', () => {
  const diff = process.hrtime(start);
  console.log(`IPC往返时间: ${diff[0] * 1e3 + diff[1] / 1e6}ms`);
});

// 使用--inspect-brk调试子进程
const child = fork('child.js', [], {
  execArgv: ['--inspect-brk=9229']
});