如何使用Sharp实现单读取流多转换并分别写入不同输出流？

Hey there, let's walk through exactly how to pull this off with Sharp. The main hurdle here is that a single ReadableStream can only be consumed once, so we need to create copies of the source stream for each unique transformation. Here's a practical, step-by-step implementation with explanations:

1. Set Up Dependencies #

First, make sure you've got sharp installed, plus Node.js's built-in stream module (no extra install needed):

const sharp = require('sharp');
const { PassThrough, pipeline } = require('stream');

2. Duplicate the Source Stream #

Since we can't read the original stream multiple times, we'll use PassThrough streams to clone the source data. Each transformation will get its own copy of the stream.

function createStreamCopies(sourceStream, copyCount) {
  const streamCopies = Array.from({ length: copyCount }, () => new PassThrough());

  // Forward data, end, and error events to all copies
  sourceStream.on('data', (chunk) => {
    streamCopies.forEach(copy => copy.write(chunk));
  });
  sourceStream.on('end', () => {
    streamCopies.forEach(copy => copy.end());
  });
  sourceStream.on('error', (err) => {
    streamCopies.forEach(copy => copy.emit('error', err));
  });

  return streamCopies;
}

Note: For more robust stream duplication, you could use a package like multistream, but this vanilla approach works great for most use cases.

3. Apply Transformations & Pipe to Outputs #

Take each cloned stream, feed it into Sharp with your desired configuration, then pipe the transformed result to its corresponding output stream. We'll use pipeline to handle error cleanup properly (way better than direct .pipe()).

Let's use three example transformations to illustrate: resize to 300px wide, convert to grayscale, and compress as WebP with 80% quality.

async function processImageTransformations(sourceStream, outputConfigs) {
  // Create one stream copy per output transformation
  const streamCopies = createStreamCopies(sourceStream, outputConfigs.length);

  // Map each copy to its transformation and pipe to the output
  const transformationPromises = streamCopies.map(async (copyStream, index) => {
    const { stream: outputStream, config } = outputConfigs[index];
    let sharpInstance = sharp();

    // Apply transformations based on the config
    if (config.resize) {
      sharpInstance = sharpInstance.resize(config.resize.width);
    }
    if (config.grayscale) {
      sharpInstance = sharpInstance.grayscale();
    }
    if (config.webp) {
      sharpInstance = sharpInstance.webp({ quality: config.webp.quality });
    }
    // Add more transformations as needed (rotate, crop, format conversion, etc.)

    // Use pipeline to manage the stream chain safely
    return new Promise((resolve, reject) => {
      pipeline(
        copyStream,
        sharpInstance,
        outputStream,
        (err) => {
          if (err) {
            console.error(`Transformation ${index + 1} failed:`, err);
            reject(err);
          } else {
            console.log(`Transformation ${index + 1} completed successfully`);
            resolve();
          }
        }
      );
    });
  });

  // Wait for all transformations to finish
  await Promise.all(transformationPromises);
}

4. Usage Example #

Here's how you'd wire this up with a sample source stream and multiple outputs:

const fs = require('fs');

// Example: Source stream (could come from a file, HTTP request, S3, etc.)
const sourceStream = fs.createReadStream('./original-photo.jpg');

// Example: Output streams paired with their transformation configs
const outputConfigs = [
  {
    stream: fs.createWriteStream('./resized-photo.jpg'),
    config: { resize: { width: 300 } }
  },
  {
    stream: fs.createWriteStream('./grayscale-photo.jpg'),
    config: { grayscale: true }
  },
  {
    stream: fs.createWriteStream('./compressed-photo.webp'),
    config: { webp: { quality: 80 } }
  }
];

// Run the transformation pipeline
processImageTransformations(sourceStream, outputConfigs)
  .then(() => console.log('All transformations finished!'))
  .catch(err => console.error('Processing failed:', err));

Key Tips #

• Error Handling: Always use stream.pipeline instead of .pipe()—it automatically cleans up streams and propagates errors across the entire chain, preventing memory leaks.
• Large Images: If you're working with extremely large images, writing the source to a temporary file first (then creating separate read streams from that file) is more memory-efficient than cloning in-memory streams.
• Chainable Transformations: Sharp's methods are fully chainable, so you can combine multiple steps (e.g., resize + grayscale + format) in a single instance for any stream copy.

内容的提问来源于stack exchange，提问作者Daniel Birowsky Popeski