SQL Server 批量插入數(shù)據(jù)的完美解決方案

CREATE TABLE [dbo].[Person](
	[Id] [BIGINT] NOT NULL,
	[Name] [VARCHAR](64) NOT NULL,
	[Age] [INT] NOT NULL,
	[CreateTime] [DATETIME] NULL,
	[Sex] [INT] NOT NULL,
PRIMARY KEY CLUSTERED 
(
	[Id] ASC
)WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ON [PRIMARY]
) ON [PRIMARY]

public class Person
{
  public long Id { get; set; }
  public string Name { get; set; }
  public int Age { get; set; }
  public DateTime? CreateTime { get; set; }
  public Gender Sex { get; set; }
}

public enum Gender
{
  Man = 0,
  Woman = 1
}

//創(chuàng)建批量插入對象
using (var copy = new SqlBulkCopy(connection, options, externalTransaction))
{
  foreach (var column in ModelToDataTableTModel>.Columns)
  {
    //創(chuàng)建字段映射
    copy.ColumnMappings.Add(column.ColumnName, column.ColumnName);
  }
}

 public static class SqlConnectionExtension
  {
    /// summary>
    /// 批量復(fù)制
    /// /summary>
    /// typeparam name="TModel">插入的模型對象/typeparam>
    /// param name="source">需要批量插入的數(shù)據(jù)源/param>
    /// param name="connection">數(shù)據(jù)庫連接對象/param>
    /// param name="tableName">插入表名稱【為NULL默認(rèn)為實體名稱】/param>
    /// param name="bulkCopyTimeout">插入超時時間/param>
    /// param name="batchSize">寫入數(shù)據(jù)庫一批數(shù)量【如果為0代表全部一次性插入】最合適數(shù)量【這取決于您的環(huán)境，尤其是行數(shù)和網(wǎng)絡(luò)延遲。就個人而言，我將從BatchSize屬性設(shè)置為1000行開始，然后看看其性能如何。如果可行，那么我將使行數(shù)加倍（例如增加到2000、4000等），直到性能下降或超時。否則，如果超時發(fā)生在1000，那么我將行數(shù)減少一半（例如500），直到它起作用為止?！?param>
    /// param name="options">批量復(fù)制參數(shù)/param>
    /// param name="externalTransaction">執(zhí)行的事務(wù)對象/param>
    /// returns>插入數(shù)量/returns>
    public static int BulkCopyTModel>(this SqlConnection connection,
      IEnumerableTModel> source,
      string tableName = null,
      int bulkCopyTimeout = 30,
      int batchSize = 0,
      SqlBulkCopyOptions options = SqlBulkCopyOptions.Default,
      SqlTransaction externalTransaction = null)
    {
      //創(chuàng)建讀取器
      using (var reader = new EnumerableReaderTModel>(source))
      {
        //創(chuàng)建批量插入對象
        using (var copy = new SqlBulkCopy(connection, options, externalTransaction))
        {
          //插入的表
          copy.DestinationTableName = tableName ?? typeof(TModel).Name;
          //寫入數(shù)據(jù)庫一批數(shù)量
          copy.BatchSize = batchSize;
          //超時時間
          copy.BulkCopyTimeout = bulkCopyTimeout;
          //創(chuàng)建字段映射【如果沒有此字段映射會導(dǎo)致數(shù)據(jù)填錯位置，如果類型不對還會導(dǎo)致報錯】【因為：沒有此字段映射默認(rèn)是按照列序號對應(yīng)插入的】
          foreach (var column in ModelToDataTableTModel>.Columns)
          {
            //創(chuàng)建字段映射
            copy.ColumnMappings.Add(column.ColumnName, column.ColumnName);
          }
          //將數(shù)據(jù)批量寫入數(shù)據(jù)庫
          copy.WriteToServer(reader);
          //返回插入數(shù)據(jù)數(shù)量
          return reader.Depth;
        }
      }
    }

    /// summary>
    /// 批量復(fù)制-異步
    /// /summary>
    /// typeparam name="TModel">插入的模型對象/typeparam>
    /// param name="source">需要批量插入的數(shù)據(jù)源/param>
    /// param name="connection">數(shù)據(jù)庫連接對象/param>
    /// param name="tableName">插入表名稱【為NULL默認(rèn)為實體名稱】/param>
    /// param name="bulkCopyTimeout">插入超時時間/param>
    /// param name="batchSize">寫入數(shù)據(jù)庫一批數(shù)量【如果為0代表全部一次性插入】最合適數(shù)量【這取決于您的環(huán)境，尤其是行數(shù)和網(wǎng)絡(luò)延遲。就個人而言，我將從BatchSize屬性設(shè)置為1000行開始，然后看看其性能如何。如果可行，那么我將使行數(shù)加倍（例如增加到2000、4000等），直到性能下降或超時。否則，如果超時發(fā)生在1000，那么我將行數(shù)減少一半（例如500），直到它起作用為止?！?param>
    /// param name="options">批量復(fù)制參數(shù)/param>
    /// param name="externalTransaction">執(zhí)行的事務(wù)對象/param>
    /// returns>插入數(shù)量/returns>
    public static async Taskint> BulkCopyAsyncTModel>(this SqlConnection connection,
      IEnumerableTModel> source,
      string tableName = null,
      int bulkCopyTimeout = 30,
      int batchSize = 0,
      SqlBulkCopyOptions options = SqlBulkCopyOptions.Default,
      SqlTransaction externalTransaction = null)
    {
      //創(chuàng)建讀取器
      using (var reader = new EnumerableReaderTModel>(source))
      {
        //創(chuàng)建批量插入對象
        using (var copy = new SqlBulkCopy(connection, options, externalTransaction))
        {
          //插入的表
          copy.DestinationTableName = tableName ?? typeof(TModel).Name;
          //寫入數(shù)據(jù)庫一批數(shù)量
          copy.BatchSize = batchSize;
          //超時時間
          copy.BulkCopyTimeout = bulkCopyTimeout;
          //創(chuàng)建字段映射【如果沒有此字段映射會導(dǎo)致數(shù)據(jù)填錯位置，如果類型不對還會導(dǎo)致報錯】【因為：沒有此字段映射默認(rèn)是按照列序號對應(yīng)插入的】
          foreach (var column in ModelToDataTableTModel>.Columns)
          {
            //創(chuàng)建字段映射
            copy.ColumnMappings.Add(column.ColumnName, column.ColumnName);
          }
          //將數(shù)據(jù)批量寫入數(shù)據(jù)庫
          await copy.WriteToServerAsync(reader);
          //返回插入數(shù)據(jù)數(shù)量
          return reader.Depth;
        }
      }
    }
  }

 /// summary>
  /// 迭代器數(shù)據(jù)讀取器
  /// /summary>
  /// typeparam name="TModel">模型類型/typeparam>
  public class EnumerableReaderTModel> : IDataReader
  {
    /// summary>
    /// 實例化迭代器讀取對象
    /// /summary>
    /// param name="source">模型源/param>
    public EnumerableReader(IEnumerableTModel> source)
    {
      _source = source ?? throw new ArgumentNullException(nameof(source));
      _enumerable = source.GetEnumerator();
    }

    private readonly IEnumerableTModel> _source;
    private readonly IEnumeratorTModel> _enumerable;
    private object[] _currentDataRow = Array.Emptyobject>();
    private int _depth;
    private bool _release;

    public void Dispose()
    {
      _release = true;
      _enumerable.Dispose();
    }

    public int GetValues(object[] values)
    {
      if (values == null) throw new ArgumentNullException(nameof(values));
      var length = Math.Min(_currentDataRow.Length, values.Length);
      Array.Copy(_currentDataRow, values, length);
      return length;
    }

    public int GetOrdinal(string name)
    {
      for (int i = 0; i  ModelToDataTableTModel>.Columns.Count; i++)
      {
        if (ModelToDataTableTModel>.Columns[i].ColumnName == name) return i;
      }

      return -1;
    }

    public long GetBytes(int ordinal, long dataIndex, byte[] buffer, int bufferIndex, int length)
    {
      if (dataIndex  0) throw new Exception($"起始下標(biāo)不能小于0！");
      if (bufferIndex  0) throw new Exception("目標(biāo)緩沖區(qū)起始下標(biāo)不能小于0！");
      if (length  0) throw new Exception("讀取長度不能小于0！");
      var numArray = (byte[])GetValue(ordinal);
      if (buffer == null) return numArray.Length;
      if (buffer.Length = bufferIndex) throw new Exception("目標(biāo)緩沖區(qū)起始下標(biāo)不能大于目標(biāo)緩沖區(qū)范圍！");
      var freeLength = Math.Min(numArray.Length - bufferIndex, length);
      if (freeLength = 0) return 0;
      Array.Copy(numArray, dataIndex, buffer, bufferIndex, length);
      return freeLength;
    }

    public long GetChars(int ordinal, long dataIndex, char[] buffer, int bufferIndex, int length)
    {
      if (dataIndex  0) throw new Exception($"起始下標(biāo)不能小于0！");
      if (bufferIndex  0) throw new Exception("目標(biāo)緩沖區(qū)起始下標(biāo)不能小于0！");
      if (length  0) throw new Exception("讀取長度不能小于0！");
      var numArray = (char[])GetValue(ordinal);
      if (buffer == null) return numArray.Length;
      if (buffer.Length = bufferIndex) throw new Exception("目標(biāo)緩沖區(qū)起始下標(biāo)不能大于目標(biāo)緩沖區(qū)范圍！");
      var freeLength = Math.Min(numArray.Length - bufferIndex, length);
      if (freeLength = 0) return 0;
      Array.Copy(numArray, dataIndex, buffer, bufferIndex, length);
      return freeLength;
    }

    public bool IsDBNull(int i)
    {
      var value = GetValue(i);
      return value == null || value is DBNull;
    }
    public bool NextResult()
    {
      //移動到下一個元素
      if (!_enumerable.MoveNext()) return false;
      //行層+1
      Interlocked.Increment(ref _depth);
      //得到數(shù)據(jù)行
      _currentDataRow = ModelToDataTableTModel>.ToRowData.Invoke(_enumerable.Current);
      return true;
    }

    public byte GetByte(int i) => (byte)GetValue(i);
    public string GetName(int i) => ModelToDataTableTModel>.Columns[i].ColumnName;
    public string GetDataTypeName(int i) => ModelToDataTableTModel>.Columns[i].DataType.Name;
    public Type GetFieldType(int i) => ModelToDataTableTModel>.Columns[i].DataType;
    public object GetValue(int i) => _currentDataRow[i];
    public bool GetBoolean(int i) => (bool)GetValue(i);
    public char GetChar(int i) => (char)GetValue(i);
    public Guid GetGuid(int i) => (Guid)GetValue(i);
    public short GetInt16(int i) => (short)GetValue(i);
    public int GetInt32(int i) => (int)GetValue(i);
    public long GetInt64(int i) => (long)GetValue(i);
    public float GetFloat(int i) => (float)GetValue(i);
    public double GetDouble(int i) => (double)GetValue(i);
    public string GetString(int i) => (string)GetValue(i);
    public decimal GetDecimal(int i) => (decimal)GetValue(i);
    public DateTime GetDateTime(int i) => (DateTime)GetValue(i);
    public IDataReader GetData(int i) => throw new NotSupportedException();
    public int FieldCount => ModelToDataTableTModel>.Columns.Count;
    public object this[int i] => GetValue(i);
    public object this[string name] => GetValue(GetOrdinal(name));
    public void Close() => Dispose();
    public DataTable GetSchemaTable() => ModelToDataTableTModel>.ToDataTable(_source);
    public bool Read() => NextResult();
    public int Depth => _depth;
    public bool IsClosed => _release;
    public int RecordsAffected => 0;
  }

