EntityFramework 中支持 BulkInsert 擴展

本文為 Dennis Gao 原創技術文章，發表於博客園博客，未經作者本人允許禁止任何形式的轉載。

前言

很顯然，你應該不至於使用 EntityFramework 直接插入 10W 數據到數據庫中，那可能得用上個幾分鍾。EntityFramework 最被人詬病的地方就是它的性能，處理大量數據時的效率。此種條件下，通常會轉回使用 ADO.NET 來完成任務。

但是，如果已經在項目中使用了 EntityFramework，如果碰到需要直接向數據庫中插入 10W 的數據的需求，引入 ADO.NET 和 SqlBulkCopy 的組合將打破 EntityFramework 作為 ORM 所帶來的優勢，我們不得不再次去編寫那些 SQL 語句，關注表結構的細節，相應的代碼可維護性也在下降。

那么，假設我們將 SqlBulkCopy 的功能封裝為 EntityFramework 中的一個擴展方法，通過接口像外暴露 BulkInsert 方法。這樣，我們既沒有改變使用 EntityFramework 的習慣，同時也隱藏了 SqlBulkCopy 的代碼細節，更重要的是，合理的封裝演進出復用的可能性，可以在多個 Entity 表中使用。

環境准備

以下測試基於 EntityFramework 6.0.2 版本。

首先定義一個 Customer 類：

  public class Customer
  {
    public long Id { get; set; }
    public string Name { get; set; }
    public string Address { get; set; }
    public string Phone { get; set; }
  }

通過 CustomerMap 類將 Entity 映射到數據庫表結構：

  public class CustomerMap : EntityTypeConfiguration<Customer>
  {
    public CustomerMap()
    {
      // Primary Key
      this.HasKey(t => t.Id);

      // Properties
      this.Property(t => t.Name)
          .IsRequired()
          .HasMaxLength(256);

      this.Property(t => t.Phone)
          .IsRequired()
          .HasMaxLength(256);

      // Table & Column Mappings
      this.ToTable("Customer", "STORE");
      this.Property(t => t.Id).HasColumnName("Id");
      this.Property(t => t.Name).HasColumnName("Name");
      this.Property(t => t.Address).HasColumnName("Address");
      this.Property(t => t.Phone).HasColumnName("Phone");
    }
  }

我們定義數據庫的名字為 “Retail”，則使用 RetailEntities 類來實現 DbContext ：

  public class RetailEntities : DbContext
  {
    static RetailEntities()
    {
      Database.SetInitializer<RetailEntities>(
        new DropCreateDatabaseAlways<RetailEntities>());
    }

    public RetailEntities()
      : base("Name=RetailEntities")
    {
    }

    public DbSet<Customer> Customers { get; set; }

    protected override void OnModelCreating(DbModelBuilder modelBuilder)
    {
      modelBuilder.Configurations.Add(new CustomerMap());
    }
  }

將 DatabaseInitializer 設置為 DropCreateDatabaseAlways，這樣我們可以保證每次都針對新表進行測試。

如果需要更復雜的模型，我們將基於如下的模型進行測試：

測試主機

數據庫：Microsoft SQL Server 2012 (64-bit)

EntityFramework 插入 10W 數據需要多久

我們先來看下EntityFramework 插入 10W 數據需要多久。

構造 10W 個 Customer 實例：

      int customerCount = 100000;

      List<Customer> customers = new List<Customer>();
      for (int i = 0; i < customerCount; i++)
      {
        Customer customer = new Customer()
        {
          Name = "Dennis Gao" + i,
          Address = "Beijing" + i,
          Phone = "18888888888" + i,
        };
        customers.Add(customer);

        Console.Write(".");
      }

使用如下語法來將上面構造的 10W 數據保存到數據庫中：

        using (RetailEntities context = new RetailEntities())
        {
          foreach (var entity in customers)
          {
            context.Customers.Add(entity);
          }
          context.SaveChanges();
        }

通過 context.SaveChanges() 來保證一次事務提交。

