在之前只知道SqlServer支持?jǐn)?shù)據(jù)批量插入,殊不知道Oracle、SQLite和MySql也是支持的,不過(guò)Oracle需要使用Orace.DataAccess驅(qū)動(dòng),今天就貼出幾種數(shù)據(jù)庫(kù)的批量插入解決方法。
首先說(shuō)一下,IProvider里有一個(gè)用于實(shí)現(xiàn)批量插入的插件服務(wù)接口IBatcherProvider,此接口在前一篇文章中已經(jīng)提到過(guò)了。

/// <summary> /// 提供數(shù)據(jù)批量處理的方法。 /// </summary> public interface IBatcherProvider : IProviderService { /// <summary> /// 將 <see cref="DataTable"/> 的數(shù)據(jù)批量插入到數(shù)據(jù)庫(kù)中。 /// </summary> /// <param name="dataTable"> 要批量插入的 <see cref="DataTable"/> 。 </param> /// <param name="batchSize"> 每批次寫(xiě)入的數(shù)據(jù)量。 </param> void Insert(DataTable dataTable, int batchSize = 10000 ); }
一、SqlServer數(shù)據(jù)批量插入
??? SqlServer的批量插入很簡(jiǎn)單,使用SqlBulkCopy就可以,以下是該類(lèi)的實(shí)現(xiàn):

/// <summary> /// 為 System.Data.SqlClient 提供的用于批量操作的方法。 /// </summary> public sealed class MsSqlBatcher : IBatcherProvider { /// <summary> /// 獲取或設(shè)置提供者服務(wù)的上下文。 /// </summary> public ServiceContext ServiceContext { get ; set ; } /// <summary> /// 將 <see cref="DataTable"/> 的數(shù)據(jù)批量插入到數(shù)據(jù)庫(kù)中。 /// </summary> /// <param name="dataTable"> 要批量插入的 <see cref="DataTable"/> 。 </param> /// <param name="batchSize"> 每批次寫(xiě)入的數(shù)據(jù)量。 </param> public void Insert(DataTable dataTable, int batchSize = 10000 ) { Checker.ArgumentNull(dataTable, " dataTable " ); if (dataTable.Rows.Count == 0 ) { return ; } using ( var connection = (SqlConnection)ServiceContext.Database.CreateConnection()) { try { connection.TryOpen(); // 給表名加上前后導(dǎo)符 var tableName = DbUtility.FormatByQuote(ServiceContext.Database.Provider.GetService<ISyntaxProvider> (), dataTable.TableName); using ( var bulk = new SqlBulkCopy(connection, SqlBulkCopyOptions.KeepIdentity, null ) { DestinationTableName = tableName, BatchSize = batchSize }) { // 循環(huán)所有列,為bulk添加映射 dataTable.EachColumn(c => bulk.ColumnMappings.Add(c.ColumnName, c.ColumnName), c => ! c.AutoIncrement); bulk.WriteToServer(dataTable); bulk.Close(); } } catch (Exception exp) { throw new BatcherException(exp); } finally { connection.TryClose(); } } } }
SqlBulkCopy的ColumnMappings中列的名稱(chēng)受大小寫(xiě)敏感限制,因此在構(gòu)造DataTable的時(shí)候應(yīng)請(qǐng)注意列名要與表一致。
以上沒(méi)有使用事務(wù),使用事務(wù)在性能上會(huì)有一定的影響,如果要使用事務(wù),可以設(shè)置SqlBulkCopyOptions.UseInternalTransaction。
二、Oracle數(shù)據(jù)批量插入
???? System.Data.OracleClient不支持批量插入,因此只能使用Oracle.DataAccess組件來(lái)作為提供者。