/// summary>
  /// 對象轉(zhuǎn)換成DataTable轉(zhuǎn)換類
  /// /summary>
  /// typeparam name="TModel">泛型類型/typeparam>
  public static class ModelToDataTableTModel>
  {
    static ModelToDataTable()
    {
      //如果需要剔除某些列可以修改這段代碼
      var propertyList = typeof(TModel).GetProperties().Where(w => w.CanRead).ToArray();
      Columns = new ReadOnlyCollectionDataColumn>(propertyList
        .Select(pr => new DataColumn(pr.Name, GetDataType(pr.PropertyType))).ToArray());
      //生成對象轉(zhuǎn)數(shù)據(jù)行委托
      ToRowData = BuildToRowDataDelegation(typeof(TModel), propertyList);
    }

    /// summary>
    /// 構(gòu)建轉(zhuǎn)換成數(shù)據(jù)行委托
    /// /summary>
    /// param name="type">傳入類型/param>
    /// param name="propertyList">轉(zhuǎn)換的屬性/param>
    /// returns>轉(zhuǎn)換數(shù)據(jù)行委托/returns>
    private static FuncTModel, object[]> BuildToRowDataDelegation(Type type, PropertyInfo[] propertyList)
    {
      var source = Expression.Parameter(type);
      var items = propertyList.Select(property => ConvertBindPropertyToData(source, property));
      var array = Expression.NewArrayInit(typeof(object), items);
      var lambda = Expression.LambdaFuncTModel, object[]>>(array, source);
      return lambda.Compile();
    }

    /// summary>
    /// 將屬性轉(zhuǎn)換成數(shù)據(jù)
    /// /summary>
    /// param name="source">源變量/param>
    /// param name="property">屬性信息/param>
    /// returns>獲取屬性數(shù)據(jù)表達(dá)式/returns>
    private static Expression ConvertBindPropertyToData(ParameterExpression source, PropertyInfo property)
    {
      var propertyType = property.PropertyType;
      var expression = (Expression)Expression.Property(source, property);
      if (propertyType.IsEnum)
        expression = Expression.Convert(expression, propertyType.GetEnumUnderlyingType());
      return Expression.Convert(expression, typeof(object));
    }

    /// summary>
    /// 獲取數(shù)據(jù)類型
    /// /summary>
    /// param name="type">屬性類型/param>
    /// returns>數(shù)據(jù)類型/returns>
    private static Type GetDataType(Type type)
    {
      //枚舉默認(rèn)轉(zhuǎn)換成對應(yīng)的值類型
      if (type.IsEnum)
        return type.GetEnumUnderlyingType();
      //可空類型
      if (type.IsGenericType  type.GetGenericTypeDefinition() == typeof(Nullable>))
        return GetDataType(type.GetGenericArguments().First());
      return type;
    }

    /// summary>
    /// 列集合
    /// /summary>
    public static IReadOnlyListDataColumn> Columns { get; }

    /// summary>
    /// 對象轉(zhuǎn)數(shù)據(jù)行委托
    /// /summary>
    public static FuncTModel, object[]> ToRowData { get; }

    /// summary>
    /// 集合轉(zhuǎn)換成DataTable
    /// /summary>
    /// param name="source">集合/param>
    /// param name="tableName">表名稱/param>
    /// returns>轉(zhuǎn)換完成的DataTable/returns>
    public static DataTable ToDataTable(IEnumerableTModel> source, string tableName = "TempTable")
    {
      //創(chuàng)建表對象
      var table = new DataTable(tableName);
      //設(shè)置列
      foreach (var dataColumn in Columns)
      {
        table.Columns.Add(new DataColumn(dataColumn.ColumnName, dataColumn.DataType));
      }

      //循環(huán)轉(zhuǎn)換每一行數(shù)據(jù)
      foreach (var item in source)
      {
        table.Rows.Add(ToRowData.Invoke(item));
      }

      //返回表對象
      return table;
    }
  }