為了計算使用時間，在上面代碼的前后加上 Stopwatch 來計算：

        Stopwatch watch = Stopwatch.StartNew();

        using (RetailEntities context = new RetailEntities())
        {
          foreach (var entity in customers)
          {
            context.Customers.Add(entity);
          }
          context.SaveChanges();
        }

        watch.Stop();
        Console.WriteLine(string.Format(
          "{0} customers are created, cost {1} milliseconds.", 
          customerCount, watch.ElapsedMilliseconds));

然后運行，

好吧，我應該沒有耐心等待它運行完。

現在減少數據量進行測試，將數據數量降低到 500 條，

空表插入500 條數據耗時 5652 毫秒。我多測了幾遍，這個數據穩定在 5 秒以上。

將數據量改變到 1000 條，

將數據量改變到 1500 條，

將數據量改變到 10000 條，

那么我們估計下 10W 數據大概需要 10W / 500 * 2 = 至少 400 秒 = 至少 6 分鍾。

好吧，慢是毋庸置疑的。

SqlBulkCopy 接口描述

Microsoft SQL Server 提供一個稱為 bcp 的流行的命令提示符實用工具，用於將數據從一個表移動到另一個表（表既可以在同一個服務器上，也可以在不同服務器上）。 SqlBulkCopy 類允許編寫提供類似功能的托管代碼解決方案。 還有其他將數據加載到 SQL Server 表的方法（例如 INSERT 語句），但相比之下 SqlBulkCopy 提供明顯的性能優勢。

使用 SqlBulkCopy 類只能向 SQL Server 表寫入數據。 但是，數據源不限於 SQL Server；可以使用任何數據源，只要數據可加載到 DataTable 實例或可使用 IDataReader 實例讀取數據。

WriteToServer(DataRow[])	將所提供的 DataRow 數組中的所有行復制到 SqlBulkCopy 對象的 DestinationTableName 屬性指定的目標表中。
WriteToServer(DataTable)	將所提供的 DataTable 中的所有行復制到 SqlBulkCopy 對象的 DestinationTableName 屬性指定的目標表中。
WriteToServer(IDataReader)	將所提供的 IDataReader 中的所有行復制到 SqlBulkCopy 對象的 DestinationTableName 屬性指定的目標表中。
WriteToServer(DataTable, DataRowState)	只將與所提供 DataTable 中所提供行狀態匹配的行復制到 SqlBulkCopy 對象的 DestinationTableName 屬性指定的目標表中。

在 .NET 4.5 中還提供了支持 async 語法的接口。

WriteToServerAsync(DataRow[])	WriteToServer 的異步版本，將 DataRow 數組中提供的所有行都復制到由 SqlBulkCopy 對象的 DestinationTableName 屬性指定的目標表中。
WriteToServerAsync(DataTable)	WriteToServer 的異步版本，將 DataTable 中提供的所有行都復制到 SqlBulkCopy 對象的 DestinationTableName 屬性指定的目標表中。
WriteToServerAsync(IDataReader)	WriteToServer 的異步版本，將 IDataReader 中提供的所有行都復制到 SqlBulkCopy 對象的 DestinationTableName 屬性指定的目標表中。
WriteToServerAsync(DataRow[], CancellationToken)	WriteToServer 的異步版本，將 DataRow 數組中提供的所有行都復制到由 SqlBulkCopy 對象的 DestinationTableName 屬性指定的目標表中。取消標記可用於在命令超時超過前請求放棄操作。通過返回的任務對象將報告異常。
WriteToServerAsync(DataTable, DataRowState)	WriteToServer 的異步版本，之間與 DataTable 中所提供行狀態匹配的行復制到 SqlBulkCopy 對象的 DestinationTableName 屬性中指定的目標表中。
WriteToServerAsync(DataTable, CancellationToken)	WriteToServer 的異步版本，將 DataTable 中提供的所有行都復制到 SqlBulkCopy 對象的 DestinationTableName 屬性指定的目標表中。取消標記可用於在命令超時超過前請求放棄操作。通過返回的任務對象將報告異常。
WriteToServerAsync(IDataReader, CancellationToken)	WriteToServer 的異步版本，將 IDataReader 中提供的所有行都復制到 SqlBulkCopy 對象的 DestinationTableName 屬性指定的目標表中。取消標記可用於在命令超時超過前請求放棄操作。通過返回的任務對象將報告異常。
WriteToServerAsync(DataTable, DataRowState, CancellationToken)	WriteToServer 的異步版本，之間與 DataTable 中所提供行狀態匹配的行復制到 SqlBulkCopy 對象的 DestinationTableName 屬性中指定的目標表中。取消標記可用於在命令超時超過前請求放棄操作。通過返回的任務對象將報告異常。