/// <summary> /// Oracle.Data.Access 組件提供的用于批量操作的方法。 /// </summary> public sealed class OracleAccessBatcher : IBatcherProvider { /// <summary> /// 獲取或設(shè)置提供者服務(wù)的上下文。 /// </summary> public ServiceContext ServiceContext { get ; set ; } /// <summary> /// 將 <see cref="DataTable"/> 的數(shù)據(jù)批量插入到數(shù)據(jù)庫(kù)中。 /// </summary> /// <param name="dataTable"> 要批量插入的 <see cref="DataTable"/> 。 </param> /// <param name="batchSize"> 每批次寫(xiě)入的數(shù)據(jù)量。 </param> public void Insert(DataTable dataTable, int batchSize = 10000 ) { Checker.ArgumentNull(dataTable, " dataTable " ); if (dataTable.Rows.Count == 0 ) { return ; } using ( var connection = ServiceContext.Database.CreateConnection()) { try { connection.TryOpen(); using ( var command = ServiceContext.Database.Provider.DbProviderFactory.CreateCommand()) { if (command == null ) { throw new BatcherException( new ArgumentException( " command " )); } command.Connection = connection; command.CommandText = GenerateInserSql(ServiceContext.Database, command, dataTable); command.ExecuteNonQuery(); } } catch (Exception exp) { throw new BatcherException(exp); } finally { connection.TryClose(); } } } /// <summary> /// 生成插入數(shù)據(jù)的sql語(yǔ)句。 /// </summary> /// <param name="database"></param> /// <param name="command"></param> /// <param name="table"></param> /// <returns></returns> private string GenerateInserSql(IDatabase database, DbCommand command, DataTable table) { var names = new StringBuilder(); var values = new StringBuilder(); // 將一個(gè)DataTable的數(shù)據(jù)轉(zhuǎn)換為數(shù)組的數(shù)組 var data = table.ToArray(); // 設(shè)置ArrayBindCount屬性 command.GetType().GetProperty( " ArrayBindCount " ).SetValue(command, table.Rows.Count, null ); var syntax = database.Provider.GetService<ISyntaxProvider> (); for ( var i = 0 ; i < table.Columns.Count; i++ ) { var column = table.Columns[i]; var parameter = database.Provider.DbProviderFactory.CreateParameter(); if (parameter == null ) { continue ; } parameter.ParameterName = column.ColumnName; parameter.Direction = ParameterDirection.Input; parameter.DbType = column.DataType.GetDbType(); parameter.Value = data[i]; if (names.Length > 0 ) { names.Append( " , " ); values.Append( " , " ); } names.AppendFormat( " {0} " , DbUtility.FormatByQuote(syntax, column.ColumnName)); values.AppendFormat( " {0}{1} " , syntax.ParameterPrefix, column.ColumnName); command.Parameters.Add(parameter); } return string .Format( " INSERT INTO {0}({1}) VALUES ({2}) " , DbUtility.FormatByQuote(syntax, table.TableName), names, values); } }
以上最重要的一步,就是將DataTable轉(zhuǎn)為數(shù)組的數(shù)組表示,即object[][],前數(shù)組的上標(biāo)是列的個(gè)數(shù),后數(shù)組是行的個(gè)數(shù),因此循環(huán) Columns將后數(shù)組作為Parameter的值,也就是說(shuō),參數(shù)的值是一個(gè)數(shù)組。而insert語(yǔ)句與一般的插入語(yǔ)句沒(méi)有什么不一樣。
三、SQLite數(shù)據(jù)批量插入
???? SQLite的批量插入只需開(kāi)啟事務(wù)就可以了,這個(gè)具體的原理不得而知。

