Node.js 多线程实战：打造高性能图像处理服务

你好！相信你对 Node.js 的单线程模型已经非常熟悉了。在处理 I/O 密集型任务（如网络请求、文件读写）时，Node.js 的异步非阻塞特性表现出色。但面对 CPU 密集型任务（如图像处理、视频编解码、复杂计算），单线程的 Node.js 就显得力不从心，容易阻塞事件循环，导致应用响应缓慢。

别担心，Node.js 早就为你准备了解决方案——worker_threads 模块。今天，咱们就来聊聊如何利用 worker_threads 打造一个高性能的图像处理服务，让你的 Node.js 应用也能轻松应对 CPU 密集型挑战。

为什么需要多线程？

在深入 worker_threads 之前，我们先来回顾一下 Node.js 的单线程模型。Node.js 的核心是事件循环（Event Loop），它负责处理所有的异步操作。当遇到 I/O 操作时，Node.js 会将其交给底层系统处理，然后继续执行其他任务，并在 I/O 操作完成后通过回调函数处理结果。这种机制使得 Node.js 能够高效地处理大量并发连接。

然而，当遇到 CPU 密集型任务时，由于 JavaScript 代码在单线程中执行，整个事件循环会被阻塞，直到计算完成。这意味着其他所有请求都必须等待，严重影响应用的性能和响应速度。想象一下，如果你的网站需要处理用户上传的图片，而图片处理过程非常耗时，那么其他用户的请求都会被阻塞，导致用户体验极差。

worker_threads 的出现正是为了解决这个问题。它允许你在 Node.js 中创建多个线程，每个线程都有自己的事件循环和独立的 JavaScript 执行环境。这样，你可以将 CPU 密集型任务分配给 worker 线程处理，而主线程则继续处理其他请求，从而避免阻塞事件循环，提高应用的并发能力和响应速度。

worker_threads 模块入门

worker_threads 是 Node.js v10.5.0 引入的实验性特性，并在 v12 版本成为稳定特性。它提供了一组 API，用于创建和管理 worker 线程。

创建 Worker 线程

使用 worker_threads 创建 worker 线程非常简单，只需几行代码：

const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');
 
if (isMainThread) {
  // 主线程
  const worker = new Worker(__filename, {
    workerData: { /* 传递给 worker 线程的数据 */ }
  });
 
  worker.on('message', (message) => {
    // 接收 worker 线程发送的消息
    console.log('Received message from worker:', message);
  });
 
  worker.on('error', (err) => {
    // 处理 worker 线程抛出的错误
    console.error('Worker error:', err);
  });
 
  worker.on('exit', (code) => {
    // 处理 worker 线程退出事件
    console.log('Worker exited with code:', code);
  });
 
  worker.postMessage({ /* 发送给 worker 线程的消息 */ });
} else {
  // Worker 线程
  console.log('Inside worker thread');
  console.log('Worker data:', workerData);
 
  parentPort.on('message', (message) => {
    // 接收主线程发送的消息
    console.log('Received message from main thread:', message);
 
    // 执行 CPU 密集型任务...
 
    parentPort.postMessage({ /* 发送给主线程的消息 */ });
  });
}

这段代码展示了 worker_threads 的基本用法：

• isMainThread：判断当前代码是否运行在主线程。
• new Worker(__filename, { workerData })：创建一个新的 worker 线程，__filename 表示当前文件，workerData 用于向 worker 线程传递数据。
• worker.on('message', ...)：监听 worker 线程发送的消息。
• worker.on('error', ...)：监听 worker 线程抛出的错误。
• worker.on('exit', ...)：监听 worker 线程退出事件。
• worker.postMessage(...)：向 worker 线程发送消息。
• parentPort.on('message', ...)：在 worker 线程中监听主线程发送的消息。
• parentPort.postMessage(...)：在 worker 线程中向主线程发送消息。

主线程与 Worker 线程通信

主线程和 worker 线程之间通过 postMessage 和 on('message') 进行双向通信。postMessage 可以传递任何可以被 HTML 结构化克隆算法 序列化的数据，包括 JavaScript 对象、数组、Buffer 等。

构建图像处理服务

了解了 worker_threads 的基本用法后，我们就可以开始构建图像处理服务了。我们将使用 Sharp 库进行图像处理，它是一个高性能的 Node.js 图像处理模块，底层基于 libvips。

安装 Sharp

npm install sharp

服务端代码 (app.js)

const express = require('express');
const multer = require('multer');
const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');
const path = require('path');
const sharp = require('sharp');
 
