net core天马行空系列-各大数据库快速批量插入数据方法汇总 ...

1.前言

hi，大家好，我是三合。我是怎么想起写一篇关于数据库快速批量插入的博客的呢？事情起源于我们工作中的一个需求，简单来说，就是有一个定时任务，从数据库里获取大量数据，在应用层面经过处理后再把结果批量插入回到数据库里。这个任务每十分钟执行一次，但是有的时候数据量太大，循环插入数据库的时候会超时，导致任务失败，所以这个时候我就开始研究怎么快速批量插入数据库，因为我们用的数据库是Oracle，所以我首先研究了Oracle的快速批量插入，后面我一想那其他类型的数据库肯定也有这样的需求，于是我在找了很多资料，并且反复实验后，终于完美解决了mysql,sqlServer以及Oracle的快速批量插入，sqlite自身不支持，所以没有sqlite，特地整理成这篇文章，分享给大家。
2.测试前准备

添加一个具有绝大多数类型属性的实体类，用来完整测验批量插入效果,该实体类用于mysql和sqlserver的测试。

1. public class NullableTable
2. {
3.     [DatabaseGenerated(DatabaseGeneratedOption.Identity)]
4.     [Key]
5.     public int Id { get; set; }
6.     [Description("Int2")]
7.     public int? Int2 { get; set; }
8.     [Description("Long2")]
9.     public long? Long2 { get; set; }
11.     public float? Float2 { get; set; }
12.     public double? Double2 { get; set; }
14.     public decimal? Decimal2 { get; set; }
16.     [DecimalPrecision(20,4)]
17.     public decimal? Decimal3 { get; set; }
19.     public Guid? Guid2 { get; set; }
21.     public short? Short2 { get; set; }
23.     public DateTime? DateTime2 { get; set; }
25.     public bool? Bool2 { get; set; }
27.     public TimeSpan? TimeSpan2 { get; set; }
29.     public byte? Byte2 { get; set; }
32.     [StringLength(100)]
33.     public string String2 { get; set; }
34.     public string String3 { get; set; }
36.     public Enum2? Enum2 { get; set; }
38.     [Column("TestInt3")]
39.     [Description("Int2")]
40.     public int? Int3 { get; set; }
41. }
43. public enum Enum2
44.     {
45.         x,
46.         y
47.     }

复制代码

因为oracle数据库我们习惯于表名和字段名大写，所以oracle的测试实体类定义如下：

1. [Table("NULLABLETABLE")]
2. [Description("NullableTable")]
3. public class NullableTable
4. {
5.     [DatabaseGenerated(DatabaseGeneratedOption.Identity)]
6.     [Key]
7.     [Column("ID")]
8.     public int Id { get; set; }
9.     [Description("Int2")]
10.     [Column("INT2")]
11.     public int? Int2 { get; set; }
12.     [Description("Long2")]
13.     [Column("LONG2")]
14.     public long? Long2 { get; set; }
15.     [Column("FLOAT2")]
16.     public float? Float2 { get; set; }
17.     [Column("DOUBLE2")]
18.     public double? Double2 { get; set; }
19.     [Column("DECIMAL2")]
20.     public decimal? Decimal2 { get; set; }
21.     [Column("DECIMAL3")]
22.     [DecimalPrecision(20,4)]
23.     public decimal? Decimal3 { get; set; }
24.     [Column("GUID2")]
25.     public Guid? Guid2 { get; set; }
26.     [Column("SHORT2")]
27.     public short? Short2 { get; set; }
28.     [Column("DATETIME2")]
29.     public DateTime? DateTime2 { get; set; }
30.     [Column("BOOL2")]
31.     public bool? Bool2 { get; set; }
32.     [Column("TIMESPAN2")]
33.     public TimeSpan? TimeSpan2 { get; set; }
34.     [Column("BYTE2")]
35.     public byte? Byte2 { get; set; }
37.     [Column("STRING2")]
38.     [StringLength(100)]
39.     public string String2 { get; set; }
40.     [Column("STRING3")]
41.     public string String3 { get; set; }
42.     [Column("ENUM2")]
43.     public Enum2? Enum2 { get; set; }
45.     [Column("TESTINT3")]
46.     [Description("Int2")]
47.     public int? Int3 { get; set; }
48. }

