非同期プログラミング
このトピックでは、.NET Framework 4.5 で導入された非同期プログラミング機能をサポートするための拡張機能など、.NET Framework Data Provider for SQL Server (SqlClient) における非同期プログラミングのサポートについて説明します。
従来の非同期プログラミング
.NET Framework 4.5 より前のバージョンでは、SqlClient による非同期プログラミングは、以下のメソッドと Asynchronous Processing=true 接続プロパティを使用して行われていました。
この機能は .NET Framework 4.5 の SqlClient に残っています。
ヒント
.NET Framework 4.5 以降では、これらのレガシ メソッドの接続文字列に Asynchronous Processing=true は必要なくなりました。
.NET Framework 4.5 で追加された非同期プログラミング機能
新しい非同期プログラミング機能を使用すると、コードを簡単に非同期にすることができます。
.NET Framework 4.5 で導入された非同期プログラミング機能について詳しくは、以下をご覧ください。
ユーザー インターフェイスが応答しない場合やサーバーのパフォーマンスが向上しない場合は、コードをさらに非同期にする必要がある可能性があります。 非同期コードの記述には、従来、非同期操作の完了後に発生するロジックを表すコールバック (または継続とも呼ばれます) のインストールが伴っていました。 これにより、同期コードと比較して、非同期コードの構造は複雑になります。
現在は、コールバックを使用したり、コードを複数のメソッドやラムダ式で分割したりせずに、非同期メソッドを呼び出すことができるようになりました。
async 修飾子はメソッドが非同期であることを示します。 async メソッドを呼び出すと、タスクが返されます。 await 演算子がタスクに適用されると、現在のメソッドはすぐに終了します。 タスクが終了すると、同じメソッド内で実行が再開されます。
警告
非同期呼び出しは、アプリケーションで Context Connection 接続文字列キーワードも使用されている場合はサポートされません。
async メソッドを呼び出すと、追加のスレッドは割り当てられません。 既存の I/O 完了スレッドが最後に簡単に使用される可能性があります。
非同期プログラミングをサポートするために .NET Framework 4.5 に追加されたメソッドは次のとおりです。
他の非同期メンバーは、SqlClient ストリーミングのサポートのために追加されました。
ヒント
新しい非同期メソッドでは、接続文字列に Asynchronous Processing=true は必要ありません。
同期から非同期接続を開く
既存のアプリケーションをアップグレードして、新しい非同期機能を使用できます。 たとえば、同期接続アルゴリズムを使用したアプリケーションで、データベースに接続するたびに UI スレッドをブロックし、接続すると、ユーザーがサインインしたことを他のユーザーに通知するストアド プロシージャが呼び出されるとします。
using SqlConnection conn = new SqlConnection("…");
{
conn.Open();
using (SqlCommand cmd = new SqlCommand("StoredProcedure_Logon", conn))
{
cmd.ExecuteNonQuery();
}
}
新しい非同期機能を使用するように変換した場合、プログラムは次のようになります。
using System;
using System.Data.SqlClient;
using System.Threading.Tasks;
class A {
static async Task<int> Method(SqlConnection conn, SqlCommand cmd) {
await conn.OpenAsync();
await cmd.ExecuteNonQueryAsync();
return 1;
}
public static void Main() {
using (SqlConnection conn = new SqlConnection("Data Source=(local); Initial Catalog=NorthWind; Integrated Security=SSPI")) {
SqlCommand command = new SqlCommand("select top 2 * from orders", conn);
int result = A.Method(conn, command).Result;
SqlDataReader reader = command.ExecuteReader();
while (reader.Read())
Console.WriteLine(reader[0]);
}
}
}
既存アプリケーションに新しい非同期機能を追加する (古いパターンと新しいパターンが混在)
既存の非同期ロジックを変更せずに、新しい非同期機能 (SqlConnection::OpenAsync) を追加することもできます。 たとえば、現在、次のアルゴリズムを使用するアプリケーションがあるとします。
AsyncCallback productList = new AsyncCallback(ProductList);
SqlConnection conn = new SqlConnection("…");
conn.Open();
SqlCommand cmd = new SqlCommand("SELECT * FROM [Current Product List]", conn);
IAsyncResult ia = cmd.BeginExecuteReader(productList, cmd);
既存のアルゴリズムを大幅に変更せずに、新しい非同期パターンの使用を開始できます。
using System;
using System.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 Distributed Transaction Controller (MSDTC) サービスを有効にする必要があります。 MSDTC を有効にする方法の詳細については、「Web サーバーで MSDTC を有効にする方法」を参照してください。
using System;
using System.Data.Common;
using System.Data.SqlClient;
using System.Threading.Tasks;
class A {
static async Task PerformDBOperationsUsingProviderModel(string connectionString, string providerName) {
DbProviderFactory factory = DbProviderFactories.GetFactory(providerName);
using (DbConnection connection = factory.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 = "your_server";
builder.InitialCatalog = "pubs";
builder.IntegratedSecurity = true;
string provider = "System.Data.SqlClient";
Task task = PerformDBOperationsUsingProviderModel(builder.ConnectionString, provider);
task.Wait();
}
}
SQL トランザクションと新しい非同期機能を使用する
using System;
using System.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.
try {
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);
}
}
}
}
}
SQL トランザクションと新しい非同期機能を使用する
エンタープライズ アプリケーションでは、複数のデータベース サーバー間のトランザクションを有効にするために、一部のシナリオで分散トランザクションの追加が必要になる場合があります。 次のように、System.Transactions 名前空間を使用して分散トランザクションを登録できます。
using System;
using System.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
builder.DataSource = "your_server";
builder.InitialCatalog = "your_data_source";
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);
try {
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);
try {
transaction.Rollback();
}
catch (Exception ex2) {
Console.WriteLine("Rollback Exception Type: {0}", ex2.GetType());
Console.WriteLine(" Message: {0}", ex2.Message);
}
}
}
}
}
}
非同期操作を取り消す
CancellationToken を使用して、非同期要求を取り消すことができます。
using System;
using System.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
try {
Task result = CancellingAsynchronousOperations(source.Token);
result.Wait();
}
catch (AggregateException exception) {
if (exception.InnerException is SqlException) {
Console.WriteLine("Operation canceled");
}
else {
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 を使用した非同期操作
非同期機能は、System.Data.SqlClient.SqlBulkCopy を使用して、SqlBulkCopy.WriteToServerAsync にも追加されました。
using System;
using System.Collections.Generic;
using System.Data;
using System.Data.Odbc;
using System.Data.SqlClient;
using System.Linq;
using System.Text;
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 odbcConnectionString = @"Driver={SQL Server};Server=(localdb)\V11.0;UID=oledb;Pwd=[PLACEHOLDER];Database=Demo";
static string odbcConnectionString = @"Driver={SQL Server};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 Framework 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 Framework 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 (OdbcConnection odbcconn = new OdbcConnection(odbcConnectionString)) {
odbcconn.Open();
using (OdbcCommand odbccmd = new OdbcCommand(selectStatement, odbcconn)) {
using (OdbcDataReader odbcreader = odbccmd.ExecuteReader()) {
using (SqlConnection conn = new SqlConnection(connectionString)) {
await conn.OpenAsync();
string temptable = "temptable";//"[#" + Guid.NewGuid().ToString("N") + "]";
SqlCommand createCmd = new SqlCommand(string.Format(createDestTableStatement, temptable), conn);
await createCmd.ExecuteNonQueryAsync();
using (SqlBulkCopy bcp = new SqlBulkCopy(conn)) {
bcp.DestinationTableName = temptable;
await bcp.WriteToServerAsync(odbcreader);
}
}
}
}
}
}
// 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 Framework 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 データベースへの 1 つの接続を開きます。 SqlCommand オブジェクトを使用することで、SqlDataReader が作成されます。 そのリーダーが使用されるとき、最初の SqlDataReader のデータが 2 番目のリーダーの WHERE 句への入力として使用され、2 番目の SqlDataReader が開きます。
Note
次の例では、SQL Server に含まれるサンプルの AdventureWorks データベースを使用します。 サンプル コードに示されている接続文字列では、データベースがローカル コンピューターにインストールされ、使用可能であることを前提としています。 必要に応じて、お使いの環境に合わせて接続文字列を変更してください。
using System;
using System.Data;
using System.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 VendorId, 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.VendorID = @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["VendorId"];
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 (データ操作言語) 操作の両方で 1 つの接続を使用することができます。 この機能により、アプリケーションが接続ビジー エラーを処理する必要がなくなります。 さらに、MARS によって、通常多くのリソースを消費するサーバー側カーソルのユーザーを置き換えることができます。 最後に、複数の操作を単一の接続で実行できるので、同じトランザクション コンテキストを共有することにより、システムのストアド プロシージャである sp_getbindtoken と sp_bindsession を使用する必要がなくなります。
次のコンソール アプリケーションでは、2 つの SqlDataReader オブジェクトを 3 つの SqlCommand オブジェクトと使用する方法、および 1 つの SqlConnection オブジェクトを MARS を有効にして使用する方法について示します。 最初のコマンド オブジェクトは、信用格付けの評価が 5 であるベンダーの一覧を取得します。 2 番目のコマンド オブジェクトは、SqlDataReader から提供されているベンダー ID を使用して、特定のベンダーの全製品を含む 2 番目の SqlDataReader を読み込みます。 各製品レコードには、2 番目の SqlDataReader によってアクセスされます。 計算が実行され、新規 OnOrderQty を判定します。 3 番目のコマンド オブジェクトでは、ProductVendor テーブルを新しい値で更新します。 このプロセスはすべて 1 つのトランザクション内で実行され、最後にロールバックされます。
Note
次の例では、SQL Server に含まれるサンプルの AdventureWorks データベースを使用します。 サンプル コードに示されている接続文字列では、データベースがローカル コンピューターにインストールされ、使用可能であることを前提としています。 必要に応じて、お使いの環境に合わせて接続文字列を変更してください。
using System;
using System.Collections.Generic;
using System.Text;
using System.Data;
using System.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 VendorID, Name FROM Purchasing.Vendor " +
"WHERE CreditRating = 5";
string prodVendSQL =
"SELECT ProductID, MaxOrderQty, MinOrderQty, OnOrderQty " +
"FROM Purchasing.ProductVendor " +
"WHERE VendorID = @VendorID";
string updateSQL =
"UPDATE Purchasing.ProductVendor " +
"SET OnOrderQty = @OrderQty " +
"WHERE ProductID = @ProductID AND VendorID = @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["VendorID"];
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;
}
else {
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";
}
}