如何在Entity Framework Core中优化关联记录批量插入（不使用EFCore.BulkExtensions）

1. 无付费库时EF Core批量插入最佳实践 #

核心思路：减少数据库往返次数、降低上下文跟踪开销

• 批量添加+单次SaveChanges：放弃逐个Add后调用SaveChanges的方式，用AddRange一次性批量添加实体，最后执行一次SaveChanges：

  var questions = new List<Question>();
  // 批量填充Question数据
  using var context = new YourDbContext();
  context.Questions.AddRange(questions);
  context.SaveChanges();
  

• 禁用变更跟踪优化：批量插入时关闭自动变更检测和实体跟踪，减少上下文的内存开销：

  using var context = new YourDbContext();
  context.ChangeTracker.AutoDetectChangesEnabled = false;
  context.ChangeTracker.QueryTrackingBehavior = QueryTrackingBehavior.NoTracking;
  context.Questions.AddRange(questions);
  context.SaveChanges();
  

• 分批次提交：将400条数据拆分为4-10批（比如每批50条），避免单次提交过大事务导致的锁等待和内存压力，每批使用新的上下文实例进一步降低跟踪开销：

  var batchSize = 50;
  var batches = questions.Chunk(batchSize);
  foreach (var batch in batches)
  {
      using var context = new YourDbContext();
      context.Questions.AddRange(batch);
      context.SaveChanges();
  }
  

• 手动构建参数化批量INSERT SQL：直接用ExecuteSqlRaw执行MySQL支持的多值INSERT语句，性能远高于EF的ORM映射，同时通过参数化避免SQL注入：

  var parameters = new List<object>();
  var valuePlaceholders = new List<string>();
  int paramIndex = 0;
  foreach (var q in questions)
  {
      valuePlaceholders.Add($"(@p{paramIndex++}, @p{paramIndex++})");
      parameters.Add(q.QuestionId);
      parameters.Add(q.Content);
      // 依次添加其他字段参数
  }
  var sql = $"INSERT INTO Question (question_id, content) VALUES {string.Join(", ", valuePlaceholders)}";
  using var context = new YourDbContext();
  context.Database.ExecuteSqlRaw(sql, parameters.ToArray());
  

2. 免费性能提升方案/替代库 #

• Dapper：轻量ORM，直接编写批量SQL，性能接近原生SQL，完全免费。示例：

  using var connection = new MySqlConnection("your_connection_string");
  connection.Open();
  var sql = "INSERT INTO Question (question_id, content) VALUES (@QuestionId, @Content)";
  await connection.ExecuteAsync(sql, questions); // ExecuteAsync自动处理批量逻辑
  

• MySQL LOAD DATA INFILE：如果能将Excel转换为CSV格式，用该命令导入速度最快（比ORM快一个数量级），需要确保数据库用户拥有FILE权限。示例SQL：

  LOAD DATA INFILE '/path/to/questions.csv'
  INTO TABLE Question
  FIELDS TERMINATED BY ',' ENCLOSED BY '"'
  LINES TERMINATED BY '\n'
  (question_id, content, ...);
  

  关联表可先导入主表数据，再用CSV中已有的question_id关联插入。
• 开源EF Core扩展：可以使用EFCore.BulkExtensions的免费开源版本（MIT协议），其基础批量插入功能完全免费，足够覆盖你的场景。

3. 批量场景下question_id高效生成方案 #

核心目标：预生成唯一ID，避免插入后查询ID的数据库往返开销

• 雪花算法批量生成：生成全局唯一的long型ID，适合分布式场景，提前为所有Question生成ID后，关联表直接使用预生成的值。简单实现示例：

  public static class SnowflakeIdGenerator
  {
      private static long _lastTimestamp = -1;
      private static long _sequence = 0;
      private static readonly object _lock = new object();
      private const long Twepoch = 1600000000000L; // 自定义起始时间戳
      private const int WorkerIdBits = 5;
      private const int DatacenterIdBits = 5;
      private const int SequenceBits = 12;
      private const long MaxWorkerId = -1L ^ (-1L << WorkerIdBits);
      private const long MaxDatacenterId = -1L ^ (-1L << DatacenterIdBits);
      private const int WorkerIdShift = SequenceBits;
      private const int DatacenterIdShift = SequenceBits + WorkerIdBits;
      private const int TimestampLeftShift = SequenceBits + WorkerIdBits + DatacenterIdBits;
      private const long SequenceMask = -1L ^ (-1L << SequenceBits);
  
      public static long NextId(long workerId, long datacenterId)
      {
          lock (_lock)
          {
              var timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
              if (timestamp < _lastTimestamp)
                  throw new Exception("时钟回拨异常");
              if (timestamp == _lastTimestamp)
              {
                  _sequence = (_sequence + 1) & SequenceMask;
                  if (_sequence == 0)
                      timestamp = WaitNextTimestamp(_lastTimestamp);
              }
              else
                  _sequence = 0;
              _lastTimestamp = timestamp;
              return ((timestamp - Twepoch) << TimestampLeftShift) 
                     | (datacenterId << DatacenterIdShift) 
                     | (workerId << WorkerIdShift) 
                     | _sequence;
          }
      }
  
      private static long WaitNextTimestamp(long lastTimestamp)
      {
          var timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
          while (timestamp <= lastTimestamp)
              timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
          return timestamp;
      }
  }
  
  // 批量生成400个ID
  var questionIds = Enumerable.Range(0, 400)
                               .Select(_ => SnowflakeIdGenerator.NextId(1, 1))
                               .ToList();
  // 分配给实体及关联表
  for (int i = 0; i < questions.Count; i++)
  {
      questions[i].QuestionId = questionIds[i];
      questionAnswers[i].QuestionId = questionIds[i];
  }
  

• 数据库端预取ID范围：如果不需要分布式唯一ID，可先查询当前最大ID，然后递增生成批量ID，通过事务加锁避免并发冲突：

  using var context = new YourDbContext();
  var transaction = context.Database.BeginTransaction();
  try
  {
      var maxId = context.Questions.Max(q => q.QuestionId);
      var batchIds = Enumerable.Range(0, 400).Select(i => maxId + 1 + i).ToList();
      // 分配ID并执行批量插入
      context.SaveChanges();
      transaction.Commit();
  }
  catch
  {
      transaction.Rollback();
      throw;
  }
  

• 修改表结构为自增列：如果业务允许，将question_id设为MySQL自增列（AUTO_INCREMENT），并在EF实体中配置：

  public class Question
  {
      [Key]
      [DatabaseGenerated(DatabaseGeneratedOption.Identity)]
      public long QuestionId { get; set; }
      // 其他字段
  }
  

  插入后EF会自动获取生成的ID，关联表可直接使用，这是最简单的方案，但需要修改表结构。

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