时间:2021-05-02
目录
关于 SqlServer 批量插入的方式,有三种比较常用的插入方式,Insert、BatchInsert、SqlBulkCopy,下面我们对比以下三种方案的速度
1.普通的Insert插入方法
? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 public static void Insert(IEnumerable<Person> persons) { using (var con = new SqlConnection("Server=.;Database=DemoDataBase;User ID=sa;Password=8888;")) { con.Open(); foreach (var person in persons) { using (var com = new SqlCommand( "INSERT INTO dbo.Person(Id,Name,Age,CreateTime,Sex)VALUES(@Id,@Name,@Age,@CreateTime,@Sex)", con)) { com.Parameters.AddRange(new[] { new SqlParameter("@Id", SqlDbType.BigInt) {Value = person.Id}, new SqlParameter("@Name", SqlDbType.VarChar, 64) {Value = person.Name}, new SqlParameter("@Age", SqlDbType.Int) {Value = person.Age}, new SqlParameter("@CreateTime", SqlDbType.DateTime) {Value = person.CreateTime ?? (object) DBNull.Value}, new SqlParameter("@Sex", SqlDbType.Int) {Value = (int)person.Sex}, }); com.ExecuteNonQuery(); } } } }2.拼接BatchInsert插入语句
? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 public static void BatchInsert(Person[] persons) { using (var con = new SqlConnection("Server=.;Database=DemoDataBase;User ID=sa;Password=8888;")) { con.Open(); var pageCount = (persons.Length - 1) / 1000 + 1; for (int i = 0; i < pageCount; i++) { var personList = persons.Skip(i * 1000).Take(1000).ToArray(); var values = personList.Select(p => $"({p.Id},'{p.Name}',{p.Age},{(p.CreateTime.HasValue ? $"'{p.CreateTime:yyyy-MM-dd HH:mm:ss}'" : "NULL")},{(int) p.Sex})"); var insertSql = $"INSERT INTO dbo.Person(Id,Name,Age,CreateTime,Sex)VALUES{string.Join(",", values)}"; using (var com = new SqlCommand(insertSql, con)) { com.ExecuteNonQuery(); } } } }3.SqlBulkCopy插入方案
? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 public static void BulkCopy(IEnumerable<Person> persons) { using (var con = new SqlConnection("Server=.;Database=DemoDataBase;User ID=sa;Password=8888;")) { con.Open(); var table = new DataTable(); table.Columns.AddRange(new [] { new DataColumn("Id", typeof(long)), new DataColumn("Name", typeof(string)), new DataColumn("Age", typeof(int)), new DataColumn("CreateTime", typeof(DateTime)), new DataColumn("Sex", typeof(int)), }); foreach (var p in persons) { table.Rows.Add(new object[] {p.Id, p.Name, p.Age, p.CreateTime, (int) p.Sex}); } using (var copy = new SqlBulkCopy(con)) { copy.DestinationTableName = "Person"; copy.WriteToServer(table); } } }3.三种方案速度对比
方案 数量 时间 Insert 1千条 145.4351ms BatchInsert 1千条 103.9061ms SqlBulkCopy 1千条 7.021ms Insert 1万条 1501.326ms BatchInsert 1万条 850.6274ms SqlBulkCopy 1万条 30.5129ms Insert 10万条 13875.4934ms BatchInsert 10万条 8278.9056ms SqlBulkCopy 10万条 314.8402ms两者插入效率对比,Insert明显比SqlBulkCopy要慢太多,大概20~40倍性能差距,下面我们将SqlBulkCopy封装一下,让批量插入更加方便
1.方法介绍
批量插入扩展方法签名
方法 方法参数 介绍 BulkCopy 同步的批量插入方法 SqlConnection connection sql server 连接对象 IEnumerable<T> source 需要批量插入的数据源 string tableName = null 插入表名称【为NULL默认为实体名称】 int bulkCopyTimeout = 30 批量插入超时时间 int batchSize = 0 写入数据库一批数量【如果为0代表全部一次性插入】最合适数量【这取决于您的环境,尤其是行数和网络延迟。就个人而言,我将从BatchSize属性设置为1000行开始,然后看看其性能如何。如果可行,那么我将使行数加倍(例如增加到2000、4000等),直到性能下降或超时。否则,如果超时发生在1000,那么我将行数减少一半(例如500),直到它起作用为止。】 SqlBulkCopyOptions options = SqlBulkCopyOptions.Default 批量复制参数 SqlTransaction externalTransaction = null 执行的事务对象 BulkCopyAsync 异步的批量插入方法 SqlConnection connection sql server 连接对象 IEnumerable<T> source 需要批量插入的数据源 string tableName = null 插入表名称【为NULL默认为实体名称】 int bulkCopyTimeout = 30 批量插入超时时间 int batchSize = 0 写入数据库一批数量【如果为0代表全部一次性插入】最合适数量【这取决于您的环境,尤其是行数和网络延迟。就个人而言,我将从BatchSize属性设置为1000行开始,然后看看其性能如何。如果可行,那么我将使行数加倍(例如增加到2000、4000等),直到性能下降或超时。否则,如果超时发生在1000,那么我将行数减少一半(例如500),直到它起作用为止。】 SqlBulkCopyOptions options = SqlBulkCopyOptions.Default 批量复制参数 SqlTransaction externalTransaction = null 执行的事务对象这个方法主要解决了两个问题:
此方案也是在我公司中使用,以满足公司的批量插入数据的需求,例如第三方的对账数据此方法使用的是Expression动态生成数据转换函数,其效率和手写的原生代码差不多,和原生手写代码相比,多余的转换损失很小【最大的性能损失都是在值类型拆装箱上】
此方案和其他网上的方案有些不同的是:不是将List先转换成DataTable,然后写入SqlBulkCopy的,而是使用一个实现IDataReader的读取器包装List,每往SqlBulkCopy插入一行数据才会转换一行数据
IDataReader方案和DataTable方案相比优点
效率高:DataTable方案需要先完全转换后,才能交由SqlBulkCopy写入数据库,而IDataReader方案可以边转换边交给SqlBulkCopy写入数据库(例如:10万数据插入速度可提升30%)
占用内存少:DataTable方案需要先完全转换后,才能交由SqlBulkCopy写入数据库,需要占用大量内存,而IDataReader方案可以边转换边交给SqlBulkCopy写入数据库,无须占用过多内存
强大:因为是边写入边转换,而且EnumerableReader传入的是一个迭代器,可以实现持续插入数据的效果
2.实现原理
① 实体Model与表映射
数据库表代码
? 1 2 3 4 5 6 7 8 9 10 11 CREATE TABLE [dbo].[Person]( [Id] [BIGINT] NOT NULL, [Name] [VARCHAR](64) NOT NULL, [Age] [INT] NOT NULL, [CreateTime] [DATETIME] NULL, [Sex] [INT] NOT NULL, PRIMARY KEY CLUSTERED ( [Id] ASC )WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ON [PRIMARY] ) ON [PRIMARY]实体类代码
? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 public class Person { public long Id { get; set; } public string Name { get; set; } public int Age { get; set; } public DateTime? CreateTime { get; set; } public Gender Sex { get; set; } } public enum Gender { Man = 0, Woman = 1 }② 实体转换成数据行
将数据转换成数据行采用的是:反射+Expression来完成
其中反射是用于获取编写Expression所需程序类,属性等信息
其中Expression是用于生成高效转换函数其中ModelToDataTable<TModel>类型利用了静态泛型类特性,实现泛型参数的缓存效果
在ModelToDataTable<TModel>的静态构造函数中,生成转换函数,获取需要转换的属性信息,并存入静态只读字段中,完成缓存
③ 使用IDataReader插入数据的重载
EnumerableReader是实现了IDataReader接口的读取类,用于将模型对象,在迭代器中读取出来,并转换成数据行,可供SqlBulkCopy读取
SqlBulkCopy只会调用三个方法:GetOrdinal、Read、GetValue
3.完整代码
扩展方法类
? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 public static class SqlConnectionExtension { /// <summary> /// 批量复制 /// </summary> /// <typeparam name="TModel">插入的模型对象</typeparam> /// <param name="source">需要批量插入的数据源</param> /// <param name="connection">数据库连接对象</param> /// <param name="tableName">插入表名称【为NULL默认为实体名称】</param> /// <param name="bulkCopyTimeout">插入超时时间</param> /// <param name="batchSize">写入数据库一批数量【如果为0代表全部一次性插入】最合适数量【这取决于您的环境,尤其是行数和网络延迟。就个人而言,我将从BatchSize属性设置为1000行开始,然后看看其性能如何。如果可行,那么我将使行数加倍(例如增加到2000、4000等),直到性能下降或超时。否则,如果超时发生在1000,那么我将行数减少一半(例如500),直到它起作用为止。】</param> /// <param name="options">批量复制参数</param> /// <param name="externalTransaction">执行的事务对象</param> /// <returns>插入数量</returns> public static int BulkCopy<TModel>(this SqlConnection connection, IEnumerable<TModel> source, string tableName = null, int bulkCopyTimeout = 30, int batchSize = 0, SqlBulkCopyOptions options = SqlBulkCopyOptions.Default, SqlTransaction externalTransaction = null) { //创建读取器 using (var reader = new EnumerableReader<TModel>(source)) { //创建批量插入对象 using (var copy = new SqlBulkCopy(connection, options, externalTransaction)) { //插入的表 copy.DestinationTableName = tableName ?? typeof(TModel).Name; //写入数据库一批数量 copy.BatchSize = batchSize; //超时时间 copy.BulkCopyTimeout = bulkCopyTimeout; //创建字段映射【如果没有此字段映射会导致数据填错位置,如果类型不对还会导致报错】【因为:没有此字段映射默认是按照列序号对应插入的】 foreach (var column in ModelToDataTable<TModel>.Columns) { //创建字段映射 copy.ColumnMappings.Add(column.ColumnName, column.ColumnName); } //将数据批量写入数据库 copy.WriteToServer(reader); //返回插入数据数量 return reader.Depth; } } } /// <summary> /// 批量复制-异步 /// </summary> /// <typeparam name="TModel">插入的模型对象</typeparam> /// <param name="source">需要批量插入的数据源</param> /// <param name="connection">数据库连接对象</param> /// <param name="tableName">插入表名称【为NULL默认为实体名称】</param> /// <param name="bulkCopyTimeout">插入超时时间</param> /// <param name="batchSize">写入数据库一批数量【如果为0代表全部一次性插入】最合适数量【这取决于您的环境,尤其是行数和网络延迟。就个人而言,我将从BatchSize属性设置为1000行开始,然后看看其性能如何。如果可行,那么我将使行数加倍(例如增加到2000、4000等),直到性能下降或超时。否则,如果超时发生在1000,那么我将行数减少一半(例如500),直到它起作用为止。】</param> /// <param name="options">批量复制参数</param> /// <param name="externalTransaction">执行的事务对象</param> /// <returns>插入数量</returns> public static async Task<int> BulkCopyAsync<TModel>(this SqlConnection connection, IEnumerable<TModel> source, string tableName = null, int bulkCopyTimeout = 30, int batchSize = 0, SqlBulkCopyOptions options = SqlBulkCopyOptions.Default, SqlTransaction externalTransaction = null) { //创建读取器 using (var reader = new EnumerableReader<TModel>(source)) { //创建批量插入对象 using (var copy = new SqlBulkCopy(connection, options, externalTransaction)) { //插入的表 copy.DestinationTableName = tableName ?? typeof(TModel).Name; //写入数据库一批数量 copy.BatchSize = batchSize; //超时时间 copy.BulkCopyTimeout = bulkCopyTimeout; //创建字段映射【如果没有此字段映射会导致数据填错位置,如果类型不对还会导致报错】【因为:没有此字段映射默认是按照列序号对应插入的】 foreach (var column in ModelToDataTable<TModel>.Columns) { //创建字段映射 copy.ColumnMappings.Add(column.ColumnName, column.ColumnName); } //将数据批量写入数据库 await copy.WriteToServerAsync(reader); //返回插入数据数量 return reader.Depth; } } } }封装的迭代器数据读取器
? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 /// <summary> /// 迭代器数据读取器 /// </summary> /// <typeparam name="TModel">模型类型</typeparam> public class EnumerableReader<TModel> : IDataReader { /// <summary> /// 实例化迭代器读取对象 /// </summary> /// <param name="source">模型源</param> public EnumerableReader(IEnumerable<TModel> source) { _source = source ?? throw new ArgumentNullException(nameof(source)); _enumerable = source.GetEnumerator(); } private readonly IEnumerable<TModel> _source; private readonly IEnumerator<TModel> _enumerable; private object[] _currentDataRow = Array.Empty<object>(); private int _depth; private bool _release; public void Dispose() { _release = true; _enumerable.Dispose(); } public int GetValues(object[] values) { if (values == null) throw new ArgumentNullException(nameof(values)); var length = Math.Min(_currentDataRow.Length, values.Length); Array.Copy(_currentDataRow, values, length); return length; } public int GetOrdinal(string name) { for (int i = 0; i < ModelToDataTable<TModel>.Columns.Count; i++) { if (ModelToDataTable<TModel>.Columns[i].ColumnName == name) return i; } return -1; } public long GetBytes(int ordinal, long dataIndex, byte[] buffer, int bufferIndex, int length) { if (dataIndex < 0) throw new Exception($"起始下标不能小于0!"); if (bufferIndex < 0) throw new Exception("目标缓冲区起始下标不能小于0!"); if (length < 0) throw new Exception("读取长度不能小于0!"); var numArray = (byte[])GetValue(ordinal); if (buffer == null) return numArray.Length; if (buffer.Length <= bufferIndex) throw new Exception("目标缓冲区起始下标不能大于目标缓冲区范围!"); var freeLength = Math.Min(numArray.Length - bufferIndex, length); if (freeLength <= 0) return 0; Array.Copy(numArray, dataIndex, buffer, bufferIndex, length); return freeLength; } public long GetChars(int ordinal, long dataIndex, char[] buffer, int bufferIndex, int length) { if (dataIndex < 0) throw new Exception($"起始下标不能小于0!"); if (bufferIndex < 0) throw new Exception("目标缓冲区起始下标不能小于0!"); if (length < 0) throw new Exception("读取长度不能小于0!"); var numArray = (char[])GetValue(ordinal); if (buffer == null) return numArray.Length; if (buffer.Length <= bufferIndex) throw new Exception("目标缓冲区起始下标不能大于目标缓冲区范围!"); var freeLength = Math.Min(numArray.Length - bufferIndex, length); if (freeLength <= 0) return 0; Array.Copy(numArray, dataIndex, buffer, bufferIndex, length); return freeLength; } public bool IsDBNull(int i) { var value = GetValue(i); return value == null || value is DBNull; } public bool NextResult() { //移动到下一个元素 if (!_enumerable.MoveNext()) return false; //行层+1 Interlocked.Increment(ref _depth); //得到数据行 _currentDataRow = ModelToDataTable<TModel>.ToRowData.Invoke(_enumerable.Current); return true; } public byte GetByte(int i) => (byte)GetValue(i); public string GetName(int i) => ModelToDataTable<TModel>.Columns[i].ColumnName; public string GetDataTypeName(int i) => ModelToDataTable<TModel>.Columns[i].DataType.Name; public Type GetFieldType(int i) => ModelToDataTable<TModel>.Columns[i].DataType; public object GetValue(int i) => _currentDataRow[i]; public bool GetBoolean(int i) => (bool)GetValue(i); public char GetChar(int i) => (char)GetValue(i); public Guid GetGuid(int i) => (Guid)GetValue(i); public short GetInt16(int i) => (short)GetValue(i); public int GetInt32(int i) => (int)GetValue(i); public long GetInt64(int i) => (long)GetValue(i); public float GetFloat(int i) => (float)GetValue(i); public double GetDouble(int i) => (double)GetValue(i); public string GetString(int i) => (string)GetValue(i); public decimal GetDecimal(int i) => (decimal)GetValue(i); public DateTime GetDateTime(int i) => (DateTime)GetValue(i); public IDataReader GetData(int i) => throw new NotSupportedException(); public int FieldCount => ModelToDataTable<TModel>.Columns.Count; public object this[int i] => GetValue(i); public object this[string name] => GetValue(GetOrdinal(name)); public void Close() => Dispose(); public DataTable GetSchemaTable() => ModelToDataTable<TModel>.ToDataTable(_source); public bool Read() => NextResult(); public int Depth => _depth; public bool IsClosed => _release; public int RecordsAffected => 0; }模型对象转数据行工具类
? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 /// <summary> /// 对象转换成DataTable转换类 /// </summary> /// <typeparam name="TModel">泛型类型</typeparam> public static class ModelToDataTable<TModel> { static ModelToDataTable() { //如果需要剔除某些列可以修改这段代码 var propertyList = typeof(TModel).GetProperties().Where(w => w.CanRead).ToArray(); Columns = new ReadOnlyCollection<DataColumn>(propertyList .Select(pr => new DataColumn(pr.Name, GetDataType(pr.PropertyType))).ToArray()); //生成对象转数据行委托 ToRowData = BuildToRowDataDelegation(typeof(TModel), propertyList); } /// <summary> /// 构建转换成数据行委托 /// </summary> /// <param name="type">传入类型</param> /// <param name="propertyList">转换的属性</param> /// <returns>转换数据行委托</returns> private static Func<TModel, object[]> BuildToRowDataDelegation(Type type, PropertyInfo[] propertyList) { var source = Expression.Parameter(type); var items = propertyList.Select(property => ConvertBindPropertyToData(source, property)); var array = Expression.NewArrayInit(typeof(object), items); var lambda = Expression.Lambda<Func<TModel, object[]>>(array, source); return lambda.Compile(); } /// <summary> /// 将属性转换成数据 /// </summary> /// <param name="source">源变量</param> /// <param name="property">属性信息</param> /// <returns>获取属性数据表达式</returns> private static Expression ConvertBindPropertyToData(ParameterExpression source, PropertyInfo property) { var propertyType = property.PropertyType; var expression = (Expression)Expression.Property(source, property); if (propertyType.IsEnum) expression = Expression.Convert(expression, propertyType.GetEnumUnderlyingType()); return Expression.Convert(expression, typeof(object)); } /// <summary> /// 获取数据类型 /// </summary> /// <param name="type">属性类型</param> /// <returns>数据类型</returns> private static Type GetDataType(Type type) { //枚举默认转换成对应的值类型 if (type.IsEnum) return type.GetEnumUnderlyingType(); //可空类型 if (type.IsGenericType && type.GetGenericTypeDefinition() == typeof(Nullable<>)) return GetDataType(type.GetGenericArguments().First()); return type; } /// <summary> /// 列集合 /// </summary> public static IReadOnlyList<DataColumn> Columns { get; } /// <summary> /// 对象转数据行委托 /// </summary> public static Func<TModel, object[]> ToRowData { get; } /// <summary> /// 集合转换成DataTable /// </summary> /// <param name="source">集合</param> /// <param name="tableName">表名称</param> /// <returns>转换完成的DataTable</returns> public static DataTable ToDataTable(IEnumerable<TModel> source, string tableName = "TempTable") { //创建表对象 var table = new DataTable(tableName); //设置列 foreach (var dataColumn in Columns) { table.Columns.Add(new DataColumn(dataColumn.ColumnName, dataColumn.DataType)); } //循环转换每一行数据 foreach (var item in source) { table.Row声明:本页内容来源网络,仅供用户参考;我单位不保证亦不表示资料全面及准确无误,也不保证亦不表示这些资料为最新信息,如因任何原因,本网内容或者用户因倚赖本网内容造成任何损失或损害,我单位将不会负任何法律责任。如涉及版权问题,请提交至online#300.cn邮箱联系删除。
前言在项目开发中遇到了需要批量插入数据和更新数据的操作,但是在某度上搜并没有找到有用的东西,于是到stackoverflow中搜到如下解决方案:实践一、Bulk
本文实例讲述了C#实现SQL批量插入数据到表的方法。分享给大家供大家参考,具体如下:#region帮助实例:SQL批量插入数据多种方法//////SqlBulk
通过Python脚本批量生成插入数据的SQL语句原始SQL语句:INSERTINTOsystem_user(id,login_name,name,passwor
在操作数据库时,经常会碰到批量插入、批量删除的情况,直接执行SQL语句还好做一点,当使用Mybatis进行批量插入、批量删除时会有一些问题。下面对使用Mybat
java解决分布式环境中高并发环境下数据插入重复问题前言原因:服务器同时接受到的重复请求现象:数据重复插入/修改操作解决方案:分布式锁对请求报文生成摘要信息+r