NodeJS 多线程发展史

// parent.js
const childProcess = require('child_process')
const child = childProcess.fork(__dirname + '/son_process.js')
child.on('message', m => {
  console.log('message from child: ' + JSON.stringify(m));
})
child.send({ from: 'parent' });
// son_process.js
process.on('message', (m) => {
  // you can do anything at here
  console.log('message from parent: ' + JSON.stringify(m));
})
process.send({ from: 'child' });
复制代码

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}`);
});
复制代码

const Koa = require('koa2');
const app = new Koa();
app.use(async (ctx, next) => {
  // 制造随机错误, 子进程会在出错时技=结束
  Math.random() > 0.9 ? func() : 'a'
  await next();
  ctx.response.type = 'text/html';
  ctx.response.body = `<h1>Hello Koa2</h1>`
})
// console.log(module.parent);
// 如果是主模块就监听3000
if (!module.parent) {
  app.listen(3000, () => {
    console.log('listening port 3000');
  })
} else {
  // 不是主模块就导出server代码
  module.exports = app
}
复制代码

const os = require('os');
const cluster = require('cluster');
const process = require('process');
// 计算机CPU核心数
const cpus = os.cpus().length;
console.log("cpus: ", cpus);
// map的形式记录worker及对应的进程pid {8035: Worker}
const workers = {};
// 主进程用来管理子进程
if (cluster.isMaster) {
  // 在主进程添加death事件监听，当线程挂掉时重启一个线程
  cluster.on('death', worker => {
    // 创建子进程
    worker = cluster.fork();
    workers[worker.pid] = worker;
  });
  // 根据CPU数量来建立进程
  for (let i = 0; i < cpus; i++) {
    const worker = cluster.fork();
    workers[worker.pid] = worker;
  }
} else {
  // 子进程用来执行server端逻辑
  const app = require('./app');
  app.use(async (ctx, next) => {
    console.log('worker' + cluster.worker.id + 'PID: ' + process.pid);
  })
  app.listen(3000)
}
// 结束时kill掉workers中保存的所有进程
process.on('SIGTERM', () => {
  for (let pid in workers) {
    process.kill(pid)
  }
  process.exit(0)
})
// 引入测试文件
require('./test')
// test.js
// 自动化测试，每秒钟请求接口
const http = require('http');
setInterval(async () => {
  try {
    await http.get('http://localhost:3000')
  } catch (error) {
    console.log(error);
  }
}, 1000)
复制代码

const { Worker, isMainThread, parentPort } = require('worker_threads');
// isMainThread: 该对象用于区分是主线程（true）还是工作线程（false）
if (isMainThread) {
  // Worker对象 必填参数 __filename（文件路径），该文件会被 worker 执行
  const worker = new Worker(__filename);
  // 工作进程监听 message 事件
  worker.on('message', (msg) => { console.log(msg); });
} else {
  // parentPort: 该对象的 postMessage 方法用于 worker 线程向主线程发送消息
  parentPort.postMessage('Hello world!');
}
复制代码

const fib = (times) => {
  if (times === 0) return 0;
  else if (times === 1) return 1;
  else return fib(times - 1) + fib(times - 2)
}
// 记录当前时间戳
const now = new Date().getTime();
const result1 = fib(40);
console.log('result1用时:' + (new Date().getTime() - now));
const result2 = fib(40);
console.log('result2用时:' + (new Date().getTime() - now));
const result3 = fib(40);
console.log('result3用时:' + (new Date().getTime() - now));
复制代码

const { Worker,
  isMainThread,
  parentPort,
  workerData
} = require('worker_threads');
if (isMainThread) {
  const now = new Date().getTime();
  const worker1 = new Worker(__filename, { workerData: 40 })
  worker1.on('message', () => {
    console.log('result1用时:' + (new Date().getTime() - now))
  })
  const worker2 = new Worker(__filename, { workerData: 40 })
  worker2.on('message', () => {
    console.log('result2用时:' + (new Date().getTime() - now))
  })
  const worker3 = new Worker(__filename, { workerData: 40 })
  worker3.on('message', () => {
    console.log('result3用时:' + (new Date().getTime() - now))
  })
} else {
  const fib = (times) => {
    if (times === 0) return 0;
    else if (times === 1) return 1;
    else return fib(times - 1) + fib(times - 2)
  }
  const times = workerData
  const result = fib(times)
  parentPort.postMessage(result)
}
复制代码