AsyncCallback productList = new AsyncCallback(ProductList);
SqlConnection conn = new SqlConnection("Data Source=(local); Initial Catalog=NorthWind; Integrated Security=SSPI");
conn.Open();
SqlCommand cmd = new SqlCommand("select top 2 * from orders", conn);
IAsyncResult ia = cmd.BeginExecuteReader(productList, cmd);
可以开始使用异步模式，而不会显著更改现有算法。
using Microsoft.Data.SqlClient;
using System.Threading.Tasks;
class A
static void ProductList(IAsyncResult result) { }
public static void Main()
// AsyncCallback productList = new AsyncCallback(ProductList);
// SqlConnection conn = new SqlConnection("Data Source=(local); Initial Catalog=NorthWind; Integrated Security=SSPI");
// conn.Open();
// SqlCommand cmd = new SqlCommand("select top 2 * from orders", conn);
// IAsyncResult ia = cmd.BeginExecuteReader(productList, cmd);
AsyncCallback productList = new AsyncCallback(ProductList);
SqlConnection conn = new SqlConnection("Data Source=(local); Initial Catalog=NorthWind; Integrated Security=SSPI");
conn.OpenAsync().ContinueWith((task) => {
SqlCommand cmd = new SqlCommand("select top 2 * from orders", conn);
IAsyncResult ia = cmd.BeginExecuteReader(productList, cmd);
}, TaskContinuationOptions.OnlyOnRanToCompletion);
使用基本提供程序模型和异步功能
可能需要创建可连接到不同数据库和执行查询的工具。 可以使用基本提供程序模型和异步功能。
必须在服务器上启用 Microsoft 分布式事务处理控制器 (MSDTC) 以使用分布式事务。 有关如何启用 MSDTC 的信息，请参阅如何在 Web 服务器上启用 MSDTC。
using System.Data.Common;
using Microsoft.Data.SqlClient;
using System.Threading.Tasks;
class program
static async Task PerformDBOperationsUsingProviderModel(string connectionString)
using (DbConnection connection = SqlClientFactory.Instance.CreateConnection())
connection.ConnectionString = connectionString;
await connection.OpenAsync();
DbCommand command = connection.CreateCommand();
command.CommandText = "SELECT * FROM AUTHORS";
using (DbDataReader reader = await command.ExecuteReaderAsync())
while (await reader.ReadAsync())
for (int i = 0; i < reader.FieldCount; i++)
// Process each column as appropriate
object obj = await reader.GetFieldValueAsync<object>(i);
Console.WriteLine(obj);
public static void Main()
SqlConnectionStringBuilder builder = new SqlConnectionStringBuilder();
// replace these with your own values
builder.DataSource = "localhost";
builder.InitialCatalog = "pubs";
builder.IntegratedSecurity = true;
Task task = PerformDBOperationsUsingProviderModel(builder.ConnectionString);
task.Wait();
使用 SQL 事务和新异步功能
using Microsoft.Data.SqlClient;
using System.Threading.Tasks;
class Program
static void Main()
string connectionString =
"Persist Security Info=False;Integrated Security=SSPI;database=Northwind;server=(local)";
Task task = ExecuteSqlTransaction(connectionString);
task.Wait();
static async Task ExecuteSqlTransaction(string connectionString)
using (SqlConnection connection = new SqlConnection(connectionString))
await connection.OpenAsync();
SqlCommand command = connection.CreateCommand();
SqlTransaction transaction = null;
// Start a local transaction.
transaction = await Task.Run<SqlTransaction>(
() => connection.BeginTransaction("SampleTransaction")
// Must assign both transaction object and connection
// to Command object for a pending local transaction
command.Connection = connection;
command.Transaction = transaction;
try {
command.CommandText =
"Insert into Region (RegionID, RegionDescription) VALUES (555, 'Description')";
await command.ExecuteNonQueryAsync();
command.CommandText =
"Insert into Region (RegionID, RegionDescription) VALUES (556, 'Description')";
await command.ExecuteNonQueryAsync();
// Attempt to commit the transaction.
await Task.Run(() => transaction.Commit());
Console.WriteLine("Both records are written to database.");
catch (Exception ex)
Console.WriteLine("Commit Exception Type: {0}", ex.GetType());
Console.WriteLine(" Message: {0}", ex.Message);
// Attempt to roll back the transaction.
transaction.Rollback();
catch (Exception ex2)
// This catch block will handle any errors that may have occurred
// on the server that would cause the rollback to fail, such as
// a closed connection.
Console.WriteLine("Rollback Exception Type: {0}", ex2.GetType());
Console.WriteLine(" Message: {0}", ex2.Message);
使用分布式事务和新异步功能
在企业应用程序中，可能需要在某些情况下添加分布式事务，以支持多个数据库服务器之间的事务。 你可以使用 System.Transactions 命名空间并登记分布式事务，如下所示：
using Microsoft.Data.SqlClient;
using System.Threading.Tasks;
using System.Transactions;
class Program
public static void Main()
SqlConnectionStringBuilder builder = new SqlConnectionStringBuilder();
// replace these with your own values
// create two tables RegionTable1 and RegionTable2
// and add a constraint in one of these tables
// to avoid duplicate RegionID
builder.DataSource = "localhost";
builder.InitialCatalog = "Northwind";
builder.IntegratedSecurity = true;
Task task = ExecuteDistributedTransaction(builder.ConnectionString, builder.ConnectionString);
task.Wait();
static async Task ExecuteDistributedTransaction(string connectionString1, string connectionString2)
using (SqlConnection connection1 = new SqlConnection(connectionString1))
using (SqlConnection connection2 = new SqlConnection(connectionString2))
using (CommittableTransaction transaction = new CommittableTransaction())
await connection1.OpenAsync();
connection1.EnlistTransaction(transaction);
await connection2.OpenAsync();
connection2.EnlistTransaction(transaction);
SqlCommand command1 = connection1.CreateCommand();
command1.CommandText = "Insert into RegionTable1 (RegionID, RegionDescription) VALUES (100, 'Description')";
await command1.ExecuteNonQueryAsync();
SqlCommand command2 = connection2.CreateCommand();
command2.CommandText = "Insert into RegionTable2 (RegionID, RegionDescription) VALUES (100, 'Description')";
await command2.ExecuteNonQueryAsync();
transaction.Commit();
catch (Exception ex)
Console.WriteLine("Exception Type: {0}", ex.GetType());
Console.WriteLine(" Message: {0}", ex.Message);
transaction.Rollback();
catch (Exception ex2)
Console.WriteLine("Rollback Exception Type: {0}", ex2.GetType());
Console.WriteLine(" Message: {0}", ex2.Message);
取消异步操作
可通过使用 CancellationToken 来取消异步请求。
using Microsoft.Data.SqlClient;
using System.Threading;
using System.Threading.Tasks;
namespace Samples
class CancellationSample
public static void Main(string[] args)
CancellationTokenSource source = new CancellationTokenSource();
source.CancelAfter(2000); // give up after 2 seconds
Task result = CancellingAsynchronousOperations(source.Token);
result.Wait();
catch (AggregateException exception)
if (exception.InnerException is SqlException)
Console.WriteLine("Operation canceled");
throw;
static async Task CancellingAsynchronousOperations(CancellationToken cancellationToken)
using (SqlConnection connection = new SqlConnection("Server=(local);Integrated Security=true"))
await connection.OpenAsync(cancellationToken);
SqlCommand command = new SqlCommand("WAITFOR DELAY '00:10:00'", connection);
await command.ExecuteNonQueryAsync(cancellationToken);
使用 SqlBulkCopy 的异步操作
在带有 SqlBulkCopy.WriteToServerAsync 的 Microsoft.Data.SqlClient.SqlBulkCopy 中也提供异步功能。
using System.Data;
using Microsoft.Data.SqlClient;
using System.Threading;
using System.Threading.Tasks;
namespace SqlBulkCopyAsyncCodeSample
class Program
static string selectStatement = "SELECT * FROM [pubs].[dbo].[titles]";
static string createDestTableStatement =
@"CREATE TABLE {0} (
[title_id] [varchar](6) NOT NULL,
[title] [varchar](80) NOT NULL,
[type] [char](12) NOT NULL,
[pub_id] [char](4) NULL,
[price] [money] NULL,
[advance] [money] NULL,
[royalty] [int] NULL,
[ytd_sales] [int] NULL,
[notes] [varchar](200) NULL,
[pubdate] [datetime] NOT NULL)";
// Replace the connection string if needed, for instance to connect to SQL Express: @"Server=(local)\SQLEXPRESS;Database=Demo;Integrated Security=true"
// static string connectionString = @"Server=(localdb)\V11.0;Database=Demo";
static string connectionString = @"Server=(local);Database=Demo;Integrated Security=true";
// static string marsConnectionString = @"Server=(localdb)\V11.0;Database=Demo;MultipleActiveResultSets=true;";
static string marsConnectionString = @"Server=(local);Database=Demo;MultipleActiveResultSets=true;Integrated Security=true";
// Replace the Server name with your actual sql azure server name and User ID/Password
static string azureConnectionString = @"Server=SqlAzure;User ID=<myUserID>;Password=<myPassword>;Database=Demo";
static void Main(string[] args)
SynchronousSqlBulkCopy();
AsyncSqlBulkCopy().Wait();
MixSyncAsyncSqlBulkCopy().Wait();
AsyncSqlBulkCopyNotifyAfter().Wait();
AsyncSqlBulkCopyDataRows().Wait();
AsyncSqlBulkCopySqlServerToSqlAzure().Wait();
AsyncSqlBulkCopyCancel().Wait();
AsyncSqlBulkCopyMARS().Wait();
// 3.1.1 Synchronous bulk copy in .NET 4.5
private static void SynchronousSqlBulkCopy()
using (SqlConnection conn = new SqlConnection(connectionString))
conn.Open();
DataTable dt = new DataTable();
using (SqlCommand cmd = new SqlCommand(selectStatement, conn))
SqlDataAdapter adapter = new SqlDataAdapter(cmd);
adapter.Fill(dt);
string temptable = "[#" + Guid.NewGuid().ToString("N") + "]";
cmd.CommandText = string.Format(createDestTableStatement, temptable);
cmd.ExecuteNonQuery();
using (SqlBulkCopy bcp = new SqlBulkCopy(conn))
bcp.DestinationTableName = temptable;
bcp.WriteToServer(dt);
// 3.1.2 Asynchronous bulk copy in .NET 4.5
private static async Task AsyncSqlBulkCopy()
using (SqlConnection conn = new SqlConnection(connectionString))
await conn.OpenAsync();
DataTable dt = new DataTable();
using (SqlCommand cmd = new SqlCommand(selectStatement, conn))
SqlDataAdapter adapter = new SqlDataAdapter(cmd);
adapter.Fill(dt);
string temptable = "[#" + Guid.NewGuid().ToString("N") + "]";
cmd.CommandText = string.Format(createDestTableStatement, temptable);
await cmd.ExecuteNonQueryAsync();
using (SqlBulkCopy bcp = new SqlBulkCopy(conn))
bcp.DestinationTableName = temptable;
await bcp.WriteToServerAsync(dt);
// 3.2 Add new Async.NET capabilities in an existing application (Mixing synchronous and asynchronous calls)
private static async Task MixSyncAsyncSqlBulkCopy()
using (SqlConnection conn1 = new SqlConnection(connectionString))
conn1.Open();
using (SqlCommand cmd = new SqlCommand(selectStatement, conn1))
using (SqlDataReader reader = cmd.ExecuteReader())
using (SqlConnection conn2 = new SqlConnection(connectionString))
await conn2.OpenAsync();
string temptable = "[#" + Guid.NewGuid().ToString("N") + "]";
SqlCommand createCmd = new SqlCommand(string.Format(createDestTableStatement, temptable), conn2);
await createCmd.ExecuteNonQueryAsync();
using (SqlBulkCopy bcp = new SqlBulkCopy(conn2))
bcp.DestinationTableName = temptable;
await bcp.WriteToServerAsync(reader);
// 3.3 Using the NotifyAfter property
private static async Task AsyncSqlBulkCopyNotifyAfter()
using (SqlConnection conn = new SqlConnection(connectionString))
await conn.OpenAsync();
DataTable dt = new DataTable();
using (SqlCommand cmd = new SqlCommand(selectStatement, conn))
SqlDataAdapter adapter = new SqlDataAdapter(cmd);
adapter.Fill(dt);
string temptable = "[#" + Guid.NewGuid().ToString("N") + "]";
cmd.CommandText = string.Format(createDestTableStatement, temptable);
await cmd.ExecuteNonQueryAsync();
using (SqlBulkCopy bcp = new SqlBulkCopy(conn))
bcp.DestinationTableName = temptable;
bcp.NotifyAfter = 5;
bcp.SqlRowsCopied += new SqlRowsCopiedEventHandler(OnSqlRowsCopied);
await bcp.WriteToServerAsync(dt);
private static void OnSqlRowsCopied(object sender, SqlRowsCopiedEventArgs e)
Console.WriteLine("Copied {0} so far...", e.RowsCopied);
// 3.4 Using the new SqlBulkCopy Async.NET capabilities with DataRow[]
private static async Task AsyncSqlBulkCopyDataRows()
using (SqlConnection conn = new SqlConnection(connectionString))
await conn.OpenAsync();
DataTable dt = new DataTable();
using (SqlCommand cmd = new SqlCommand(selectStatement, conn))
SqlDataAdapter adapter = new SqlDataAdapter(cmd);
adapter.Fill(dt);
DataRow[] rows = dt.Select();
string temptable = "[#" + Guid.NewGuid().ToString("N") + "]";
cmd.CommandText = string.Format(createDestTableStatement, temptable);
await cmd.ExecuteNonQueryAsync();
using (SqlBulkCopy bcp = new SqlBulkCopy(conn))
bcp.DestinationTableName = temptable;
await bcp.WriteToServerAsync(rows);
// 3.5 Copying data from SQL Server to SQL Azure in .NET 4.5
private static async Task AsyncSqlBulkCopySqlServerToSqlAzure()
using (SqlConnection srcConn = new SqlConnection(connectionString))
using (SqlConnection destConn = new SqlConnection(azureConnectionString))
await srcConn.OpenAsync();
await destConn.OpenAsync();
using (SqlCommand srcCmd = new SqlCommand(selectStatement, srcConn))
using (SqlDataReader reader = await srcCmd.ExecuteReaderAsync())
string temptable = "[#" + Guid.NewGuid().ToString("N") + "]";
using (SqlCommand destCmd = new SqlCommand(string.Format(createDestTableStatement, temptable), destConn))
await destCmd.ExecuteNonQueryAsync();
using (SqlBulkCopy bcp = new SqlBulkCopy(destConn))
bcp.DestinationTableName = temptable;
await bcp.WriteToServerAsync(reader);
// 3.6 Cancelling an Asynchronous Operation to SQL Azure
private static async Task AsyncSqlBulkCopyCancel()
CancellationTokenSource cts = new CancellationTokenSource();
using (SqlConnection srcConn = new SqlConnection(connectionString))
using (SqlConnection destConn = new SqlConnection(azureConnectionString))
await srcConn.OpenAsync(cts.Token);
await destConn.OpenAsync(cts.Token);
using (SqlCommand srcCmd = new SqlCommand(selectStatement, srcConn))
using (SqlDataReader reader = await srcCmd.ExecuteReaderAsync(cts.Token))
string temptable = "[#" + Guid.NewGuid().ToString("N") + "]";
using (SqlCommand destCmd = new SqlCommand(string.Format(createDestTableStatement, temptable), destConn))
await destCmd.ExecuteNonQueryAsync(cts.Token);
using (SqlBulkCopy bcp = new SqlBulkCopy(destConn))
bcp.DestinationTableName = temptable;
await bcp.WriteToServerAsync(reader, cts.Token);
//Cancel Async SqlBulCopy Operation after 200 ms
cts.CancelAfter(200);
// 3.7 Using Async.Net and MARS
private static async Task AsyncSqlBulkCopyMARS()
using (SqlConnection marsConn = new SqlConnection(marsConnectionString))
await marsConn.OpenAsync();
SqlCommand titlesCmd = new SqlCommand("SELECT * FROM [pubs].[dbo].[titles]", marsConn);
SqlCommand authorsCmd = new SqlCommand("SELECT * FROM [pubs].[dbo].[authors]", marsConn);
//With MARS we can have multiple active results sets on the same connection
using (SqlDataReader titlesReader = await titlesCmd.ExecuteReaderAsync())
using (SqlDataReader authorsReader = await authorsCmd.ExecuteReaderAsync())
await authorsReader.ReadAsync();
string temptable = "[#" + Guid.NewGuid().ToString("N") + "]";
using (SqlConnection destConn = new SqlConnection(connectionString))
await destConn.OpenAsync();
using (SqlCommand destCmd = new SqlCommand(string.Format(createDestTableStatement, temptable), destConn))
await destCmd.ExecuteNonQueryAsync();
using (SqlBulkCopy bcp = new SqlBulkCopy(destConn))
bcp.DestinationTableName = temptable;
await bcp.WriteToServerAsync(titlesReader);
以异步方式将多个命令与 MARS 结合使用
该示例将打开与 AdventureWorks 数据库的单个连接。 使用 SqlCommand 对象时，将创建一个 SqlDataReader。 使用阅读器时，打开第二个 SqlDataReader，使用第一个 SqlDataReader 的数据作为第二个阅读器的 WHERE 子句的输入。
下面的示例使用 AdventureWorks 示例数据库。 示例代码中提供的连接字符串假定数据库已安装并且在本地计算机上可用。 根据环境需要修改连接字符串。
using System.Data.Common;
using Microsoft.Data.SqlClient;
using System.Threading.Tasks;
class Class1
static void Main()
Task task = MultipleCommands();
task.Wait();
static async Task MultipleCommands()
// By default, MARS is disabled when connecting to a MARS-enabled.
// It must be enabled in the connection string.
string connectionString = GetConnectionString();
int vendorID;
SqlDataReader productReader = null;
string vendorSQL =
"SELECT BusinessEntityID, Name FROM Purchasing.Vendor";
string productSQL =
"SELECT Production.Product.Name FROM Production.Product " +
"INNER JOIN Purchasing.ProductVendor " +
"ON Production.Product.ProductID = " +
"Purchasing.ProductVendor.ProductID " +
"WHERE Purchasing.ProductVendor.BusinessEntityID = @VendorId";
using (SqlConnection awConnection =
new SqlConnection(connectionString))
SqlCommand vendorCmd = new SqlCommand(vendorSQL, awConnection);
SqlCommand productCmd =
new SqlCommand(productSQL, awConnection);
productCmd.Parameters.Add("@VendorId", SqlDbType.Int);
await awConnection.OpenAsync();
using (SqlDataReader vendorReader = await vendorCmd.ExecuteReaderAsync())
while (await vendorReader.ReadAsync())
Console.WriteLine(vendorReader["Name"]);
vendorID = (int)vendorReader["BusinessEntityID"];
productCmd.Parameters["@VendorId"].Value = vendorID;
// The following line of code requires a MARS-enabled connection.
productReader = await productCmd.ExecuteReaderAsync();
using (productReader)
while (await productReader.ReadAsync())
Console.WriteLine(" " +
productReader["Name"].ToString());
private static string GetConnectionString()
// To avoid storing the connection string in your code, you can retrieve it from a configuration file.
return "Data Source=(local);Integrated Security=SSPI;Initial Catalog=AdventureWorks;MultipleActiveResultSets=True";
使用 MARS 以异步方式读取和更新数据
MARS 允许将连接用于读取操作和数据操作语言 (DML) 操作，其中有多个待处理操作。 此功能使应用程序无需处理连接繁忙错误。 此外，MARS 可取代对服务器端游标的使用，游标通常会消耗更多资源。 最后，因为可以在单个连接上执行多个操作，所以，这些操作可以共享相同的事务上下文，不需要使用 sp_getbindtoken 和 sp_bindsession 系统存储过程。
下面的控制台应用程序演示如何使用两个包含三个 SqlCommand 对象的 SqlDataReader 对象和一个启用了 MARS 的 SqlConnection 对象。 第一个命令对象检索信用评级为 5 的供应商列表。 第二个命令对象使用 SqlDataReader 提供的供应商 ID 为第二个 SqlDataReader 加载特定供应商的所有产品。 每个产品记录由第二个 SqlDataReader 访问。 通过执行计算来确定新的 OnOrderQty。 然后，通过第三个命令对象来使用新值更新 ProductVendor 表。 整个过程发生在单个事务中，该事务在结束时回滚。
下面的示例使用 AdventureWorks 示例数据库。 示例代码中提供的连接字符串假定数据库已安装并且在本地计算机上可用。 根据环境需要修改连接字符串。
using System.Data.Common;
using Microsoft.Data.SqlClient;
using System.Threading.Tasks;
class Program
static void Main()
Task task = ReadingAndUpdatingData();
task.Wait();
static async Task ReadingAndUpdatingData()
// By default, MARS is disabled when connecting to a MARS-enabled host.
// It must be enabled in the connection string.
string connectionString = GetConnectionString();
SqlTransaction updateTx = null;
SqlCommand vendorCmd = null;
SqlCommand prodVendCmd = null;
SqlCommand updateCmd = null;
SqlDataReader prodVendReader = null;
int vendorID = 0;
int productID = 0;
int minOrderQty = 0;
int maxOrderQty = 0;
int onOrderQty = 0;
int recordsUpdated = 0;
int totalRecordsUpdated = 0;
string vendorSQL =
"SELECT BusinessEntityID, Name FROM Purchasing.Vendor " +
"WHERE CreditRating = 5";
string prodVendSQL =
"SELECT ProductID, MaxOrderQty, MinOrderQty, OnOrderQty " +
"FROM Purchasing.ProductVendor " +
"WHERE BusinessEntityID = @VendorID";
string updateSQL =
"UPDATE Purchasing.ProductVendor " +
"SET OnOrderQty = @OrderQty " +
"WHERE ProductID = @ProductID AND BusinessEntityID = @VendorID";
using (SqlConnection awConnection =
new SqlConnection(connectionString))
await awConnection.OpenAsync();
updateTx = await Task.Run(() => awConnection.BeginTransaction());
vendorCmd = new SqlCommand(vendorSQL, awConnection);
vendorCmd.Transaction = updateTx;
prodVendCmd = new SqlCommand(prodVendSQL, awConnection);
prodVendCmd.Transaction = updateTx;
prodVendCmd.Parameters.Add("@VendorId", SqlDbType.Int);
updateCmd = new SqlCommand(updateSQL, awConnection);
updateCmd.Transaction = updateTx;
updateCmd.Parameters.Add("@OrderQty", SqlDbType.Int);
updateCmd.Parameters.Add("@ProductID", SqlDbType.Int);
updateCmd.Parameters.Add("@VendorID", SqlDbType.Int);
using (SqlDataReader vendorReader = await vendorCmd.ExecuteReaderAsync())
while (await vendorReader.ReadAsync())
Console.WriteLine(vendorReader["Name"]);
vendorID = (int)vendorReader["BusinessEntityID"];
prodVendCmd.Parameters["@VendorID"].Value = vendorID;
prodVendReader = await prodVendCmd.ExecuteReaderAsync();
using (prodVendReader)
while (await prodVendReader.ReadAsync())
productID = (int)prodVendReader["ProductID"];
if (prodVendReader["OnOrderQty"] == DBNull.Value)
minOrderQty = (int)prodVendReader["MinOrderQty"];
onOrderQty = minOrderQty;
maxOrderQty = (int)prodVendReader["MaxOrderQty"];
onOrderQty = (int)(maxOrderQty / 2);
updateCmd.Parameters["@OrderQty"].Value = onOrderQty;
updateCmd.Parameters["@ProductID"].Value = productID;
updateCmd.Parameters["@VendorID"].Value = vendorID;
recordsUpdated = await updateCmd.ExecuteNonQueryAsync();
totalRecordsUpdated += recordsUpdated;
Console.WriteLine("Total Records Updated: ", totalRecordsUpdated.ToString());
await Task.Run(() => updateTx.Rollback());
Console.WriteLine("Transaction Rolled Back");
private static string GetConnectionString()
// To avoid storing the connection string in your code, you can retrieve it from a configuration file.
return "Data Source=(local);Integrated Security=SSPI;Initial Catalog=AdventureWorks;MultipleActiveResultSets=True";
在 ADO.NET 中检索和修改数据
用于 SQL Server 的 Microsoft ADO.NET
.NET Framework
