.NET Core 各大数据库快速批量插入方法汇总

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

publicclassNullableTable
{
[DatabaseGenerated(DatabaseGeneratedOption.Identity)]
[Key]
publicintId{get;set;}
[Description("Int2")]
publicint?Int2{get;set;}
[Description("Long2")]
publiclong?Long2{get;set;}

publicfloat?Float2{get;set;}
publicdouble?Double2{get;set;}

publicdecimal?Decimal2{get;set;}

[DecimalPrecision(20,4)]
publicdecimal?Decimal3{get;set;}

publicGuid?Guid2{get;set;}

publicshort?Short2{get;set;}

publicDateTime?DateTime2{get;set;}

publicbool?Bool2{get;set;}

publicTimeSpan?TimeSpan2{get;set;}

publicbyte?Byte2{get;set;}


[StringLength(100)]
publicstringString2{get;set;}
publicstringString3{get;set;}

publicEnum2?Enum2{get;set;}

[Column("TestInt3")]
[Description("Int2")]
publicint?Int3{get;set;}
}
publicenumEnum2
{
x,
y
}

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

[Table("NULLABLETABLE")]
[Description("NullableTable")]
publicclassNullableTable
{
[DatabaseGenerated(DatabaseGeneratedOption.Identity)]
[Key]
[Column("ID")]
publicintId{get;set;}
[Description("Int2")]
[Column("INT2")]
publicint?Int2{get;set;}
[Description("Long2")]
[Column("LONG2")]
publiclong?Long2{get;set;}
[Column("FLOAT2")]
publicfloat?Float2{get;set;}
[Column("DOUBLE2")]
publicdouble?Double2{get;set;}
[Column("DECIMAL2")]
publicdecimal?Decimal2{get;set;}
[Column("DECIMAL3")]
[DecimalPrecision(20,4)]
publicdecimal?Decimal3{get;set;}
[Column("GUID2")]
publicGuid?Guid2{get;set;}
[Column("SHORT2")]
publicshort?Short2{get;set;}
[Column("DATETIME2")]
publicDateTime?DateTime2{get;set;}
[Column("BOOL2")]
publicbool?Bool2{get;set;}
[Column("TIMESPAN2")]
publicTimeSpan?TimeSpan2{get;set;}
[Column("BYTE2")]
publicbyte?Byte2{get;set;}

[Column("STRING2")]
[StringLength(100)]
publicstringString2{get;set;}
[Column("STRING3")]
publicstringString3{get;set;}
[Column("ENUM2")]
publicEnum2?Enum2{get;set;}

[Column("TESTINT3")]
[Description("Int2")]
publicint?Int3{get;set;}
}

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

varnullableTableList3=newList<NullableTable>();
varnow=DateTime.Now;
for(inti=0;i<20000;i++)
{
vara=newNullableTable()
{
Int2=2,
Bool2=true,
Byte2=1,
DateTime2=now,
Decimal2=1m,
Decimal3=1.1m,
Double2=1.1,
Float2=(float)1.1,
Guid2=Guid.NewGuid(),
Id=0,
Short2=1,
TimeSpan2=TimeSpan.FromHours(1),
String2="sb",
String3="sb",
Long2=2,
Enum2=Model.Enum2.y,
Int3=4
};
nullableTableList3.Add(a);
}

数据库驱动上的选择是这样的，sqlserver采用微软官方驱动System.Data.SqlClient，Oracle采用官方驱动Oracle.ManagedDataAccess.Core，MySql采用社区驱动MySqlConnector（为啥mysql不采用官方的驱动呢？因为官方的驱动封装的太差了，社区的驱动支持列名映射，同时项目里官方驱动和社区驱动可以共存）。同时快速批量插入均支持异步同步，这里仅演示同步，异步的实现基本一样。
三、SqlServer快速批量插入SqlServer官方提供的批量插入方式是SqlBulkCopy，参数为一个dataTable对象，原生的批量插入代码如下,采用StopWatch类进行计时,测试前都会用DELETE from NullableTable 语句清空表，测试里循环跑5次，获取总时间后除以5获取平均值，合计插入10w条数据。