复制代码

实验我们采用的是code first，先利用SummerBoot框架的可用于依赖注入的，数据库表和c#实体类互相转换的接口实现功能从实体类生成相应的数据库表，本次实验批量插入2w条数据来对比时间，定义一个列表，用循环的方式给这个列表添加2w条数据。

1. var nullableTableList3 = new List<NullableTable>();
2. var now = DateTime.Now;
3. for (int i = 0; i < 20000; i++)
4. {
5.     var a = new NullableTable()
6.     {
7.         Int2 = 2,
8.         Bool2 = true,
9.         Byte2 = 1,
10.         DateTime2 = now,
11.         Decimal2 = 1m,
12.         Decimal3 = 1.1m,
13.         Double2 = 1.1,
14.         Float2 = (float)1.1,
15.         Guid2 = Guid.NewGuid(),
16.         Id = 0,
17.         Short2 = 1,
18.         TimeSpan2 = TimeSpan.FromHours(1),
19.         String2 = "sb",
20.         String3 = "sb",
21.         Long2 = 2,
22.         Enum2 = Model.Enum2.y,
23.         Int3 = 4
24.     };
25.     nullableTableList3.Add(a);
26. }

复制代码

数据库驱动上的选择是这样的，sqlserver采用微软官方驱动System.Data.SqlClient，oracle采用官方驱动Oracle.ManagedDataAccess.Core，mysql采用社区驱动MySqlConnector（为啥mysql不采用官方的驱动呢？因为官方的驱动封装的太差了，社区的驱动支持列名映射，同时项目里官方驱动和社区驱动可以共存）。
同时快速批量插入均支持异步同步，这里仅演示同步，异步的实现基本一样。
3.sqlserver快速批量插入

sqlserver官方提供的批量插入方式是SqlBulkCopy，参数为一个dataTable对象，原生的批量插入代码如下,采用StopWatch类进行计时,测试前都会用DELETE  from NullableTable 语句清空表，测试里循环跑5次，获取总时间后除以5获取平均值，合计插入10w条数据。

1. var sw = new Stopwatch();
2. sw.Start();
3. for (int i = 0; i < 5; i++)
4. {
5.   using (var dbConnection = new SqlConnection(connectionString))
6.   {
7.       dbConnection.Open();
9.       SqlBulkCopy sqlBulkCopy = new SqlBulkCopy(dbConnection, SqlBulkCopyOptions.KeepIdentity,
10.           null);
11.       sqlBulkCopy.BatchSize = 20000;
12.       sqlBulkCopy.DestinationTableName = "NullableTable";
13.       //针对列名做一下映射
14.       sqlBulkCopy.ColumnMappings.Add("Int2", "Int2");
15.       sqlBulkCopy.ColumnMappings.Add("Bool2", "Bool2");
16.       sqlBulkCopy.ColumnMappings.Add("Byte2", "Byte2");
17.       sqlBulkCopy.ColumnMappings.Add("DateTime2", "DateTime2");
18.       sqlBulkCopy.ColumnMappings.Add("Decimal2", "Decimal2");
19.       sqlBulkCopy.ColumnMappings.Add("Decimal3", "Decimal3");
20.       sqlBulkCopy.ColumnMappings.Add("Double2", "Double2");
21.       sqlBulkCopy.ColumnMappings.Add("Float2", "Float2");
22.       sqlBulkCopy.ColumnMappings.Add("Guid2", "Guid2");
23.       sqlBulkCopy.ColumnMappings.Add("Short2", "Short2");
24.       sqlBulkCopy.ColumnMappings.Add("TimeSpan2", "TimeSpan2");
26.       sqlBulkCopy.ColumnMappings.Add("String2", "String2");
27.       sqlBulkCopy.ColumnMappings.Add("String3", "String3");
28.       sqlBulkCopy.ColumnMappings.Add("Long2", "Long2");
29.       sqlBulkCopy.ColumnMappings.Add("Enum2", "Enum2");
30.       sqlBulkCopy.ColumnMappings.Add("Int3", "TestInt3");
31.       //将实体类列表转换成dataTable
32.       var table = nullableTableList3.ToDataTable();
33.       sqlBulkCopy.WriteToServer(table);
34.   }
36. }
37. sw.Stop();            
38. var totalTime= sw.ElapsedMilliseconds;
39. var avgValue = totalTime / 5;