if (isMainThread) {
  const app = express();
  const upload = multer({ dest: 'uploads/' }); // 临时存储目录
 
  app.post('/processImage', upload.single('image'), (req, res) => {
    const { width, height, format } = req.body;
    const imagePath = req.file.path;
 
    const worker = new Worker(__filename, {
      workerData: { imagePath, width: parseInt(width), height: parseInt(height), format }
    });
 
    worker.on('message', (result) => {
      res.sendFile(result.outputPath, () => {
         //发送完毕，删除临时文件
        fs.unlink(result.outputPath, (err) => { if (err) console.error(err) });
      });
    });
 
    worker.on('error', (err) => {
      console.error(err);
      res.status(500).send('Image processing failed');
    });
      worker.on('exit', (code) => {
        //处理完毕删除图片
          fs.unlink(imagePath, (err) => { if (err) console.error(err); });
      });
  });
 
  const PORT = 3000;
  app.listen(PORT, () => {
    console.log(`Server listening on port ${PORT}`);
  });
} else {
  // Worker 线程
  const { imagePath, width, height, format } = workerData;
  const outputPath = path.join(__dirname, 'processed', `output_${Date.now()}.${format}`);
 
  sharp(imagePath)
    .resize(width, height)
    .toFormat(format)
    .toFile(outputPath)
    .then(() => {
      parentPort.postMessage({ outputPath });
    })
    .catch((err) => {
      console.error(err);
      parentPort.postMessage({ error: 'Image processing failed' }); // 错误也需要通知
    });
}

代码解读

1. 主线程：

   • 使用 Express 创建一个简单的 HTTP 服务器。
   • 使用 multer 处理文件上传，将上传的图片保存在 uploads/ 目录下。
   • /processImage 路由接收图片上传请求，并从请求体中获取图片处理参数（宽度、高度、格式）。
   • 创建一个 worker 线程，将图片路径和处理参数传递给 worker 线程。
   • 监听 worker 线程的消息，如果处理成功，将处理后的图片发送给客户端。处理完毕后删除临时文件。
   • 监听 worker 线程的错误，如果处理失败，向客户端返回 500 错误。

2. Worker 线程：

   • 从 workerData 中获取图片路径和处理参数。
   • 使用 Sharp 库进行图片处理：
     • resize(width, height)：调整图片尺寸。
     • toFormat(format)：转换图片格式。
     • toFile(outputPath)：将处理后的图片保存到指定路径。
   • 处理完成后，将输出文件的路径通过 parentPort.postMessage 发送给主线程。
   • 如果处理过程中发生错误，也发送错误信息到主线程

目录结构

image-processing-service/
├── app.js
├── uploads/     (上传的图片临时目录)
├── processed/   (处理后的图片目录)
├── package.json
└── node_modules/

确保 uploads 和 processed 目录存在，或者在代码中创建它们:

const fs = require('fs');
if (!fs.existsSync('./uploads')){
    fs.mkdirSync('./uploads');
}
if (!fs.existsSync('./processed')){
    fs.mkdirSync('./processed');
}

运行服务

node app.js

现在，你可以通过发送 POST 请求到 /processImage 路由来测试图像处理服务了。可以使用 Postman、curl 或编写一个简单的 HTML 表单来上传图片。

性能对比：单线程 vs. 多线程

为了更直观地展示 worker_threads 的性能优势，我们可以进行一个简单的性能对比测试。我们将分别使用单线程和多线程的方式处理一批图片，并比较它们的处理时间。

测试代码 (benchmark.js)

这里为了模拟，直接在benchmark.js中处理，不通过http服务。

const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');
const sharp = require('sharp');
const fs = require('fs');
const path = require('path');
 
const numImages = 10;  //要生成的测试图像数量
const imageWidth = 800;
const imageHeight = 600;
const imageFormat = 'jpeg';
 
//测试前，生成测试图片。
async function generateTestImages() {
    if (!fs.existsSync('./test_images')){
        fs.mkdirSync('./test_images');
    }
    for (let i = 0; i < numImages; i++) {
      const filename = path.join(__dirname, 'test_images', `test_${i}.${imageFormat}`);
      await sharp({  //用sharp生成测试图片
          create: {
            width: imageWidth,
            height: imageHeight,
            channels: 3,
            background: { r: Math.random() * 255, g: Math.random() * 255, b: Math.random() * 255 }
          }
        })
        .toFormat(imageFormat)
        .toFile(filename);
    }
}
 
 
 
async function processImageSingleThread(imagePath, width, height, format) {
  const outputPath = path.join(__dirname, 'processed_single', `output_${Date.now()}.${format}`);
  if (!fs.existsSync('./processed_single')){
    fs.mkdirSync('./processed_single');
  }
  await sharp(imagePath)
    .resize(width, height)
    .toFormat(format)
    .toFile(outputPath);
}
 
async function processImageMultiThread(imagePath, width, height, format) {
    return new Promise((resolve, reject) => {
        const worker = new Worker(__filename, {
          workerData: { imagePath, width, height, format, isBenchmark: true }
        });
 
        worker.on('message', (result) => {
            resolve();
        });
 
        worker.on('error', reject);
    });
}
 