varsw=newStopwatch();
sw.Start();
for(inti=0;i<5;i++)
{
using(vardbConnection=newSqlConnection(connectionString))
{
dbConnection.Open();

SqlBulkCopysqlBulkCopy=newSqlBulkCopy(dbConnection,SqlBulkCopyOptions.KeepIdentity,
null);
sqlBulkCopy.BatchSize=20000;
sqlBulkCopy.DestinationTableName="NullableTable";
//针对列名做一下映射
sqlBulkCopy.ColumnMappings.Add("Int2","Int2");
sqlBulkCopy.ColumnMappings.Add("Bool2","Bool2");
sqlBulkCopy.ColumnMappings.Add("Byte2","Byte2");
sqlBulkCopy.ColumnMappings.Add("DateTime2","DateTime2");
sqlBulkCopy.ColumnMappings.Add("Decimal2","Decimal2");
sqlBulkCopy.ColumnMappings.Add("Decimal3","Decimal3");
sqlBulkCopy.ColumnMappings.Add("Double2","Double2");
sqlBulkCopy.ColumnMappings.Add("Float2","Float2");
sqlBulkCopy.ColumnMappings.Add("Guid2","Guid2");
sqlBulkCopy.ColumnMappings.Add("Short2","Short2");
sqlBulkCopy.ColumnMappings.Add("TimeSpan2","TimeSpan2");

sqlBulkCopy.ColumnMappings.Add("String2","String2");
sqlBulkCopy.ColumnMappings.Add("String3","String3");
sqlBulkCopy.ColumnMappings.Add("Long2","Long2");
sqlBulkCopy.ColumnMappings.Add("Enum2","Enum2");
sqlBulkCopy.ColumnMappings.Add("Int3","TestInt3");
//将实体类列表转换成dataTable
vartable=nullableTableList3.ToDataTable();
sqlBulkCopy.WriteToServer(table);
}

}
sw.Stop();
vartotalTime=sw.ElapsedMilliseconds;
varavgValue=totalTime/5;

实验结果如下SqlServer中：采用快速批量插入10w条数据，时间合计1858毫秒，平均插入2w条数据仅需371毫秒。采用insert into语句，循环插入10w条数据，时间合计457606毫秒，平均插入2w条数据需91521毫秒。四、实体类列表转dataTable的扩展方法这里有一个实体类列表转dataTable的扩展方法，采用的是表达式树+构建委托的方式，性能不错，大家可以参考，代码实现如下。

publicstaticConcurrentDictionary<string,object>CacheDictionary=newConcurrentDictionary<string,object>();
///<summary>
///构建一个object数据转换成一维数组数据的委托
///</summary>
///<paramname="objType"></param>
///<paramname="propertyInfos"></param>
///<returns></returns>
publicstaticFunc<T,object[]>BuildObjectGetValuesDelegate<T>(List<PropertyInfo>propertyInfos)whereT:class
{
varobjParameter=Expression.Parameter(typeof(T),"model");
varselectExpressions=propertyInfos.Select(it=>BuildObjectGetValueExpression(objParameter,it));
vararrayExpression=Expression.NewArrayInit(typeof(object),selectExpressions);
varresult=Expression.Lambda<Func<T,object[]>>(arrayExpression,objParameter).Compile();
returnresult;
}

///<summary>
///构建对象获取单个值得
///</summary>
///<paramname="modelExpression"></param>
///<paramname="propertyInfo"></param>
///<returns></returns>
publicstaticExpressionBuildObjectGetValueExpression(ParameterExpressionmodelExpression,PropertyInfopropertyInfo)
{
varpropertyExpression=Expression.Property(modelExpression,propertyInfo);
varconvertExpression=Expression.Convert(propertyExpression,typeof(object));
returnconvertExpression;
}