复制代码

实验结果如下，sql server中：
采用快速批量插入10w条数据，时间合计1858毫秒，平均插入2w条数据仅需371毫秒。
采用insert into语句，循环插入10w条数据，时间合计457606毫秒，平均插入2w条数据需91521毫秒。
4.实体类列表转dataTable的扩展方法

这里有一个实体类列表转dataTable的扩展方法，采用的是表达式树+构建委托的方式，性能不错，大家可以参考，代码实现如下。

1. public static ConcurrentDictionary<string, object> CacheDictionary = new ConcurrentDictionary<string, object>();
2. /// <summary>
3. /// 构建一个object数据转换成一维数组数据的委托
4. /// </summary>
5. /// <param name="objType"></param>
6. /// <param name="propertyInfos"></param>
7. /// <returns></returns>
8. public static Func<T, object[]> BuildObjectGetValuesDelegate<T>(List<PropertyInfo> propertyInfos) where T : class
9. {
10.     var objParameter = Expression.Parameter(typeof(T), "model");
11.     var selectExpressions = propertyInfos.Select(it => BuildObjectGetValueExpression(objParameter, it));
12.     var arrayExpression = Expression.NewArrayInit(typeof(object), selectExpressions);
13.     var result = Expression.Lambda<Func<T, object[]>>(arrayExpression, objParameter).Compile();
14.     return result;
15. }
18. /// <summary>
19. /// 构建对象获取单个值得
20. /// </summary>
21. /// <param name="modelExpression"></param>
22. /// <param name="propertyInfo"></param>
23. /// <returns></returns>
24. public static Expression BuildObjectGetValueExpression(ParameterExpression modelExpression, PropertyInfo propertyInfo)
25. {
26.     var propertyExpression = Expression.Property(modelExpression, propertyInfo);
27.     var convertExpression = Expression.Convert(propertyExpression, typeof(object));
28.     return convertExpression;
29. }
31. public static DataTable ToDataTable<T>(this IEnumerable<T> source, List<PropertyInfo> propertyInfos = null,bool useColumnAttribute=false) where T : class
32. {
33.     var table = new DataTable("template");
34.     if (propertyInfos == null || propertyInfos.Count == 0)
35.     {
36.         propertyInfos = typeof(T).GetProperties().Where(it => it.CanRead).ToList();
37.     }
38.     foreach (var propertyInfo in propertyInfos)
39.     {
40.         var columnName=useColumnAttribute?(propertyInfo.GetCustomAttribute<ColumnAttribute>()?.Name?? propertyInfo.Name) : propertyInfo.Name;
41.         table.Columns.Add(columnName, ChangeType(propertyInfo.PropertyType));
42.     }
44.     Func<T, object[]> func;
45.     var key = typeof(T).FullName + propertyInfos.Select(it => it.Name).ToList().StringJoin();
46.     if (CacheDictionary.TryGetValue(key, out var cacheFunc))
47.     {
48.         func = (Func<T, object[]>)cacheFunc;
49.     }
50.     else
51.     {
52.         func = BuildObjectGetValuesDelegate<T>(propertyInfos);
53.         CacheDictionary.TryAdd(key, func);
54.     }
56.     foreach (var model in source)
57.     {
58.         var rowData = func(model);
59.         table.Rows.Add(rowData);
60.     }
62.     return table;
63. }
65. private static Type ChangeType(Type type)
66. {
67.     if (type.IsNullable())
68.     {
69.         type = Nullable.GetUnderlyingType(type);
70.     }
72.     return type;
73. }

复制代码

5.oracle快速批量插入

oracle官方提供的批量插入方式是ArrayBindCount，即数组批量插入，原生的批量插入代码如下，计时方式与sqlserver相同

