使用Prisma导入CSV至MySQL时重复创建分组问题求助

我需要读取CSV文件并通过Prisma将数据上传至MySQL数据库。目前学生数据写入正常，但「分组存在则复用，不存在则创建」的逻辑失效，持续生成重复分组。尝试了多种方案都没解决，附上相关代码片段：

router.post('/', async (req, res) => {
  fs.createReadStream("./Upload/StudentenEnGroepen001.csv")
  .pipe(parse({ delimiter: ",", from_line: 2 }))
  .on("data", async (row) => {
    let group = null;
    let inschrijving = row[6];
    if (inschrijving === "Student") {
      let parts = row[8].split(",");
      let groep = parts[1];
      let groepNaam = groep.split(" - ")[0].trim();
      const student = await prisma.student.create({
        data: {
            Code: row[0],
            Gebruikersnaam: row[1],
            Familienaam: row[2],
            Voornaam: row[3],
            Sorteernaam: row[4],
            Email: row[5],
        },
      })
      const groupCount = await prisma.groep.count({
        where: {
            Naam: { equals: groepNaam },
        }
      });
      if (groupCount > 0) {
        const existingGroup = await prisma.groep.findFirst({
            where: {
                Naam: { equals: groepNaam },
            }
        });
        group = existingGroup;
      } else {
        group = await prisma.groep.create({
            data: {
                Naam: groepNaam,
            }
        });
      }        
      await prisma.groepstudent.create({
        data: {
          GroepID: group.ID,
          StudentID: student.ID,
        },
      });
      }
  })
  res.json({message: "Studenten en groepen zijn toegevoegd."});
});

我最近试了这段代码，还是无效：

const groupCount = await prisma.groep.count({
    where: {
      Naam: { equals: groepNaam },
    }
  });
  if (groupCount > 0) {
    const existingGroup = await prisma.groep.findFirst({
      where: {
        Naam: { equals: groepNaam },
      }
    });
    group = existingGroup;
  } else {

    group = await prisma.groep.create({
      data: {
        Naam: groepNaam,
      }
    });
  } 

还试过直接用findFirst，同样没效果：

group = await prisma.groep.findFirst({
            where: {
              Naam: { equals: groepNaam },
            }
          });
if (group) {
// 复用该分组
else{
// 创建新分组
...

问题原因及解决办法 #

核心问题：异步流并发导致的竞态条件 #

fs.createReadStream的data事件会异步触发，多个行的处理任务会并行执行。当多个行对应同一个分组名时，可能出现多个请求同时检测到分组不存在，然后各自创建重复分组的情况——这是典型的竞态条件问题。

另外，你在data事件回调中使用async/await，但流的事件处理器不会等待异步操作完成，导致多个异步任务同时运行，进一步加剧了冲突。

解决步骤 #

1. 改用串行处理流数据 + 原子操作upsert
   先收集所有CSV行，再按顺序处理，同时用Prisma的upsert原子操作替代原有的查询+创建逻辑，彻底避免竞态：

   router.post('/', async (req, res) => {
     try {
       // 先收集所有CSV行
       const rows = [];
       await new Promise((resolve, reject) => {
         fs.createReadStream("./Upload/StudentenEnGroepen001.csv")
           .pipe(parse({ delimiter: ",", from_line: 2 }))
           .on("data", (row) => rows.push(row))
           .on("end", resolve)
           .on("error", reject);
       });
   
       // 按顺序处理每一行
       for (const row of rows) {
         let inschrijving = row[6];
         if (inschrijving === "Student") {
           let parts = row[8].split(",");
           let groep = parts[1];
           let groepNaam = groep.split(" - ")[0].trim();
   
           // 创建学生
           const student = await prisma.student.create({
             data: {
               Code: row[0],
               Gebruikersnaam: row[1],
               Familienaam: row[2],
               Voornaam: row[3],
               Sorteernaam: row[4],
               Email: row[5],
             },
           });
   
           // 原子操作：存在则复用，不存在则创建
           const group = await prisma.groep.upsert({
             where: { Naam: groepNaam }, // 需确保Naam字段有唯一约束
             update: {}, // 存在时无需更新
             create: { Naam: groepNaam }, // 不存在时创建
           });
   
           // 关联学生和分组
           await prisma.groepstudent.create({
             data: {
               GroepID: group.ID,
               StudentID: student.ID,
             },
           });
         }
       }
   
       res.json({ message: "Studenten en groepen zijn toegevoegd." });
     } catch (error) {
       console.error(error);
       res.status(500).json({ message: "导入失败", error: error.message });
     }
   });
   

2. 给分组名添加数据库唯一约束
   必须在Prisma Schema的groep模型里给Naam字段添加@unique约束，确保数据库层面拒绝重复值：

   model groep {
     ID    Int     @id @default(autoincrement())
     Naam  String  @unique // 添加唯一约束
     groepstudent groepstudent[]
   }
   

   添加后运行npx prisma migrate dev同步数据库结构。

为什么之前的方案无效？ #

• 原有的count+findFirst/create不是原子操作，并发场景下会出现多个请求同时通过count检测，然后重复创建。
• upsert是数据库层面的原子操作，会保证「查询-创建/更新」的连贯性，彻底避免竞态。
• 流的异步并发处理让多个任务同时操作数据库，即便单任务逻辑正确，也会因并发导致重复，改成串行处理后可消除这个问题。

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