publicstaticDataTableToDataTable<T>(thisIEnumerable<T>source,List<PropertyInfo>propertyInfos=null,booluseColumnAttribute=false)whereT:class
{
vartable=newDataTable("template");
if(propertyInfos==null||propertyInfos.Count==0)
{
propertyInfos=typeof(T).GetProperties().Where(it=>it.CanRead).ToList();
}
foreach(varpropertyInfoinpropertyInfos)
{
varcolumnName=useColumnAttribute?(propertyInfo.GetCustomAttribute<ColumnAttribute>()?.Name??propertyInfo.Name):propertyInfo.Name;
table.Columns.Add(columnName,ChangeType(propertyInfo.PropertyType));
}

Func<T,object[]>func;
varkey=typeof(T).FullName+propertyInfos.Select(it=>it.Name).ToList().StringJoin();
if(CacheDictionary.TryGetValue(key,outvarcacheFunc))
{
func=(Func<T,object[]>)cacheFunc;
}
else
{
func=BuildObjectGetValuesDelegate<T>(propertyInfos);
CacheDictionary.TryAdd(key,func);
}

foreach(varmodelinsource)
{
varrowData=func(model);
table.Rows.Add(rowData);
}

returntable;
}

privatestaticTypeChangeType(Typetype)
{
if(type.IsNullable())
{
type=Nullable.GetUnderlyingType(type);
}

returntype;
}

五、Oracle快速批量插入

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