1. var total = 20000;
2. var sw = new Stopwatch();
3. sw.Start();
4. for (int i = 0; i < 5; i++)
5. {
6.     var connection = new OracleConnection(connectionString);
7.     connection.Open();
8.     int?[] Int2 = new int?[total];
9.     bool[] Bool2 = new bool[total];
10.     byte[] Byte2 = new byte[total];
11.     DateTime[] DateTime2 = new DateTime[total];
12.     decimal?[] Decimal2 = new decimal?[total];
13.     decimal[] Decimal3 = new decimal[total];
14.     double[] Double2 = new double[total];
15.     float[] Float2 = new float[total];
16.     Guid?[] Guid2 = new Guid?[total];
17.     short[] Short2 = new short[total];
18.     TimeSpan[] TimeSpan2 = new TimeSpan[total];
19.     string[] String2 = new string[total];
20.     string[] String3 = new string[total];
21.     long[] Long2 = new long[total];
22.     Enum2[] Enum2 = new Enum2[total];
24.     for (int j = 0; j < total; j++)
25.     {
26.         Int2[j] = 2;
27.         Bool2[j] = true;
28.         Byte2[j] = 1;
29.         DateTime2[j] = now;
30.         Decimal2[j] = 1m;
31.         Decimal3[j] = 1.1m;
32.         Double2[j] = 1.1;
33.         Float2[j] = (float) 1.1;
34.         Guid2[j] = Guid.NewGuid();
35.         Short2[j] = 1;
36.         TimeSpan2[j] = TimeSpan.FromHours(1);
37.         String2[j] = "sb";
38.         String3[j] = "sb";
39.         Long2[j] = 2;
40.         Enum2[j] = Model.Enum2.y;
41.     }
43.     var c = (int) Model.Enum2.y;
44.     OracleParameter pInt2 = new OracleParameter();
45.     pInt2.OracleDbType = OracleDbType.Int32;
46.     pInt2.Value = Int2;
48.     OracleParameter pBool2 = new OracleParameter();
49.     pBool2.OracleDbType = OracleDbType.Byte;
50.     pBool2.Value = Bool2;
52.     OracleParameter pByte2 = new OracleParameter();
53.     pByte2.OracleDbType = OracleDbType.Byte;
54.     pByte2.Value = Byte2;
56.     OracleParameter pDateTime2 = new OracleParameter();
57.     pDateTime2.OracleDbType = OracleDbType.TimeStamp;
58.     pDateTime2.Value = DateTime2;
60.     OracleParameter pDecimal2 = new OracleParameter();
61.     pDecimal2.OracleDbType = OracleDbType.Decimal;
62.     pDecimal2.Value = Decimal2;
64.     OracleParameter pDecimal3 = new OracleParameter();
65.     pDecimal3.OracleDbType = OracleDbType.Decimal;
66.     pDecimal3.Value = Decimal3;
68.     OracleParameter pDouble2 = new OracleParameter();
69.     pDouble2.OracleDbType = OracleDbType.Double;
70.     pDouble2.Value = Double2;
72.     OracleParameter pFloat2 = new OracleParameter();
73.     pFloat2.OracleDbType = OracleDbType.BinaryFloat;
74.     pFloat2.Value = Float2;
77.     OracleParameter pGuid2 = new OracleParameter();
78.     pGuid2.OracleDbType = OracleDbType.Raw;
79.     pGuid2.Value = Guid2;
81.     OracleParameter pShort2 = new OracleParameter();
82.     pShort2.OracleDbType = OracleDbType.Int16;
83.     pShort2.Value = Short2;
85.     OracleParameter pTimeSpan2 = new OracleParameter();
86.     pTimeSpan2.OracleDbType = OracleDbType.IntervalDS;
87.     pTimeSpan2.Value = TimeSpan2;
89.     OracleParameter pString2 = new OracleParameter();
90.     pString2.OracleDbType = OracleDbType.Varchar2;
91.     pString2.Value = String2;
93.     OracleParameter pString3 = new OracleParameter();
94.     pString3.OracleDbType = OracleDbType.Varchar2;
95.     pString3.Value = String3;
98.     OracleParameter pLong2 = new OracleParameter();
99.     pLong2.OracleDbType = OracleDbType.Long;
100.     pLong2.Value = Long2;
102.     OracleParameter pEnum2 = new OracleParameter();
103.     pEnum2.OracleDbType = OracleDbType.Byte;
104.     pEnum2.Value = Enum2;
105.     // create command and set properties
106.     OracleCommand cmd = connection.CreateCommand();
107.     cmd.CommandText =
108.         "INSERT INTO NULLABLETABLE (INT2, LONG2, FLOAT2, DOUBLE2, DECIMAL2, DECIMAL3, GUID2, SHORT2, DATETIME2, BOOL2, TIMESPAN2, BYTE2, STRING2, STRING3,ENUM2) VALUES(:1,:2,:3,:4,:5,:6,:7,:8,:9,:10,:11,:12,:13,:14,:15)";
109.     cmd.ArrayBindCount = total;
110.     cmd.Parameters.Add(pInt2);
111.     cmd.Parameters.Add(pLong2);
112.     cmd.Parameters.Add(pFloat2);
113.     cmd.Parameters.Add(pDouble2);
114.     cmd.Parameters.Add(pDecimal2);
115.     cmd.Parameters.Add(pDecimal3);
116.     cmd.Parameters.Add(pGuid2);
117.     cmd.Parameters.Add(pShort2);
118.     cmd.Parameters.Add(pDateTime2);
119.     cmd.Parameters.Add(pBool2);
120.     cmd.Parameters.Add(pTimeSpan2);
121.     cmd.Parameters.Add(pByte2);
122.     cmd.Parameters.Add(pString2);
123.     cmd.Parameters.Add(pString3);
124.     cmd.Parameters.Add(pEnum2);
125.     cmd.ExecuteNonQuery();
126. }
127. sw.Stop();
129. var totalTime = sw.ElapsedMilliseconds;
130. var avgValue = totalTime / 5;

复制代码

实验结果如下，oracle中：
采用快速批量插入10w条数据，时间合计2323毫秒，平均插入2w条数据仅需464毫秒。
采用insert into语句，循环插入10w条数据，时间合计462837毫秒，平均插入2w条数据仅需92567毫秒。
6.mysql快速批量插入

mysql社区驱动MySqlConnector提供的批量插入方式是SqlBulkCopy，基于mysql自身的文件上传机制进行批量插入，参数为一个dataTable对象，原生的批量插入代码如下,计时方式与sqlserver相同,同时，mysql的连接字符串里要添加";AllowLoadLocalInfile=true"，即连接字符串的形式应该是"Server= ;Database=;User ID=;Password=;AllowLoadLocalInfile=true"，同时在mysql数据库上执行"set global local_infile=1"开启批量上传

1. var sw = new Stopwatch();
2. sw.Start();
3. for (int j = 0; j < 5; j++)
4. {
5.     using (var dbConnection = new MySqlConnection(connectionString))
6.     {
7.         dbConnection.Open();
9.         MySqlBulkCopy sqlBulkCopy = new MySqlBulkCopy(dbConnection, null);
10.         sqlBulkCopy.DestinationTableName = "NullableTable";
11.         var propertys = typeof(NullableTable).GetProperties()
12.             .Where(it => it.CanRead && it.GetCustomAttribute<NotMappedAttribute>() == null).ToList();
14.         for (int i = 0; i < propertys.Count; i++)
15.         {
16.             var property = propertys[i];
17.             var columnName = property.GetCustomAttribute<ColumnAttribute>()?.Name ?? property.Name;
19.             if (property.PropertyType.GetUnderlyingType() == typeof(Guid))
20.             {
21.                 sqlBulkCopy.ColumnMappings.Add(new MySqlBulkCopyColumnMapping(i, "@tmp",
22.                     $"{columnName} =unhex(@tmp)"));
23.             }
24.             else
25.             {
26.                 sqlBulkCopy.ColumnMappings.Add(new MySqlBulkCopyColumnMapping(i, columnName));
27.             }
28.         }
30.         var table = nullableTableList3.ToDataTable();
32.         SbUtil.ReplaceDataTableColumnType<Guid, byte[]>(table, guid1 => guid1.ToByteArray());
33.         var c = sqlBulkCopy.WriteToServer(table);
34.     }
35. }
37. sw.Stop();
39. var totalTime = sw.ElapsedMilliseconds;
40. var avgValue = totalTime / 5;

复制代码

实验结果如下，mysql中：
采用快速批量插入10w条数据，时间合计2350毫秒，平均插入2w条数据仅需470毫秒。
采用insert into语句，循环插入10w条数据，时间合计414700毫秒，平均插入2w条数据需82940毫秒。
在mysql中c#的guid对应的mysql字段类型为varbinary(16)，所以table里的guid要转换为字节数组，否则插入数据库后，guid的值就会变成乱码,字节数组传递到mysql服务端后利用unhex函数进行解析，即可正常保存guid类型。 将table里guid的值转为字节数组的方法-SbUtil.ReplaceDataTableColumnType的代码实现如下：

1. /// <summary>
2. /// 替换dataTable里的列类型
3. /// </summary>
4. /// <param name="dt"></param>
5. public  static void ReplaceDataTableColumnType<OldType,NewType>(DataTable dt,Func<OldType, NewType> replaceFunc)
6. {
7.     var needUpdateColumnIndexList = new List<int>();
8.     var needUpdateColumnNameList = new List<string>();
9.    
10.     for (int i = 0; i < dt.Columns.Count; i++)
11.     {
12.         var column = dt.Columns[i];
13.         if (column.DataType.GetUnderlyingType() == typeof(OldType))
14.         {
15.             needUpdateColumnIndexList.Add(i);
16.             needUpdateColumnNameList.Add(column.ColumnName);
17.          
18.         }
19.     }
21.     if (needUpdateColumnIndexList.Count == 0)
22.     {
23.         return;
24.     }
26.     var nameMapping = new Dictionary<string, string>();
27.     for (int i = 0; i < needUpdateColumnIndexList.Count; i++)
28.     {
29.         var oldColumnName = needUpdateColumnNameList[i];
30.         var newColumnName = Guid.NewGuid().ToString("N");
31.         nameMapping.Add(newColumnName, oldColumnName);
32.       
33.         dt.Columns.Add(newColumnName, typeof(byte[])).SetOrdinal(needUpdateColumnIndexList[i]);
34.         for (int j = 0; j < dt.Rows.Count; j++)
35.         {
36.             var c = (dt.Rows[j][oldColumnName]);
37.             dt.Rows[j][newColumnName] = replaceFunc((OldType)(dt.Rows[j][oldColumnName]));
38.         }
39.         dt.Columns.Remove(oldColumnName);
40.     }
41.    
42.     for (int i = 0; i < dt.Columns.Count; i++)
43.     {
44.         var columnName = dt.Columns[i].ColumnName;
45.         if (nameMapping.ContainsKey(columnName))
46.         {
47.             dt.Columns[i].ColumnName = nameMapping[columnName];
48.         }
49.     }
51. }

复制代码

7.SummerBoot对各数据库快速批量插入的封装

基于以上各种数据库对于快速批量插入的原生写法过于复杂难记，SummerBoot对其进行了封装，在声明式编程的理念下，封装后仅需3步即可快速批量插入，这里以sqlserver举例。
7.1在StartUp.cs中添加summerBoot的服务支持

1. services.AddSummerBoot();
2. services.AddSummerBootRepository(it =>
3. {
4.     it.DbConnectionType = typeof(SqlConnection);
5.     it.ConnectionString = connectionString;
6. });

复制代码

7.2添加仓储接口

1. [AutoRepository]
2. public interface INullableTableRepository : IBaseRepository<NullableTable>
3. {
4.    
5. }

复制代码

7.3注入仓储接口后直接调用FastBatchInsert方法

1. var sw = new Stopwatch();
2. sw.Start();
3. for (int i = 0; i < 5; i++)
4. {
5.     nullableTableRepository.FastBatchInsert(nullableTableList3);
6. }
7. sw.Stop();
8.   
9. var totalTime= sw.ElapsedMilliseconds;
10. var avgValue = totalTime / 5;

复制代码

实验结果如下，sql server中：
采用SummerBoot统一封装后快速批量插入10w条数据，时间合计3926（原生快速批量写法1858）毫秒，平均插入2w条数据仅需785(原生快速批量写法371)毫秒。从对比可以看出，经过SummerBoot封装后，快速批量插入所花费的时间有所增加，但是对于这么大数据量而言，这点多消耗的时间和节省的开发量对比，不值一提。
写在最后

SummerBoot是一款声明式编程框架，专注于”做什么”而不是”如何去做”，更多用法，可参考SummerBoot文档，也可以加入QQ群:799648362反馈建议。同时各位看官，如果你觉得这篇文章还不错的话，请帮忙一键三连哦(推荐+关注+github star)

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！