在之前只知道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ì)您有幫助就好】元
