c#几种数据库的大数据批量插入(SqlServer、Oracle、SQLite和MySql)

在之前只知道SqlServer支持数据批量插入,殊不知道Oracle、SQLite和MySql也是支持的,不过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();
        }
      }
    }
  }

以上没有使用事务,使用事务在性能上会有一定的影响,如果要使用事务,可以设置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);
    }
  }

四、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')。

五、测试

接下来写一个测试用例来看一下使用批量插入的效果。

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

          })));
    }

以下表中列出了四种数据库生成10万条数据各耗用的时间


数据库


耗用时间

MsSql 00:00:02.9376300
Oracle 00:00:01.5155959
SQLite 00:00:01.6275634
MySql 00:00:05.4166891
时间: 2016-11-08

c#连接sqlserver数据库、插入数据、从数据库获取时间示例

c#连接sqlserver.插入数据.从数据库获取时间 复制代码 代码如下: using System;using System.Data.SqlClient; namespace Test{    //连接数据库    public class Connection    {        private static string connectionString =            "Server = 192.168.1.222;" +            "D

C#查询SqlServer数据库并返回单个值的方法

本文实例讲述了C#查询SqlServer数据库并返回单个值的方法.分享给大家供大家参考.具体实现方法如下: static public string GetSqlAsString(string sqlText, SqlParameter[] sqlParameters, string databaseConnectionString) { string result = ""; SqlDataReader reader; SqlConnection connection = new S

c#操作sqlserver数据库的简单示例

1.在用windows模式登陆sql server 数据库 简历一个student的数据库,然后新建查询: 复制代码 代码如下: create table student( id     int  auto_increment  primary key, name char(10) not null, sex    char(10) not null, age   char(10) not null,   ) 2.在vs中新建一个项目,输入一下代码: 复制代码 代码如下: using Syste

C#实现Excel表数据导入Sql Server数据库中的方法

本文实例讲述了C#实现Excel表数据导入Sql Server数据库中的方法.分享给大家供大家参考,具体如下: Excel表数据导入Sql Server数据库的方法很多,这里只是介绍了其中一种: 1.首先,我们要先在test数据库中新建一个my_test表,该表具有三个字段tid int类型, tname nvarchar类型, tt nvarchar类型 (注意:my_test表中的数据类型必须与Excel中相应字段的类型一致) 2. 我们用SELECT * FROM  OPENROWSET(

C#访问SQL Server数据库的实现方法

本文实例讲述了C#访问SQL Server数据库的实现方法.分享给大家供大家参考.具体如下: 这里介绍一下用C#访问SQL Server数据库.这里要涉及到几个类:SqlConnection类.SqlCommand类.SqlDataReader类. SqlConnection类用于建立与数据库的连接,SqlCommand类用于创建SQL命令,SqlDataReader对象是执行SqlCommand对象返回结果集的查询语句后的结果. 下面是部分代码: /*其中server表示服务器,"."

C#实现异步连接Sql Server数据库的方法

本文实例讲述了C#实现异步连接Sql Server数据库的方法.分享给大家供大家参考.具体分析如下: .net最新版提供了await方法,可以使我们可以很容易实现到数据库的异步连接 复制代码 代码如下: readonly string ConnectionString = "Data Source=database_server_name;Initial Catalog=Store;Integrated Security=True"; protected async void Exec

C#如何实现对sql server数据库的增删改查

一个专门实现sql server数据库的增删改查,以及将查询的结果返回成表格等功能,分享代码如下 using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Data.SqlClient;//第一步:引用与sql相关的命名空间 using System.Data;//引用表的命名空间 //行注释 ///段落注释 /// <summary> ///第二步: 把

C#连接到sql server2008数据库的实例代码

