a_bu

 

   SqlBulkCopy  做为SQL Server 官方 批量入库类,性能不会太差。其事务部份官方说明较模糊,因此 针对事务配置做了一些测试。

 

  A. 先准备测试场景 ,关于SqlBulkCopyOptions.KeepIdentity 应用。  新建两张表 , 每张表列a 做为自增列,同时做为主键 , 其中 test_sqlbulk 源表 ,先添加一组数据。

CREATE TABLE [dbo].[test_sqlbulk](
    [a] [int] IDENTITY(1,1) NOT NULL,
    [b] [int] NULL,
 CONSTRAINT [PK_test_sqlbulk] PRIMARY KEY CLUSTERED 
(
    [a] ASC
)WITH (PAD_INDEX  = OFF, STATISTICS_NORECOMPUTE  = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS  = ON, ALLOW_PAGE_LOCKS  = ON) ON [PRIMARY]
) ON [PRIMARY]




CREATE TABLE [dbo].[test_sqlbulk_des](
    [a] [int] IDENTITY(1,1) NOT NULL,
    [b] [int] NULL,
 CONSTRAINT [PK_test_sqlbulk_des] PRIMARY KEY CLUSTERED 
(
    [a] ASC
)WITH (PAD_INDEX  = OFF, STATISTICS_NORECOMPUTE  = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS  = ON, ALLOW_PAGE_LOCKS  = ON) ON [PRIMARY]
) ON [PRIMARY]

 

添加数据:

 

测试代码:

 SqlConnection sqlConn = new SqlConnection( sourceConnection) ;
            sqlConn.Open() ;
            SqlCommand commandSourceData = new SqlCommand("select * from test_sqlbulk ", sqlConn);
            SqlDataReader reader = commandSourceData.ExecuteReader();

            // Set up the bulk copy object using the KeepIdentity option. 
            using (SqlBulkCopy bulkCopy = new SqlBulkCopy(
                       sourceConnection))
            {
                bulkCopy.BatchSize = 2;
                bulkCopy.DestinationTableName =
                    "dbo.test_sqlbulk_des";

                // Write from the source to the destination.
                // This should fail with a duplicate key error
                // after some of the batches have been copied.
                try
                {
                    bulkCopy.WriteToServer(reader);
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.Message);
                }
                finally
                {
                    reader.Close();
                }
            }
View Code

 

再看看目标表 test_sqlbulk_des 结果 , 由于SqlBulkCopy  构造参数没有 KeepIdentity 标识,目标表自增列与源表不一致 :

test_sqlbulk_des 查询结果:

 

 =>调整代码,增加 SqlBulkCopyOptions.KeepIdentity 选项,再次导入:

 using (SqlBulkCopy bulkCopy = new SqlBulkCopy(
                       sourceConnection, SqlBulkCopyOptions.KeepIdentity))
            {
                bulkCopy.BatchSize = 2;
                bulkCopy.DestinationTableName =
                    "dbo.test_sqlbulk_des";

                // Write from the source to the destination.
                // This should fail with a duplicate key error
                // after some of the batches have been copied.
                try
                {
                    bulkCopy.WriteToServer(reader);
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.Message);
                }
                finally
                {
                    reader.Close();
                }
            }

 

 =>目标表,源表自增列一致了:

 

B. 测试无事务,内部事务(SqlBulkCopyOptions.UseInternalTransaction) , 外部传入事务 之间区别 , 以下所有测试均事先在目标表保留一行数据 使copy 过程中造成主键冲突,查看回退情况。

1. 无事务测试。初使化源表,目标表数据:

 

测试代码 , 不意外,如下代码最终会产生主键冲突错误, 执行结束后,再查询源表,目标表数据 。 BatchSize 设置为2 , 结果表明在第二批次主键冲突, 第一批次数据成功提交了(没有显示声明事务), 再将 BatchSize 设置为3 验证结果。

 SqlConnection sqlConn = new SqlConnection( sourceConnection) ;
            sqlConn.Open() ;
            SqlCommand commandSourceData = new SqlCommand("select * from test_sqlbulk ", sqlConn);
            SqlDataReader reader = commandSourceData.ExecuteReader();

            // Set up the bulk copy object using the KeepIdentity option. 
            using (SqlBulkCopy bulkCopy = new SqlBulkCopy(
                       sourceConnection, SqlBulkCopyOptions.KeepIdentity))
            {
                bulkCopy.BatchSize = 2;
                bulkCopy.DestinationTableName =
                    "dbo.test_sqlbulk_des";

                // Write from the source to the destination.
                // This should fail with a duplicate key error
                // after some of the batches have been copied.
                try
                {
                    bulkCopy.WriteToServer(reader);
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.Message);
                }
                finally
                {
                    reader.Close();
                }
            }

 

BatchSize = 2 结果:

 

BatchSize = 3 结果:

 

 