這里，我們選用 DataTable 來構建數據源，將 10W 數據導入 DataTable 中。可以看出，我們需要構建出給定 Entity 類型所對應的數據表的 DataTable，將所有的 entities 數據插入到 DataTable 中。

構建 TableMapping 映射

此時，我並不想手工書寫表中的各字段名稱，同時，我可能甚至都不想關心 Entity 類到底被映射到了數據庫中的哪一張表上。

此處，我們定義一個 TableMapping 類，用於存儲一張數據庫表的映射信息。

在獲取和生成 TableMapping 之前，我們需要先定義和獲取 DbMapping 類。

internal class DbMapping
{
    public DbMapping(DbContext context)
    {
      _context = context;

      var objectContext = ((IObjectContextAdapter)context).ObjectContext;
      _metadataWorkspace = objectContext.MetadataWorkspace;

      _codeFirstEntityContainer = _metadataWorkspace.GetEntityContainer("CodeFirstDatabase", DataSpace.SSpace);

      MapDb();
    }
}

通過讀取 CodeFirstEntityContainer 中的元數據，我們可以獲取到指定數據庫中的所有表的信息。

    private void MapDb()
    {
      ExtractTableColumnEdmMembers();

      List<EntityType> tables =
        _metadataWorkspace
          .GetItems(DataSpace.OCSpace)
          .Select(x => x.GetPrivateFieldValue("EdmItem") as EntityType)
          .Where(x => x != null)
          .ToList();

      foreach (var table in tables)
      {
        MapTable(table);
      }
    }

進而，根據表映射類型的定義，可以獲取到表中字段的映射信息。

    private void MapTable(EntityType tableEdmType)
    {
      string identity = tableEdmType.FullName;
      EdmType baseEdmType = tableEdmType;
      EntitySet storageEntitySet = null;

      while (!_codeFirstEntityContainer.TryGetEntitySetByName(baseEdmType.Name, false, out storageEntitySet))
      {
        if (baseEdmType.BaseType == null) break;
        baseEdmType = baseEdmType.BaseType;
      }
      if (storageEntitySet == null) return;

      var tableName = (string)storageEntitySet.MetadataProperties["Table"].Value;
      var schemaName = (string)storageEntitySet.MetadataProperties["Schema"].Value;

      var tableMapping = new TableMapping(identity, schemaName, tableName);
      _tableMappings.Add(identity, tableMapping);
      _primaryKeysMapping.Add(identity, storageEntitySet.ElementType.KeyMembers.Select(x => x.Name).ToList());

      foreach (var prop in storageEntitySet.ElementType.Properties)
      {
        MapColumn(identity, _tableMappings[identity], prop);
      }
    }

然后，可以將表信息和字段信息存放到 TableMapping 和 ColumnMapping 當中。

internal class TableMapping
{
    public string TableTypeFullName { get; private set; }
    public string SchemaName { get; private set; }
    public string TableName { get; private set; }

    public ColumnMapping[] Columns
    {
      get { return _columnMappings.Values.ToArray(); }
    }
}

構建 DataTable 數據