public sealed class SQLiteBatcher : IBatcherProvider { /// <summary> /// 獲取或設(shè)置提供者服務(wù)的上下文。 /// </summary> public ServiceContext ServiceContext { get ; set ; } /// <summary> /// 將 <see cref="DataTable"/> 的數(shù)據(jù)批量插入到數(shù)據(jù)庫(kù)中。 /// </summary> /// <param name="dataTable"> 要批量插入的 <see cref="DataTable"/> 。 </param> /// <param name="batchSize"> 每批次寫(xiě)入的數(shù)據(jù)量。 </param> public void Insert(DataTable dataTable, int batchSize = 10000 ) { Checker.ArgumentNull(dataTable, " dataTable " ); if (dataTable.Rows.Count == 0 ) { return ; } using ( var connection = ServiceContext.Database.CreateConnection()) { DbTransaction transcation = null ; try { connection.TryOpen(); transcation = connection.BeginTransaction(); using ( var command = ServiceContext.Database.Provider.DbProviderFactory.CreateCommand()) { if (command == null ) { throw new BatcherException( new ArgumentException( " command " )); } command.Connection = connection; command.CommandText = GenerateInserSql(ServiceContext.Database, dataTable); if (command.CommandText == string .Empty) { return ; } var flag = new AssertFlag(); dataTable.EachRow(row => { var first = flag.AssertTrue(); ProcessCommandParameters(dataTable, command, row, first); command.ExecuteNonQuery(); }); } transcation.Commit(); } catch (Exception exp) { if (transcation != null ) { transcation.Rollback(); } throw new BatcherException(exp); } finally { connection.TryClose(); } } } private void ProcessCommandParameters(DataTable dataTable, DbCommand command, DataRow row, bool first) { for ( var c = 0 ; c < dataTable.Columns.Count; c++ ) { DbParameter parameter; // 首次創(chuàng)建參數(shù),是為了使用緩存 if (first) { parameter = ServiceContext.Database.Provider.DbProviderFactory.CreateParameter(); parameter.ParameterName = dataTable.Columns[c].ColumnName; command.Parameters.Add(parameter); } else { parameter = command.Parameters[c]; } parameter.Value = row[c]; } } /// <summary> /// 生成插入數(shù)據(jù)的sql語(yǔ)句。 /// </summary> /// <param name="database"></param> /// <param name="table"></param> /// <returns></returns> private string GenerateInserSql(IDatabase database, DataTable table) { var syntax = database.Provider.GetService<ISyntaxProvider> (); var names = new StringBuilder(); var values = new StringBuilder(); var flag = new AssertFlag(); table.EachColumn(column => { if (! flag.AssertTrue()) { names.Append( " , " ); values.Append( " , " ); } names.Append(DbUtility.FormatByQuote(syntax, column.ColumnName)); values.AppendFormat( " {0}{1} " , syntax.ParameterPrefix, column.ColumnName); }); return string .Format( " INSERT INTO {0}({1}) VALUES ({2}) " , DbUtility.FormatByQuote(syntax, table.TableName), names, values); } }
四、MySql數(shù)據(jù)批量插入

/// <summary> /// 為 MySql.Data 組件提供的用于批量操作的方法。 /// </summary> public sealed class MySqlBatcher : IBatcherProvider { /// <summary> /// 獲取或設(shè)置提供者服務(wù)的上下文。 /// </summary> public ServiceContext ServiceContext { get ; set ; } /// <summary> /// 將 <see cref="DataTable"/> 的數(shù)據(jù)批量插入到數(shù)據(jù)庫(kù)中。 /// </summary> /// <param name="dataTable"> 要批量插入的 <see cref="DataTable"/> 。 </param> /// <param name="batchSize"> 每批次寫(xiě)入的數(shù)據(jù)量。 </param> public void Insert(DataTable dataTable, int batchSize = 10000 ) { Checker.ArgumentNull(dataTable, " dataTable " ); if (dataTable.Rows.Count == 0 ) { return ; } using ( var connection = ServiceContext.Database.CreateConnection()) { try { connection.TryOpen(); using ( var command = ServiceContext.Database.Provider.DbProviderFactory.CreateCommand()) { if (command == null ) { throw new BatcherException( new ArgumentException( " command " )); } command.Connection = connection; command.CommandText = GenerateInserSql(ServiceContext.Database, command, dataTable); if (command.CommandText == string .Empty) { return ; } command.ExecuteNonQuery(); } } catch (Exception exp) { throw new BatcherException(exp); } finally { connection.TryClose(); } } } /// <summary> /// 生成插入數(shù)據(jù)的sql語(yǔ)句。 /// </summary> /// <param name="database"></param> /// <param name="command"></param> /// <param name="table"></param> /// <returns></returns> private string GenerateInserSql(IDatabase database, DbCommand command, DataTable table) { var names = new StringBuilder(); var values = new StringBuilder(); var types = new List<DbType> (); var count = table.Columns.Count; var syntax = database.Provider.GetService<ISyntaxProvider> (); table.EachColumn(c => { if (names.Length > 0 ) { names.Append( " , " ); } names.AppendFormat( " {0} " , DbUtility.FormatByQuote(syntax, c.ColumnName)); types.Add(c.DataType.GetDbType()); }); var i = 0 ; foreach (DataRow row in table.Rows) { if (i > 0 ) { values.Append( " , " ); } values.Append( " ( " ); for ( var j = 0 ; j < count; j++ ) { if (j > 0 ) { values.Append( " , " ); } var isStrType = IsStringType(types[j]); var parameter = CreateParameter(database.Provider, isStrType, types[j], row[j], syntax.ParameterPrefix, i, j); if (parameter != null ) { values.Append(parameter.ParameterName); command.Parameters.Add(parameter); } else if (isStrType) { values.AppendFormat( " '{0}' " , row[j]); } else { values.Append(row[j]); } } values.Append( " ) " ); i ++ ; } return string .Format( " INSERT INTO {0}({1}) VALUES {2} " , DbUtility.FormatByQuote(syntax, table.TableName), names, values); } /// <summary> /// 判斷是否為字符串類(lèi)別。 /// </summary> /// <param name="dbType"></param> /// <returns></returns> private bool IsStringType(DbType dbType) { return dbType == DbType.AnsiString || dbType == DbType.AnsiStringFixedLength || dbType == DbType.String || dbType == DbType.StringFixedLength; } /// <summary> /// 創(chuàng)建參數(shù)。 /// </summary> /// <param name="provider"></param> /// <param name="isStrType"></param> /// <param name="dbType"></param> /// <param name="value"></param> /// <param name="parPrefix"></param> /// <param name="row"></param> /// <param name="col"></param> /// <returns></returns> private DbParameter CreateParameter(IProvider provider, bool isStrType, DbType dbType, object value, char parPrefix, int row, int col) { // 如果生成全部的參數(shù),則速度會(huì)很慢,因此,只有數(shù)據(jù)類(lèi)型為字符串(包含'號(hào))和日期型時(shí)才添加參數(shù) if ((isStrType && value.ToString().IndexOf( ' \' ' ) != - 1 ) || dbType == DbType.DateTime) { var name = string .Format( " {0}p_{1}_{2} " , parPrefix, row, col); var parameter = provider.DbProviderFactory.CreateParameter(); parameter.ParameterName = name; parameter.Direction = ParameterDirection.Input; parameter.DbType = dbType; parameter.Value = value; return parameter; } return null ; } }
MySql的批量插入,是將值全部寫(xiě)在語(yǔ)句的values里,例如,insert batcher(id, name) values(1, '1', 2, '2', 3, '3', ........ 10, '10')。
五、測(cè)試
???? 接下來(lái)寫(xiě)一個(gè)測(cè)試用例來(lái)看一下使用批量插入的效果。

[Test] public void TestBatchInsert() { Console.WriteLine(TimeWatcher.Watch(() => InvokeTest(database => { var table = new DataTable( " Batcher " ); table.Columns.Add( " Id " , typeof ( int )); table.Columns.Add( " Name1 " , typeof ( string )); table.Columns.Add( " Name2 " , typeof ( string )); table.Columns.Add( " Name3 " , typeof ( string )); table.Columns.Add( " Name4 " , typeof ( string )); // 構(gòu)造100000條數(shù)據(jù) for ( var i = 0 ; i < 100000 ; i++ ) { table.Rows.Add(i, i.ToString(), i.ToString(), i.ToString(), i.ToString()); } // 獲取 IBatcherProvider var batcher = database.Provider.GetService<IBatcherProvider> (); if (batcher == null ) { Console.WriteLine( " 不支持批量插入。 " ); } else { batcher.Insert(table); } // 輸出batcher表的數(shù)據(jù)量 var sql = new SqlCommand( " SELECT COUNT(1) FROM Batcher " ); Console.WriteLine( " 當(dāng)前共有 {0} 條數(shù)據(jù) " , database.ExecuteScalar(sql)); }))); }
?
更多文章、技術(shù)交流、商務(wù)合作、聯(lián)系博主
微信掃碼或搜索:z360901061

微信掃一掃加我為好友
QQ號(hào)聯(lián)系: 360901061
您的支持是博主寫(xiě)作最大的動(dòng)力,如果您喜歡我的文章,感覺(jué)我的文章對(duì)您有幫助,請(qǐng)用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧,狠狠點(diǎn)擊下面給點(diǎn)支持吧,站長(zhǎng)非常感激您!手機(jī)微信長(zhǎng)按不能支付解決辦法:請(qǐng)將微信支付二維碼保存到相冊(cè),切換到微信,然后點(diǎn)擊微信右上角掃一掃功能,選擇支付二維碼完成支付。
【本文對(duì)您有幫助就好】元