2.指定内部事务 , 分别将 BatchSize 指定为 2 , 3, 4  。如下执行结果与上文示例 结果一致, 因此无论是否指定 SqlBulkCopyOptions.UseInternalTransaction 参数(不包括外部传入事务) ,SqlBulkCopy 内部按照一个批次一个事务。在出现异常时只回滚当前批次 , 在此之前成功执行批次不回滚。 因此若需全部回退,需将  BatchSize 设为 总记录条数 ,此时所有数据做为一个批次提交 ,  异常回退批次,回退数据范围即全部数据。

 

测试代码:

 // Set up the bulk copy object using the KeepIdentity option. 
            using (SqlBulkCopy bulkCopy = new SqlBulkCopy(
                       sourceConnection, SqlBulkCopyOptions.KeepIdentity | SqlBulkCopyOptions.UseInternalTransaction))
            {
                bulkCopy.BatchSize = 2;
                bulkCopy.DestinationTableName =
                    "dbo.test_sqlbulk_des";

                // Write from the source to the destination.
                // This should fail with a duplicate key error
                // after some of the batches have been copied.
                try
                {
                    bulkCopy.WriteToServer(reader);
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.Message);
                }
                finally
                {
                    reader.Close();
                }
            }

 

 =>BatchSize = 2  , 目标表结果:

 =>BatchSize = 3  , 目标表结果:

 

3. 外部传入事务对象场景 , 如果 使用SqlBulkCopy入库操作只做为本次业务处理一个子集,在整个业务处理环节中异常情况,需要回滚整个业务操作。 本次再新建一张表:test_sqlbulk_update , 测试过程中,先对该表增加一条记录,再重复之前异常场景。 结束时,之前新增记录也回退了。

因此, 当外部传入事务对象时,SqlBulkCopy  内部不提交事务, 在异常情况,有可能会自动回滚。

CREATE TABLE [dbo].[test_sqlbulk_update](
    [a] [int] IDENTITY(1,1) NOT NULL,
    [b] [int] NULL,
)
View Code

 

SqlConnection sqlConn = new SqlConnection( sourceConnection) ;
            sqlConn.Open() ;

            SqlTransaction tran = sqlConn.BeginTransaction();

            SqlCommand commandInsert = new SqlCommand("insert into [test_sqlbulk_update]( b ) values (1) ", sqlConn, tran);

            int result = commandInsert.ExecuteNonQuery();

            SqlCommand commandSourceData = new SqlCommand("select * from test_sqlbulk ", sqlConn, tran);
            
            SqlDataReader reader = commandSourceData.ExecuteReader();
            
            // Set up the bulk copy object using the KeepIdentity option. 
            using (SqlBulkCopy bulkCopy = new SqlBulkCopy(sqlConn, SqlBulkCopyOptions.KeepIdentity, tran))
            {
                bulkCopy.BatchSize = 2;
                bulkCopy.DestinationTableName =
                    "dbo.test_sqlbulk_des";

                // Write from the source to the destination.
                // This should fail with a duplicate key error
                // after some of the batches have been copied.
                try
                {
                    bulkCopy.WriteToServer(reader);
                    reader.Close();
                    tran.Commit();
                }
                catch (Exception ex)
                {
                    reader.Close();
                    tran.Rollback();
            // tran.Commit(); 异常仍提交执行,同时注释上一行。 } }

 

 => 异常时执行 rollback, 最终结果:没做任何修改.

 

 

=> 外部事务,Catch 中执行 Commint ,发生异常,因为自动回滚了,但在Catch 执行 Rollback 无异常。

SqlTransaction tran = sqlConn.BeginTransaction();


            SqlCommand commandSourceData = new SqlCommand("select * from test_sqlbulk ", new SqlConnection(sourceConnection));
            
            SqlDataAdapter ada = new SqlDataAdapter(commandSourceData ) ;
            System.Data.DataSet ds = new System.Data.DataSet();
            ada.Fill(ds);
            
            // Set up the bulk copy object using the KeepIdentity option. 
            using (SqlBulkCopy bulkCopy = new SqlBulkCopy(sqlConn, SqlBulkCopyOptions.KeepIdentity, tran))
            {
                bulkCopy.BatchSize = 2;
                bulkCopy.DestinationTableName =
                    "dbo.test_sqlbulk_des";

                // Write from the source to the destination.
                // This should fail with a duplicate key error
                // after some of the batches have been copied.
                try
                {
                    
                    bulkCopy.WriteToServer(ds.Tables[0]);

                    SqlCommand commandInsert = new SqlCommand("insert into [test_sqlbulk_update]( b ) values ('a') ", sqlConn, tran);

                    int result = commandInsert.ExecuteNonQuery();


                    tran.Commit();
                }
                catch (Exception ex)
                {
                    tran.Commit();
                }
               
            }
View Code

 

 

相关文章: