几种数据库的大数据批量插入

在之前只知道SqlServer支持数据批量插入，殊不知道Oracle、SQLite也是支持的，不过Oracle需要使用Orace.DataAccess驱动，今天就贴出几种数据库的批量插入解决方法。  IProvider里有一个用于实现批量插入的插件服务接口IBatcherProvider。

    /// <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);
    }

一、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();
                }
            }
        }
    }

SqlBulkCopy的ColumnMappings中列的名称受大小写敏感限制，因此在构造DataTable的时候应请注意列名要与表一致。

     以上没有使用事务，使用事务在性能上会有一定的影响，如果要使用事务，可以设置SqlBulkCopyOptions.UseInternalTransaction。

二、Oracle数据批量插入

     System.Data.OracleClient不支持批量插入，因此只能使用Oracle.DataAccess组件来作为提供者。

    /// <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语句与一般的插入语句没有什么不一样。

     三、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);
        }
    }