是否可以使用 Node 工作线程来执行数据库插入？

Posted 2023-03-07

【中文标题】是否可以使用 Node 工作线程来执行数据库插入？

【英文标题】：Is it possible to use Node worker threads to perform database inserts?

【发布时间】：2020-11-05 12:38:54

【问题描述】：

我最近读到了 Node 的“worker_threads”模块，它允许在多个线程中并行执行 javascript 代码，这对于 CPU 密集型操作很有用。 （注意：这些不是 Chrome 在浏览器中制作的网络工作者）

我正在构建一个功能，我需要在不阻塞浏览器的情况下执行大量 Postgres INSERT。

问题是： 在我实例化 worker 的 Javascript 文件中，我不允许导入任何内容，包括本地 Node 模块或 NPM 库（如 Knex.js），这是执行数据库所必需的查询。我收到一条错误消息：Cannot use import statement outside a module 只要文件被执行。

我尝试将工作代码放在另一个文件中，并在顶部使用 import 语句（同样的错误）。我尝试将 Knex 对象提供给 workerData，但它无法克隆非原生 JS 对象。

我没有想法 - 如果我们无法导入任何 NPM 库，有谁知道如何在工作线程中与数据库交互？！？！

// mainThread.js

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

import knex from 'knex'; // --> *** UNCAUGHT EXCEPTION: Cannot use import statement outside a module ***

if (isMainThread) 
  module.exports = async function runWorker (rowsToInsert = []) 
    return new Promise((resolve, reject) => 
      const worker = new Worker(__filename,  workerData:  rowsToInsert  );

      worker.on('message', (returningRows) => resolve(returningRows));
      worker.on('error', reject);
      worker.on('exit', (code) => 
        if (code !== 0) reject(new Error(`Worker stopped with exit code $code`));
      );
    );
  ;
 else 
  const  rowsToInsert  = workerData;

  return knex('table').insert(rowsToInsert)
    .then((returningRows) => 
      parentPort.postMessage( data: returningRows );
    );

我正在关注这个网页的教程：https://blog.logrocket.com/use-cases-for-node-workers/

【问题讨论】：

是的，您可以为此使用 workerThreads，但它可能不是必需的或有用的。数据库本身就是它自己的进程。因此，所有 nodejs 进程正在做的就是以非阻塞、异步的方式向它发送网络请求。 nodejs 不太可能做任何会从 workerThreads 中受益的 CPU 密集型操作。

仅供参考，在编写 workerThread 代码时，如果您将 workerThread 代码放在自己的文件中并且不要将主线程代码与 workerThread 代码混合在同一个文件中，我发现它 200% 更清晰、更清晰、更容易调试.这样做时你不应该需要if (isMainThread) 。此外，当您将代码清晰地分成两个文件时，对代码在哪里运行的整体知识理解会更容易理解。

感谢您的解释！我感觉合理！ @jfriend00

@jfriend00 顺便说一句，如果您有任何其他见解 - 在尝试使用 Node 进行大容量插入（例如，每次摄取最多 100 万行）时，我可以使用哪些其他工具？

nodejs 不是你的挑战。挑战将是对数据库进行 100 万次插入。数据库将是那里的瓶颈，而不是 nodejs。如需进一步帮助，您可能想就您正在使用的特定数据库以及您要插入的具体内容提出一个新问题，然后也许真正了解该数据库的人可以谈论最有效的获取方法所有这些数据都进入数据库。同样，这并不是真正的 nodejs 问题，而是如何以最佳方式使用数据库。

【参考方案1】：

这当然是可能的，但这是一个非常糟糕的主意。

数据库驱动程序已经是异步且非阻塞的 JavaScript 线程。按照您的建议将插入调用移至单独的线程不仅不会提高性能，而且实际上会降低整体性能，因为线程间通信涉及开销：

同步和消息传递不是免费的 JavaScript 在线程之间移动数据时使用structured cloning。这意味着您的所有rowsToInsert 都必须被复制，这是（相对）昂贵的。

通常，真正适合使用 JS 线程的唯一时间是您的 JavaScript 代码 正在执行 CPU 密集型工作。 node docs say as much right at the top：

Worker（线程）对于执行 CPU 密集型 JavaScript 操作很有用。它们对 I/O 密集型工作没有多大帮助。 Node.js 内置的异步 I/O 操作比 Workers 效率更高。

这意味着，如果您要进行大量的解析、数学运算或类似工作，那么在线程中完成这些工作可能是合适的。然而，简单地将数据从一个地方铲到另一个地方（即 I/O）并不是线程的理想选择——毕竟，节点的设计经过调整以在这种工作中高效。

你没有说你的rowsToInsert 来自哪里，但如果它来自 HTTP 请求，那么使用线程是错误的。但是，如果您在服务器上解析例如 CSV 或 JSON 文件，可能值得在线程中执行此操作，但线程完成所有工作很重要（因此内存不需要在线程之间移动）。您发布给工作人员的消息应该只是“处理位于 /foo/bar.csv 的文件”，然后工作线程完成其余的工作。

---

您得到的错误是the same that you'd get without worker threads：您正尝试在常规的非模块JS 文件中使用import。将工作文件重命名为 *.mjs 或改用 require('knex')。

节点的ES module documentation 详细介绍了import 与require 的不同之处。

【讨论】：

感谢您的精彩回答！非常清楚，您使用require 的提示有效。因此，如果我想做大批量插入，有哪些工具可用？我应该使用 Node 集群吗？我正在谷歌搜索“大量插入节点”或“节点集群 postgres 插入”，但没有得到明确的答案。

这取决于您的数据来自哪里。它来自哪里？

数据来自 Node.js 解析多个 CSV 文件。我遍历数据并创建一个用于插入的 JS 对象数组。目前，我将每个块的插入分块为 10K 行。我正在运行 Promises 将它们一个接一个地插入 Postgres（而不是同时使用 Promise.all）

@BruceWang：如果您的输入数据已经格式良好，那么您最好的选择可能是COPY，它将 CSV 本地导入到表格中。 pg-copy-streams 会有所帮助。但是，如果您在节点中进行大量解析和转换，那么可能在工作线程或进程中完成这项工作是有意义的，但我会分析以找到真正的瓶颈（ s) 首先。另请关注prepared statements。

……一般情况下，请阅读the pg docs' performance tips。