废话不多说了,直接给大家贴代码了,具体代码如下所示: namespace MyFirstApp { class Program { static void Main(string[] args) { SqlConnection conn = null; SqlCommand comm = null; SqlDataReader sdreader = null; try { string ConStr = "server=192.168.1.110;uid=sa;pwd=woaifr0828;da

C#编程实现连接SQL SERVER数据库实例详解

本文实例讲述了C#编程实现连接SQL SERVER数据库.分享给大家供大家参考,具体如下: 首先,在SQL SEVER里建立一个名为"Exercise"的数据库名,在该数据库下建立一张名为"lianxi"的表.好,现在开始写代码. 在FORM1里拖一个DATAGIRDVIEW用于显示表,在放一个BUTTON用于执行操作.加入命名空间 复制代码 代码如下: using system.data.sqlclient; private void button1_Click(

C#实现连接SQL Server2012数据库并执行SQL语句的方法

本文实例讲述了C#实现连接SQL Server2012数据库并执行SQL语句的方法.分享给大家供大家参考,具体如下: 开发工具:Visual Studio 2012 数据库: SQL Server 2012 使用Visual Studio时还是直接和微软自家的SQL Server数据库连接比较方便,就像使用Eclipse时和MySQL连接便捷一样的道理 无论使用什么工具步骤都一样: 1. 首先保证相关工具都已经正确安装了 2. 开启数据库连接服务 3. 在开发工具中通过用户名和口令与数据库进行关

C#获取所有SQL Server数据库名称的方法

本文实例讲述了C#获取所有SQL Server数据库名称的方法.分享给大家供大家参考.具体如下: 连接到master库中,执行 select name from master..sysdatabases . sp_helpdb也可以列出数据名,但不是全部,sp_helpdb是会根据当前用户角色找出相应的数据库,而用sysdatabases表则只要有权限访问该表就能查出所有数据库信息. 默认情况下,sqlserver提供的6个数据库中,除了model数据库外,其他若干个数据库还可能拥有guest用

C#将图片存放到SQL SERVER数据库中的方法

本文实例讲述了C#将图片存放到SQL SERVER数据库中的方法.分享给大家供大家参考.具体如下: 第一步: //获取当前选择的图片 this.pictureBox1.Image = Image.FromStream(this.openFileDialog1.OpenFile()); //获取当前图片的路径 string path = openFileDialog1.FileName.ToString(); //将制定路径的图片添加到FileStream类中 FileStream fs = ne

还原sql server数据库的处理方法

如何从备份设备还原备份(企业管理器) 从备份设备还原备份 展开服务器组,然后展开服务器. 展开"数据库"文件夹,右击数据库,指向"所有任务"子菜单,然后单击"还原数据库"命令. 在"还原为数据库"框中,如果要还原的数据库名称与显示的默认数据库名称不同,请在其中进行输入或选择.若要用新名称还原数据库,请输入新的数据库名称. 说明 为数据库指定新名称将自动为从数据库备份中还原的数据库文件指定新名称. 单击"从设备&quo

MS SQL SERVER 数据库日志压缩方法与代码

MS SQL性能是很不错的,但是数据库用了一段时间之后,数据库却变得很大,实际的数据量不大.一般都是数据库日志引起的!数据库日志的增长可以达到好几百M. 网上的MSSQL虚拟主机价格也贵,要想不让数据库超容,只好压缩下数据库日志,或者删除数据库日志. 下面我给大家介绍一个方法 1.打开企业管理器 2.打开要处理的数据库 3.点击菜单>工具>SQL查询分析器 4.在输入窗口里面输入: 复制代码 代码如下: DUMP TRANSACTION [数据库名] WITH  NO_LOG  BACKUP 

SQL Server数据库安装时常见问题解决方案集锦

本文我们总结了几个在安装SQL Server数据库时常见问题的解决方案,供初学者学习参考,接下来让我们来一起看一下吧. 常见问题一: 安装Sql Server 2000时出现"以前进行的程序创建了挂起的文件操作,运行安装程序之前,必须重新启动计算机" ,重启后仍然无效. 解决方案: 1.不用退出Sql Server 2000安装程序,直接切换到桌面. 2.打开注册表编辑器(在"运行"中敲入"regedit"之后回车即可),定位到注册表的HKEY_

易语言调用MS SQL SERVER数据库方法

Microsoft SQL SERVER 数据库存储过程,根据其输入输出数据,笼统的可以分为以下几种情况或其组合:无输入,有一个或多个输入参数,无输出,直接返回(return)一个值,通过output参数返回一个或多个值,返回一个记录集(recordset).无论哪一种情况,无论输入输出参数多复杂的存储过程,都可以在易语言中正确调用,准确的传入参数,并获取正确的输出数据.下面我(liigo)分多种情况介绍在易语言中调用MS SQL SERVER数据库存储过程的详细方法,使用数据库操作支持库(eD

SQL Server数据库入门学习总结

一图胜"十"言:SQL Server 数据库总结 一个大概的总结 经过一段时间的学习,也对数据库有了一些认识. 数据库基本是由表,关系,操作组成:对于初学者首先要学的: 1.数据库是如何存储数据的 表,约束,触发器 2.数据库是如何操作数据的 insert,update,delete T-sql 函数 存储过程 触发器 3.数据库是如何显示数据的 select SQLServer数据库学习总结 1.SQL基础 SQL Server2000安装.配置,服务器启动.停止,企业管理器.查询分

SQL Server 数据库分离与附加(图文教程)

一.概述 SQL Server提供了"分离/附加"数据库."备份/还原"数据库.复制数据库等多种数据库的备份和恢复方法.这里介绍一种学习中常用的"分离/附加"方法,类似于大家熟悉的"文件拷贝"方法,即把数据库文件(.MDF)和对应的日志文件(.LDF)拷贝到其它磁盘上作备份,然后把这两个文件再拷贝到任何需要这个数据库的系统之中.比如,在实验教学过程中,同学们常常想把自己在学校实验室计算机中创建的数据库搬迁到自己的计算机中而不想