Node.js+Fastify使用@fastify/multipart上传CSV解析为空数组的解决方法

问题场景 #

使用Node.js、Fastify、@fastify/multipart及fast-csv实现CSV文件上传解析功能时，插件打印解析后的数据始终是空数组[]。以下是相关代码：

插件代码 #

instance.decorate('uploadCSV', async (parts: Multipart[]) => {
const uploads = parts.map(async (part: Multipart) => {
  try {
    if (part.file) {
      const mimetype = part.mimetype;
      const fileName = part.filename.split(' ').join('');

      if (!fileName.toLowerCase().endsWith('.csv') || mimetype !== 'text/csv') {
        throw instance.httpErrors.badRequest('file.not.csv');
      }

      const uploadDir = path.join(__dirname, 'uploads');

      // Create the 'uploads' directory if it doesn't exist
      if (!fs.existsSync(uploadDir)) {
        fs.mkdirSync(uploadDir);
      }

      const filePath = path.join(uploadDir, fileName);

      // Save the file directly using fs.promises.writeFile
      await fs.promises.writeFile(filePath, part.file, 'utf-8');

      // Parse the CSV file
      const data = await new Promise<any[]>((resolve, reject) => {
        const dataArray: any[] = [];
        fs.createReadStream(filePath)
          .pipe(csv.parse({ headers: true }))
          .on('data', (row) => dataArray.push(row))
          .on('end', () => resolve(dataArray))
          .on('error', reject);
      });

      // Print the CSV data
      console.log('CSV Data:', data);

      // You can now do further processing with the parsed data

      return { success: true, message: 'CSV file uploaded and processed successfully' };
    }
  } catch (error) {
    throw error;
  }
});

// Wait for all uploads to complete
await Promise.all(uploads);
});

调用代码 #

const { file } = request.body
await instance.uploadCSV(file)

问题原因 #

1. 文件流处理错误：@fastify/multipart返回的part.file是Readable流，直接传入fs.promises.writeFile无法正确读取流内容，导致保存的文件为空，解析自然得到空数组。
2. 文件获取方式错误：调用代码中从request.body获取file不符合@fastify/multipart的使用规范，该插件不会将文件挂载到request.body上，需要通过request.parts()或request.files()获取文件部件。
3. 目录创建存在竞态风险：使用fs.existsSync+fs.mkdirSync的组合在并发场景下可能出错，不如递归创建目录可靠。

解决方案 #

1. 修正文件保存逻辑（用stream.pipeline处理流） #

替换原有的fs.promises.writeFile，使用stream.pipeline来正确消费文件流并写入本地：

// 导入stream模块
import { pipeline } from 'stream/promises';

// 替换原保存文件的代码
await pipeline(
  part.file,
  fs.createWriteStream(filePath)
);

2. 修正调用代码，正确获取文件部件 #

使用@fastify/multipart提供的request.parts()方法获取上传的文件：

// 在路由处理函数中
const parts = [];
for await (const part of request.parts()) {
  if (part.file) {
    parts.push(part);
  }
}
await instance.uploadCSV(parts);

3. 优化目录创建逻辑 #

用递归创建目录替代existsSync+mkdirSync，避免竞态问题：

const uploadDir = path.join(__dirname, 'uploads');
await fs.promises.mkdir(uploadDir, { recursive: true });

完整修正后的插件代码 #

import { pipeline } from 'stream/promises';
import * as fs from 'fs';
import * as path from 'path';
import type { Multipart } from '@fastify/multipart';

instance.decorate('uploadCSV', async (parts: Multipart[]) => {
  const uploads = parts.map(async (part: Multipart) => {
    try {
      if (!part.file) return;

      const mimetype = part.mimetype;
      const fileName = part.filename.split(' ').join('');

      if (!fileName.toLowerCase().endsWith('.csv') || mimetype !== 'text/csv') {
        throw instance.httpErrors.badRequest('file.not.csv');
      }

      const uploadDir = path.join(__dirname, 'uploads');
      await fs.promises.mkdir(uploadDir, { recursive: true });

      const filePath = path.join(uploadDir, fileName);
      await pipeline(part.file, fs.createWriteStream(filePath));

      const data = await new Promise<any[]>((resolve, reject) => {
        const dataArray: any[] = [];
        fs.createReadStream(filePath)
          .pipe(csv.parse({ headers: true }))
          .on('data', (row) => dataArray.push(row))
          .on('end', () => resolve(dataArray))
          .on('error', reject);
      });

      console.log('CSV Data:', data);
      return { success: true, message: 'CSV file uploaded and processed successfully' };
    } catch (error) {
      throw error;
    }
  });

  await Promise.all(uploads);
});

额外提示 #

• 可以考虑直接解析流而不保存到本地，减少IO操作：直接将part.file管道到csv.parse，无需先写入文件再读取。
• 注意处理大文件场景，避免内存溢出，保持流式处理的优势。

内容的提问来源于stack exchange，提问作者user10596155