轉載：http://www.cnblogs.com/luluping/archive/2012/08/09/2629515.htmlhtml
在以前只知道SqlServer支持數據批量插入，卻不知道Oracle、SQLite和MySql也是支持的，不過Oracle須要使用Orace.DataAccess驅動，今天就貼出幾種數據庫的批量插入解決方法。c++
首先說一下，IProvider裏有一個用於實現批量插入的插件服務接口IBatcherProvider，此接口在前一篇文章中已經提到過了。sql
/// <summary>
/// 提供數據批量處理的方法。
/// </summary>
public interface IBatcherProvider : IProviderService
{
/// <summary>
/// 將 <see cref="DataTable"/> 的數據批量插入到數據庫中。
/// </summary>
/// <param name="dataTable">要批量插入的 <see cref="DataTable"/>。</param>
/// <param name="batchSize">每批次寫入的數據量。</param>
void Insert(DataTable dataTable, int batchSize = 10000);
}
1、SqlServer數據批量插入數據庫
SqlServer的批量插入很簡單，使用SqlBulkCopy就能夠，如下是該類的實現：數組
/// <summary>
/// 爲 System.Data.SqlClient 提供的用於批量操做的方法。
/// </summary>
public sealed class MsSqlBatcher : IBatcherProvider
{
/// <summary>
/// 獲取或設置提供者服務的上下文。
/// </summary>
public ServiceContext ServiceContext { get; set; }
/// <summary>
/// 將 <see cref="DataTable"/> 的數據批量插入到數據庫中。
/// </summary>
/// <param name="dataTable">要批量插入的 <see cref="DataTable"/>。</param>
/// <param name="batchSize">每批次寫入的數據量。</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();
//給表名加上先後導符
var tableName = DbUtility.FormatByQuote(ServiceContext.Database.Provider.GetService<ISyntaxProvider>(), dataTable.TableName);
using (var bulk = new SqlBulkCopy(connection, SqlBulkCopyOptions.KeepIdentity, null)
{
DestinationTableName = tableName,
BatchSize = batchSize
})
{
//循環全部列，爲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();
}
}
}
}
以上沒有使用事務，使用事務在性能上會有必定的影響，若是要使用事務，能夠設置SqlBulkCopyOptions.UseInternalTransaction。緩存
2、Oracle數據批量插入app
System.Data.OracleClient不支持批量插入，所以只能使用Oracle.DataAccess組件來做爲提供者。ide
/// <summary>
/// Oracle.Data.Access 組件提供的用於批量操做的方法。
/// </summary>
public sealed class OracleAccessBatcher : IBatcherProvider
{
/// <summary>
/// 獲取或設置提供者服務的上下文。
/// </summary>
public ServiceContext ServiceContext { get; set; }
/// <summary>
/// 將 <see cref="DataTable"/> 的數據批量插入到數據庫中。
/// </summary>
/// <param name="dataTable">要批量插入的 <see cref="DataTable"/>。</param>
/// <param name="batchSize">每批次寫入的數據量。</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>
/// 生成插入數據的sql語句。
/// </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();
//將一個DataTable的數據轉換爲數組的數組
var data = table.ToArray();
//設置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轉爲數組的數組表示，即object[][]，前數組的上標是列的個數，後數組是行的個數，所以循環Columns將後數組做爲Parameter的值，也就是說，參數的值是一個數組。而insert語句與通常的插入語句沒有什麼不同。post
3、SQLite數據批量插入性能
SQLite的批量插入只需開啓事務就能夠了，這個具體的原理不得而知。
public sealed class SQLiteBatcher : IBatcherProvider
{
/// <summary>
/// 獲取或設置提供者服務的上下文。
/// </summary>
public ServiceContext ServiceContext { get; set; }
/// <summary>
/// 將 <see cref="DataTable"/> 的數據批量插入到數據庫中。
/// </summary>
/// <param name="dataTable">要批量插入的 <see cref="DataTable"/>。</param>
/// <param name="batchSize">每批次寫入的數據量。</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;
//首次建立參數，是爲了使用緩存
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>
/// 生成插入數據的sql語句。
/// </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);
}
}
4、MySql數據批量插入
/// <summary>
/// 爲 MySql.Data 組件提供的用於批量操做的方法。
/// </summary>
public sealed class MySqlBatcher : IBatcherProvider
{
/// <summary>
/// 獲取或設置提供者服務的上下文。
/// </summary>
public ServiceContext ServiceContext { get; set; }
/// <summary>
/// 將 <see cref="DataTable"/> 的數據批量插入到數據庫中。
/// </summary>
/// <param name="dataTable">要批量插入的 <see cref="DataTable"/>。</param>
/// <param name="batchSize">每批次寫入的數據量。</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>
/// 生成插入數據的sql語句。
/// </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>
/// 判斷是否爲字符串類別。
/// </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>
/// 建立參數。
/// </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)
{
//若是生成所有的參數，則速度會很慢，所以，只有數據類型爲字符串(包含'號)和日期型時才添加參數
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的批量插入，是將值所有寫在語句的values裏，例如，insert batcher(id, name) values(1, '1', 2, '2', 3, '3', ........ 10, '10')。
5、測試
接下來寫一個測試用例來看一下使用批量插入的效果。
[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));
//構造100000條數據
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表的數據量
var sql = new SqlCommand("SELECT COUNT(1) FROM Batcher");
Console.WriteLine("當前共有 {0} 條數據", database.ExecuteScalar(sql));
})));
}