vartotal=20000;
varsw=newStopwatch();
sw.Start();
for(inti=0;i<5;i++)
{
varconnection=newOracleConnection(connectionString);
connection.Open();
int?[]Int2=newint?[total];
bool[]Bool2=newbool[total];
byte[]Byte2=newbyte[total];
DateTime[]DateTime2=newDateTime[total];
decimal?[]Decimal2=newdecimal?[total];
decimal[]Decimal3=newdecimal[total];
double[]Double2=newdouble[total];
float[]Float2=newfloat[total];
Guid?[]Guid2=newGuid?[total];
short[]Short2=newshort[total];
TimeSpan[]TimeSpan2=newTimeSpan[total];
string[]String2=newstring[total];
string[]String3=newstring[total];
long[]Long2=newlong[total];
Enum2[]Enum2=newEnum2[total];

for(intj=0;j<total;j++)
{
Int2[j]=2;
Bool2[j]=true;
Byte2[j]=1;
DateTime2[j]=now;
Decimal2[j]=1m;
Decimal3[j]=1.1m;
Double2[j]=1.1;
Float2[j]=(float)1.1;
Guid2[j]=Guid.NewGuid();
Short2[j]=1;
TimeSpan2[j]=TimeSpan.FromHours(1);
String2[j]="sb";
String3[j]="sb";
Long2[j]=2;
Enum2[j]=Model.Enum2.y;
}

varc=(int)Model.Enum2.y;
OracleParameterpInt2=newOracleParameter();
pInt2.OracleDbType=OracleDbType.Int32;
pInt2.Value=Int2;

OracleParameterpBool2=newOracleParameter();
pBool2.OracleDbType=OracleDbType.Byte;
pBool2.Value=Bool2;

OracleParameterpByte2=newOracleParameter();
pByte2.OracleDbType=OracleDbType.Byte;
pByte2.Value=Byte2;

OracleParameterpDateTime2=newOracleParameter();
pDateTime2.OracleDbType=OracleDbType.TimeStamp;
pDateTime2.Value=DateTime2;

OracleParameterpDecimal2=newOracleParameter();
pDecimal2.OracleDbType=OracleDbType.Decimal;
pDecimal2.Value=Decimal2;

OracleParameterpDecimal3=newOracleParameter();
pDecimal3.OracleDbType=OracleDbType.Decimal;
pDecimal3.Value=Decimal3;

OracleParameterpDouble2=newOracleParameter();
pDouble2.OracleDbType=OracleDbType.Double;
pDouble2.Value=Double2;

OracleParameterpFloat2=newOracleParameter();
pFloat2.OracleDbType=OracleDbType.BinaryFloat;
pFloat2.Value=Float2;


OracleParameterpGuid2=newOracleParameter();
pGuid2.OracleDbType=OracleDbType.Raw;
pGuid2.Value=Guid2;

OracleParameterpShort2=newOracleParameter();
pShort2.OracleDbType=OracleDbType.Int16;
pShort2.Value=Short2;

OracleParameterpTimeSpan2=newOracleParameter();
pTimeSpan2.OracleDbType=OracleDbType.IntervalDS;
pTimeSpan2.Value=TimeSpan2;

OracleParameterpString2=newOracleParameter();
pString2.OracleDbType=OracleDbType.Varchar2;
pString2.Value=String2;

OracleParameterpString3=newOracleParameter();
pString3.OracleDbType=OracleDbType.Varchar2;
pString3.Value=String3;


OracleParameterpLong2=newOracleParameter();
pLong2.OracleDbType=OracleDbType.Long;
pLong2.Value=Long2;

OracleParameterpEnum2=newOracleParameter();
pEnum2.OracleDbType=OracleDbType.Byte;
pEnum2.Value=Enum2;
//createcommandandsetproperties
OracleCommandcmd=connection.CreateCommand();
cmd.CommandText=
"INSERTINTONULLABLETABLE(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)";
cmd.ArrayBindCount=total;
cmd.Parameters.Add(pInt2);
cmd.Parameters.Add(pLong2);
cmd.Parameters.Add(pFloat2);
cmd.Parameters.Add(pDouble2);
cmd.Parameters.Add(pDecimal2);
cmd.Parameters.Add(pDecimal3);
cmd.Parameters.Add(pGuid2);
cmd.Parameters.Add(pShort2);
cmd.Parameters.Add(pDateTime2);
cmd.Parameters.Add(pBool2);
cmd.Parameters.Add(pTimeSpan2);
cmd.Parameters.Add(pByte2);
cmd.Parameters.Add(pString2);
cmd.Parameters.Add(pString3);
cmd.Parameters.Add(pEnum2);
cmd.ExecuteNonQuery();
}
sw.Stop();

vartotalTime=sw.ElapsedMilliseconds;
varavgValue=totalTime/5;

实验结果如下，Oracle中：采用快速批量插入10w条数据，时间合计2323毫秒，平均插入2w条数据仅需464毫秒。采用insert into语句，循环插入10w条数据，时间合计462837毫秒，平均插入2w条数据仅需92567毫秒。六、MySql快速批量插入MySql社区驱动MySqlConnector提供的批量插入方式是SqlBulkCopy，基于MySql自身的文件上传机制进行批量插入，参数为一个dataTable对象，原生的批量插入代码如下,计时方式与SqlServer相同,同时，MySql的连接字符串里要添加";AllowLoadLocalInfile=true"，即连接字符串的形式应该是"Server= ;Database=;User ID=;Password=;AllowLoadLocalInfile=true"，同时在MySql数据库上执行"set global local_infile=1"开启批量上传

