了解NodeJs中的多线程

Worker Threads

node 中的多线程是在 node 版本 v10.5.0 引入的一个的一个新特性，在很长一段时间内Worker Thread都是实验性质的，到目前为止，node 稳定的版本已经到了 14.17.4，这个特性已经变成稳定可用了。

worker_thread 是什么

先来来看看官网的描述

The `worker_threads` module enables the use of threads that
execute JavaScript in parallel.

意思就是：worker_thread模块允许使用线程来并行执行 JavaScript。

我们以前 Javascript 都是单线程然后利用一个事件循环队列(event loop)不断监听执行栈是否有函数进入。对于 worker_thread，其实可以理解为一个event loop中有多个 javascript 工作线程，创建一个线程相当于创建一个新的 js 执行环境。多线程的运行如下图

与 child_process 的区别

child_process是可以创建一个新的 node 进程，worker_thread与它的最大区别就是：worker_thread 可以共享内存，公共的数据可以在线程之间公用，而 child_process 只能通过 JSON 去传递数据。

还有就是：因为线程是在一个进程内的，创建一个线程的开销会比创建一个进程要小

worker_thread 的适用范围

worker_thread在CPU 密集型的 JS 操作中非常有用，但是在 IO 密集型操作中性能不会有太多的改善，反而 Node 自带的一步 IO 操作会比工作线程更有用。

使用 worker_thread

介绍完 worker_thread 的概念，现在来介绍一下他的用法

创建工作线程

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

if (isMainThread) {
  createChildThread();
} else {
  parentPort.on('message', (msg) => {
    console.log('parent thread listen:', msg);
  });
  parentPort.postMessage('hello child thread');
}

// 创建线程方法
function createChildThread() {
  const worker = new Worker(__filename);
  worker.on('message', (val) => {
    console.log('child thread listen:', val);
    worker.postMessage('hello parent thread');
  });
}

上面代码在主线程的时候调用创建工作线程的方法，创建一个工作线程并且新增一个，而创建工作线程后调用主线程的端口向工作线程发送消息，工作线程接受到消息后再向主线程回应。大概的流程图如下

执行后返回

创建线程前

创建线程后，线程数量+1

线程间通信

上文提到，worker_threads可以通过ArrayBuffer或SharedArrayBuffer共享内存。接下来看一下，它在代码中是如何实现的.

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

if (isMainThread) {
  createChildThread();
} else {
  // 创建一个长度为4byte的SharedArrayBuffer
  const shareBuf = new SharedArrayBuffer(4);
  // 创建一个8位无符号整型数组
  const bufInt = new Uint8Array(shareBuf);
  parentPort.on('message', () => {
    console.log('parent thread listen:', bufInt);
  });
  // 发送共享内存创建的整型
  parentPort.postMessage({ bufInt });
}

// 创建线程方法
function createChildThread() {
  const worker = new Worker(__filename);
  worker.on('message', ({ bufInt }) => {
    console.log('child thread listen:', bufInt);
    bufInt[0] = 11;
    worker.postMessage('finished');
  });
}

运行后返回，可以看到在子线程创建一个SharedArrayBuffer，用主线程广播的一个数据，在子线程中接收后赋值，因为是线程间共享的 Buffer，所以主线程这边也可以看到在子线程中修改的数据。

如图所示，使用SharedArrayBuffer创建的值会分配到共享内存中，所有线程都可以共用这块内存。

线程间通信

我们已经学会创建线程和使用共享内存了，从上面代码可以看到，线程都是从主线程中发送消息，然后子线程向主线程回复消息，没有办法让两个子线程直接直接通信，如果我想让两个子线程直接通信，那就需要用到MessageChannel这个类了，MessageChannel的具体用法可以点击这里。现在就来实现一下子线程之间的直接通信。

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

if (isMainThread) {
  createChildThread();
} else {
  parentPort.on('message', ({ port }) => {
    // 主线程接收到端口后配置通信端口方法
    port.on('message', (msg) => {
      console.log(`port${threadId} listen:`, msg);
    });
    port.postMessage(`hello, im thread ${threadId}`);
  });
  parentPort.postMessage('hello');
}

// 创建线程方法
function createChildThread() {
  const { port1, port2 } = new MessageChannel(); // 创建一个MessageChannel
  const worker1 = new Worker(__filename); // 创建子线程1
  const worker2 = new Worker(__filename); // 创建子线程2
  worker2.postMessage({ port: port1 }, [port1]); // 向主线程发送Channel的端口1
  worker1.postMessage({ port: port2 }, [port2]); // 向主线程发送Channel的端口1
}

运行代码后返回

从代码实现可以看到，最终建立子线程直接通信的步骤还是在主线程的 message 事件中。建立通信的流程图如下

实战

了解完线程的概念和用法，现在来实战一下：比如在数组中有 100 万条数据需要 md5 加密，对比一下使用工作线程和不使用工作线程的实现速度怎么样

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

const ARR_NUM = 1000000; // 数组长度
const WORKER_NUM = 1; // 线程数
const size = Math.ceil(ARR_NUM / WORKER_NUM); // 每个线程需要处理的数据量

if (isMainThread) {
  createChildThread();
} else {
  parentPort.on('message', ({ status, index, startTime }) => {
    if (index === WORKER_NUM) {
      const usedTime = Date.now() - startTime;
      console.log(`finish bussiness time: ${usedTime}ms`);
      process.exit(threadId);
    }
  });
  const data = addHasCode(threadId, size, (threadId - 1) * size);
  // 完成后
  parentPort.postMessage({
    business: 'finish work',
    data,
  });
}

// 创建线程方法
function createChildThread() {
  let finishNumBuf = new SharedArrayBuffer(4);
  let finishNum = new Uint8Array(finishNumBuf);
  const startTime = Date.now();
  for (let x = 0; x < WORKER_NUM; x++) {
    const worker = new Worker(__filename, {});
    worker.on('message', ({ business, data }) => {
      if (business === 'finish work') {
        finishNum[0]++;
        worker.postMessage({
          status: `finish worker ${x}`,
          index: finishNum[0],
          startTime: startTime,
          data,
        });
      }
    });
  }

  console.log(`${WORKER_NUM} thread start working`);
}

// 加密方法
function addHasCode(index, size, limit) {
  const result = [];
  for (let x = limit, num = index * size; x < num; x++) {
    result.push(createHash('md5').update('hello world').digest('base64'));
  }
  return result;
}