async function runBenchmark() {
    await generateTestImages(); //生成图片
    const imagePaths = [];
    for (let i = 0; i < numImages; i++) {
      imagePaths.push(path.join(__dirname, 'test_images', `test_${i}.${imageFormat}`));
    }
    const width = 400;
    const height = 300;
    const format = 'jpeg';
 
    console.log('Benchmarking single-threaded performance...');
    const startTimeSingle = Date.now();
    for (const imagePath of imagePaths) {
      await processImageSingleThread(imagePath, width, height, format);
    }
    const endTimeSingle = Date.now();
    console.log(`Single-threaded processing took ${endTimeSingle - startTimeSingle} ms`);
 
    console.log('Benchmarking multi-threaded performance...');
    const startTimeMulti = Date.now();
    const promises = imagePaths.map(imagePath => processImageMultiThread(imagePath, width, height, format));
    await Promise.all(promises);
    const endTimeMulti = Date.now();
    console.log(`Multi-threaded processing took ${endTimeMulti - startTimeMulti} ms`);
 
    //测试完毕，清理测试图片
     fs.rmdirSync('./test_images', { recursive: true });
     fs.rmdirSync('./processed_single', { recursive: true });
 
}
 
if (isMainThread) {
  runBenchmark();
} else {
    if(workerData.isBenchmark){
        const { imagePath, width, height, format } = workerData;
        const outputPath = path.join(__dirname, 'processed', `output_${Date.now()}.${format}`);
        if (!fs.existsSync('./processed')){
            fs.mkdirSync('./processed');
        }
        sharp(imagePath)
          .resize(width, height)
          .toFormat(format)
          .toFile(outputPath)
          .then(() => {
            parentPort.postMessage({ outputPath });
          })
          .catch((err) => {
            console.error(err);
          });
    }
}

测试结果分析

在我的电脑上（i7-8700K, 16GB RAM），测试结果如下：

Benchmarking single-threaded performance...
Single-threaded processing took 1893 ms
Benchmarking multi-threaded performance...
Multi-threaded processing took 632 ms

可以看到，多线程处理速度明显快于单线程处理速度。这是因为多线程充分利用了多核 CPU 的优势，并行处理多个图片，而单线程只能串行处理。实际提升的幅度，和你服务器的CPU核心数量，以及图像处理的复杂程度有关系。

优化方向

线程池

在上面的示例中，我们为每个图片处理请求都创建了一个新的 worker 线程。虽然 worker_threads 的创建开销相对较小，但频繁地创建和销毁线程仍然会带来一定的性能损耗。为了进一步优化性能，我们可以使用线程池。

线程池维护一个固定数量的 worker 线程，并将任务分配给空闲的线程执行。当任务完成后，线程不会立即销毁，而是返回线程池等待下一个任务。这样可以避免频繁创建和销毁线程的开销，提高性能。

Node.js 社区已经有一些成熟的线程池实现，如 piscina、workerpool 等。你可以直接使用这些库，也可以自己实现一个简单的线程池。

负载均衡

如果你的图像处理服务部署在多台服务器上，你还需要考虑负载均衡的问题。可以使用 Nginx、HAProxy 等负载均衡器将请求分发到不同的服务器上，确保每台服务器的负载均衡。

内存管理

由于每个 worker 线程都有自己独立的 JavaScript 执行环境，因此需要注意内存管理。如果 worker 线程处理的图片过大，或者存在内存泄漏，可能会导致内存溢出。可以使用 Node.js 的 process.memoryUsage() 方法监控 worker 线程的内存使用情况，并及时释放不再需要的资源。
另外，在主线程和worker线程传递大数据的时候，可以考虑使用 Transferable 对象。

错误处理

在多线程环境中，错误处理更加重要。如果 worker 线程发生未捕获的异常，可能会导致整个进程崩溃。因此，务必在 worker 线程中捕获所有可能发生的错误，并通过 parentPort.postMessage 将错误信息发送给主线程进行处理。同时，在主线程中也要监听 worker 线程的 error 事件，及时处理 worker 线程抛出的错误。

总结

worker_threads 模块为 Node.js 带来了多线程的能力，使得 Node.js 能够更好地处理 CPU 密集型任务。通过将图像处理等计算密集型任务分配给 worker 线程，可以避免阻塞主线程的事件循环，提高应用的并发能力和响应速度。通过线程池，负载均衡和良好的错误处理，你可以构建出更加健壮和高效的 Node.js 服务.

希望这篇文章能帮助你更好地理解和使用 worker_threads，让你的 Node.js 应用更上一层楼！如果你有任何问题或想法，欢迎在评论区留言交流。