CREATE TABLE [dbo].[Person](
	[Id] [BIGINT] NOT NULL,
	[Name] [VARCHAR](64) NOT NULL,
	[Age] [INT] NOT NULL,
	[CreateTime] [DATETIME] NULL,
	[Sex] [INT] NOT NULL,
PRIMARY KEY CLUSTERED 
(
	[Id] ASC
)WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ON [PRIMARY]
) ON [PRIMARY]

public class Person
{
  public long Id { get; set; }
  public string Name { get; set; }
  public int Age { get; set; }
  public DateTime? CreateTime { get; set; }
  public Gender Sex { get; set; }
}

public enum Gender
{
  Man = 0,
  Woman = 1
}

//生成10萬條數(shù)據(jù)
var persons = new Person[100000];
var random = new Random();
for (int i = 0; i  persons.Length; i++)
{
  persons[i] = new Person
  {
    Id = i + 1,
    Name = "張三" + i,
    Age = random.Next(1, 128),
    Sex = (Gender)random.Next(2),
    CreateTime = random.Next(2) == 0 ? null : (DateTime?) DateTime.Now.AddSeconds(i)
  };
}

//創(chuàng)建數(shù)據(jù)庫連接
using (var conn = new SqlConnection("Server=.;Database=DemoDataBase;User ID=sa;Password=8888;"))
{
  conn.Open();
  var sw = Stopwatch.StartNew();
  //批量插入數(shù)據(jù)
  var qty = conn.BulkCopy(persons);
  sw.Stop();
  Console.WriteLine(sw.Elapsed.TotalMilliseconds + "ms");
}