varsw=newStopwatch();
sw.Start();
for(intj=0;j<5;j++)
{
using(vardbConnection=newMySqlConnection(connectionString))
{
dbConnection.Open();

MySqlBulkCopysqlBulkCopy=newMySqlBulkCopy(dbConnection,null);
sqlBulkCopy.DestinationTableName="NullableTable";
varpropertys=typeof(NullableTable).GetProperties()
.Where(it=>it.CanReadit.GetCustomAttribute<NotMappedAttribute>()==null).ToList();

for(inti=0;i<propertys.Count;i++)
{
varproperty=propertys[i];
varcolumnName=property.GetCustomAttribute<ColumnAttribute>()?.Name??property.Name;

if(property.PropertyType.GetUnderlyingType()==typeof(Guid))
{
sqlBulkCopy.ColumnMappings.Add(newMySqlBulkCopyColumnMapping(i,"@tmp",
$"{columnName}=unhex(@tmp)"));
}
else
{
sqlBulkCopy.ColumnMappings.Add(newMySqlBulkCopyColumnMapping(i,columnName));
}
}

vartable=nullableTableList3.ToDataTable();

SbUtil.ReplaceDataTableColumnType<Guid,byte[]>(table,guid1=>guid1.ToByteArray());
varc=sqlBulkCopy.WriteToServer(table);
}
}

sw.Stop();

vartotalTime=sw.ElapsedMilliseconds;
varavgValue=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的代码实现如下：

///<summary>
///替换dataTable里的列类型
///</summary>
///<paramname="dt"></param>
publicstaticvoidReplaceDataTableColumnType<OldType,NewType>(DataTabledt,Func<OldType,NewType>replaceFunc)
{
varneedUpdateColumnIndexList=newList<int>();
varneedUpdateColumnNameList=newList<string>();

for(inti=0;i<dt.Columns.Count;i++)
{
varcolumn=dt.Columns[i];
if(column.DataType.GetUnderlyingType()==typeof(OldType))
{
needUpdateColumnIndexList.Add(i);
needUpdateColumnNameList.Add(column.ColumnName);

}
}

if(needUpdateColumnIndexList.Count==0)
{
return;
}

varnameMapping=newDictionary<string,string>();
for(inti=0;i<needUpdateColumnIndexList.Count;i++)
{
varoldColumnName=needUpdateColumnNameList[i];
varnewColumnName=Guid.NewGuid().ToString("N");
nameMapping.Add(newColumnName,oldColumnName);

dt.Columns.Add(newColumnName,typeof(byte[])).SetOrdinal(needUpdateColumnIndexList[i]);
for(intj=0;j<dt.Rows.Count;j++)
{
varc=(dt.Rows[j][oldColumnName]);
dt.Rows[j][newColumnName]=replaceFunc((OldType)(dt.Rows[j][oldColumnName]));
}
dt.Columns.Remove(oldColumnName);
}

for(inti=0;i<dt.Columns.Count;i++)
{
varcolumnName=dt.Columns[i].ColumnName;
if(nameMapping.ContainsKey(columnName))
{
dt.Columns[i].ColumnName=nameMapping[columnName];
}
}
}

七、SummerBoot对各数据库快速批量插入的封装

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

services.AddSummerBoot();
services.AddSummerBootRepository(it=>
{
it.DbConnectionType=typeof(SqlConnection);
it.ConnectionString=connectionString;
});

[AutoRepository]
publicinterfaceINullableTableRepository:IBaseRepository<NullableTable>
{

}

varsw=newStopwatch();
sw.Start();
for(inti=0;i<5;i++)
{
nullableTableRepository.FastBatchInsert(nullableTableList3);
}
sw.Stop();

vartotalTime=sw.ElapsedMilliseconds;
varavgValue=totalTime/5;

实验结果如下SqlServer中：采用SummerBoot统一封装后快速批量插入10w条数据，时间合计3926（原生快速批量写法1858）毫秒，平均插入2w条数据仅需785(原生快速批量写法371)毫秒。从对比可以看出，经过SummerBoot封装后，快速批量插入所花费的时间有所增加，但是对于这么大数据量而言，这点多消耗的时间和节省的开发量对比，不值一提。最后SummerBoot：https://github.com/tripleview/summerboot是一款声明式编程框架，专注于”做什么”而不是”如何去做”，更多用法，可参考SummerBoot文档，同时各位看官，如果你觉得这篇文章还不错的话，请帮忙一键三连。

转自：三合视角链接：cnblogs.com/hezp/p/16519431.html