終於，有了 TableMapping 映射之后，我們可以開始創建 DataTable 了。

    private static DataTable BuildDataTable<T>(TableMapping tableMapping)
    {
      var entityType = typeof(T);
      string tableName = string.Join(@".", tableMapping.SchemaName, tableMapping.TableName);

      var dataTable = new DataTable(tableName);
      var primaryKeys = new List<DataColumn>();

      foreach (var columnMapping in tableMapping.Columns)
      {
        var propertyInfo = entityType.GetProperty(columnMapping.PropertyName, '.');
        columnMapping.Type = propertyInfo.PropertyType;

        var dataColumn = new DataColumn(columnMapping.ColumnName);

        Type dataType;
        if (propertyInfo.PropertyType.IsNullable(out dataType))
        {
          dataColumn.DataType = dataType;
          dataColumn.AllowDBNull = true;
        }
        else
        {
          dataColumn.DataType = propertyInfo.PropertyType;
          dataColumn.AllowDBNull = columnMapping.Nullable;
        }

        if (columnMapping.IsIdentity)
        {
          dataColumn.Unique = true;
          if (propertyInfo.PropertyType == typeof(int)
            || propertyInfo.PropertyType == typeof(long))
          {
            dataColumn.AutoIncrement = true;
          }
          else continue;
        }
        else
        {
          dataColumn.DefaultValue = columnMapping.DefaultValue;
        }

        if (propertyInfo.PropertyType == typeof(string))
        {
          dataColumn.MaxLength = columnMapping.MaxLength;
        }

        if (columnMapping.IsPk)
        {
          primaryKeys.Add(dataColumn);
        }

        dataTable.Columns.Add(dataColumn);
      }

      dataTable.PrimaryKey = primaryKeys.ToArray();

      return dataTable;
    }

通過 Schema 名稱和表名稱來構建指定 Entity 類型的 DataTable 對象。

然后將，entities 數據列表中的數據導入到 DataTable 對象之中。

    private static DataTable CreateDataTable<T>(TableMapping tableMapping, IEnumerable<T> entities)
    {
      var dataTable = BuildDataTable<T>(tableMapping);

      foreach (var entity in entities)
      {
        DataRow row = dataTable.NewRow();

        foreach (var columnMapping in tableMapping.Columns)
        {
          var @value = entity.GetPropertyValue(columnMapping.PropertyName);

          if (columnMapping.IsIdentity) continue;

          if (@value == null)
          {
            row[columnMapping.ColumnName] = DBNull.Value;
          }
          else
          {
            row[columnMapping.ColumnName] = @value;
          }
        }

        dataTable.Rows.Add(row);
      }

      return dataTable;
    }

SqlBulkCopy 導入數據

終於，數據源准備好了。然后使用如下代碼結構，調用 WriteToServer 方法，將數據寫入數據庫。

1       using (DataTable dataTable = CreateDataTable(tableMapping, entities))
2       {
3         using (SqlBulkCopy sqlBulkCopy = new SqlBulkCopy(transaction.Connection, options, transaction))
4         {
5           sqlBulkCopy.BatchSize = batchSize;
6           sqlBulkCopy.DestinationTableName = dataTable.TableName;
7           sqlBulkCopy.WriteToServer(dataTable);
8         }
9       }

看下保存 500 數據的效果，用時 1.9 秒。

看下保存 1W 數據的效果，用時 2.1 秒。

看下保存 10W 數據的效果，用時 7.5 秒。

再試下 100W 數據的效果，用時 27 秒。

封裝 BulkInsert 擴展方法

我們可以為 DbContext 添加一個 BulkInsert 擴展方法。

  internal static class DbContextBulkOperationExtensions
  {
    public const int DefaultBatchSize = 1000;

    public static void BulkInsert<T>(this DbContext context, IEnumerable<T> entities, int batchSize = DefaultBatchSize)
    {
      var provider = new BulkOperationProvider(context);
      provider.Insert(entities, batchSize);
    }
  }

IBulkableRepository 接口

在下面兩篇文章中，我介紹了精煉的 IRepository 接口。

• EntityFramework用法探索（四）Repository和UnitOfWork
• EntityFramework中使用Repository裝飾器

public interface IRepository<T>
    where T : class
  {
    IQueryable<T> Query();
    void Insert(T entity);
    void Update(T entity);
    void Delete(T entity);
  }

當我們需要擴展 BulkInsert 功能時，可以通過繼承來完成功能擴展。

public interface IBulkableRepository<T> : IRepository<T>
    where T : class
  {
    void BulkInsert(IEnumberable<T> entities);
  }

這樣，我們就可以使用很自然的方式直接使用 BulkInsert 功能了。

代碼在哪里

代碼在這里：

• 在 GitHub 上
  

參考資料

• SqlBulkCopy
• Table-Valued Parameters 
• EntityFramework.BulkInsert
• Using SQL bulk copy with your LINQ-to-SQL datacontext

本文為 Dennis Gao 原創技術文章，發表於博客園博客，未經作者本人允許禁止任何形式的轉載。