Language
Note
Access to this page requires authorization. You can try signing in or changing directories.
Access to this page requires authorization. You can try changing directories.
The TPL Dataflow Library provides the System.Threading.Tasks.Dataflow.BatchBlock<T> and System.Threading.Tasks.Dataflow.BatchedJoinBlock<T1,T2> classes so that you can receive and buffer data from one or more sources and then propagate out that buffered data as one collection. This batching mechanism is useful when you collect data from one or more sources and then process multiple data elements as a batch. For example, consider an application that uses dataflow to insert records into a database. This operation can be more efficient if multiple items are inserted at the same time instead of one at a time sequentially. This document describes how to use the BatchBlock<T> class to improve the efficiency of such database insert operations. It also describes how to use the BatchedJoinBlock<T1,T2> class to capture both the results and any exceptions that occur when the program reads from a database.
Note
The TPL dataflow library (the System.Threading.Tasks.Dataflow namespace) is included in .NET 6 and later versions. For .NET Framework and .NET Standard projects, you need to install the 📦 System.Threading.Tasks.Dataflow NuGet package.
Prerequisites
Read the Join Blocks section in the Dataflow document before you start this walkthrough.
Install the .NET 10 SDK. Use a development database, not a production database.
Create a SQL database
Create or connect to a SQL database on one of the following platforms:
Set the SQL_CONNECTION_STRING environment variable to a connection string for your development database. For example, use Server=<server>,<port>;Database=<database>;User ID=<user_id>;Password=<password>;Encrypt=true;TrustServerCertificate=false. Set the actual secret in your local environment, not in source code.
For a local development container with a self-signed certificate, use TrustServerCertificate=true only for that development connection. For production connections, use a trusted server certificate and keep certificate validation enabled.
The example creates a uniquely named global temporary table in tempdb. An open connection keeps the table available to the worker connections. The program resets only its own table between demonstrations and drops it when the demonstration ends. You don't need Northwind or SQL Server Compact.
Creating the Console Application
In Visual Studio, create a Visual C# or Visual Basic Console Application project. In this document, the project is named
DataflowBatchDatabase.Target .NET 10 and install the Microsoft.Data.SqlClient NuGet package. For Microsoft Entra ID authentication with Microsoft.Data.SqlClient 7 and later versions, also install Microsoft.Data.SqlClient.Extensions.Azure. The Dataflow APIs are included in .NET 10.
Replace the generated code in Program.cs (Program.vb for Visual Basic) with the complete example, or follow the steps in this walkthrough to assemble it. Start with these
using(Importsin Visual Basic) statements.using System; using System.Collections.Generic; using Microsoft.Data.SqlClient; using System.Diagnostics; using System.Threading.Tasks.Dataflow;Imports System.Data Imports Microsoft.Data.SqlClient Imports System.Collections.Generic Imports System.Threading.Tasks.DataflowAdd the following data members to the
Programclass.// The number of employees to add to the database. // TODO: Change this value to experiment with different numbers of // employees to insert into the database. static readonly int s_insertCount = 256; // The size of a single batch of employees to add to the database. // TODO: Change this value to experiment with different batch sizes. static readonly int s_insertBatchSize = 96; static readonly string s_employeeTable = "##DataflowEmployees_" + Guid.NewGuid().ToString("N");' The number of employees to add to the database. ' TODO: Change this value to experiment with different numbers of ' employees to insert into the database. Private Shared ReadOnly s_insertCount As Integer = 256 ' The size of a single batch of employees to add to the database. ' TODO: Change this value to experiment with different batch sizes. Private Shared ReadOnly s_insertBatchSize As Integer = 96 Private Shared ReadOnly s_employeeTable As String = "##DataflowEmployees_" & Guid.NewGuid().ToString("N")
Defining the Employee Class
Add to the Program class the Employee class.
// Describes an employee. Each property maps to a
// column in the sample employee table.
class Employee
{
public int EmployeeID { get; set; }
public string LastName { get; set; }
public string FirstName { get; set; }
// A random number generator that helps tp generate
// Employee property values.
static readonly Random s_rand = new Random(42);
// Possible random first names.
static readonly string[] s_firstNames =
{ "Tom", "Mike", "Ruth", "Bob", "John" };
// Possible random last names.
static readonly string[] s_lastNames =
{ "Jones", "Smith", "Johnson", "Walker" };
// Creates an Employee object that contains random
// property values.
public static Employee Random()
{
return new Employee
{
EmployeeID = -1,
LastName = s_lastNames[s_rand.Next() % s_lastNames.Length],
FirstName = s_firstNames[s_rand.Next() % s_firstNames.Length]
};
}
}
' Describes an employee. Each property maps to a
' column in the sample employee table.
Private Class Employee
Public Property EmployeeID() As Integer
Public Property LastName() As String
Public Property FirstName() As String
' A random number generator that helps tp generate
' Employee property values.
Private Shared ReadOnly s_rand As New Random(42)
' Possible random first names.
Private Shared ReadOnly s_firstNames() As String =
{"Tom", "Mike", "Ruth", "Bob", "John"}
' Possible random last names.
Private Shared ReadOnly s_lastNames() As String =
{"Jones", "Smith", "Johnson", "Walker"}
' Creates an Employee object that contains random
' property values.
Public Shared Function Random() As Employee
Return New Employee With {
.EmployeeID = -1,
.LastName = s_lastNames(s_rand.Next() Mod s_lastNames.Length),
.FirstName = s_firstNames(s_rand.Next() Mod s_firstNames.Length)
}
End Function
End Class
The Employee class contains three properties, EmployeeID, LastName, and FirstName. These properties correspond to the Employee ID, Last Name, and First Name columns in the temporary employee table. For this demonstration, the Employee class also defines the Random method, which creates an Employee object that has random values for its properties.
Defining Employee Database Operations
Add to the Program class the InsertEmployees, GetEmployeeCount, and GetEmployeeID methods.
// Adds new employee records to the database.
static void InsertEmployees(Employee[] employees, string connectionString)
{
using (SqlConnection connection =
new SqlConnection(connectionString))
{
try
{
// Create the SQL command.
SqlCommand command = new SqlCommand(
$"INSERT INTO [{s_employeeTable}] ([Last Name], [First Name])" +
"VALUES (@lastName, @firstName)",
connection);
connection.Open();
for (int i = 0; i < employees.Length; i++)
{
// Set parameters.
command.Parameters.Clear();
command.Parameters.AddWithValue("@lastName", employees[i].LastName);
command.Parameters.AddWithValue("@firstName", employees[i].FirstName);
// Execute the command.
command.ExecuteNonQuery();
}
}
finally
{
connection.Close();
}
}
}
// Retrieves the number of entries in the sample employee table.
static int GetEmployeeCount(string connectionString)
{
int result = 0;
using (SqlConnection sqlConnection =
new SqlConnection(connectionString))
{
SqlCommand sqlCommand = new SqlCommand(
$"SELECT COUNT(*) FROM [{s_employeeTable}]", sqlConnection);
sqlConnection.Open();
try
{
result = (int)sqlCommand.ExecuteScalar();
}
finally
{
sqlConnection.Close();
}
}
return result;
}
// Retrieves the ID of the first employee that has the provided name.
static int GetEmployeeID(string lastName, string firstName,
string connectionString)
{
using (SqlConnection connection =
new SqlConnection(connectionString))
{
SqlCommand command = new SqlCommand(
$"SELECT TOP (1) [Employee ID] FROM [{s_employeeTable}] " +
"WHERE [Last Name] = @lastName AND [First Name] = @firstName ORDER BY [Employee ID]",
connection);
command.Parameters.Add("@lastName", System.Data.SqlDbType.NVarChar).Value = lastName;
command.Parameters.Add("@firstName", System.Data.SqlDbType.NVarChar).Value = firstName;
connection.Open();
try
{
object result = command.ExecuteScalar();
if (result == null)
throw new KeyNotFoundException("No matching employee.");
return (int)result;
}
finally
{
connection.Close();
}
}
}
' Adds new employee records to the database.
Private Shared Sub InsertEmployees(ByVal employees() As Employee, ByVal connectionString As String)
Using connection As New SqlConnection(connectionString)
Try
' Create the SQL command.
Dim command As New SqlCommand($"INSERT INTO [{s_employeeTable}] ([Last Name], [First Name]) VALUES (@lastName, @firstName)", connection)
connection.Open()
For i As Integer = 0 To employees.Length - 1
' Set parameters.
command.Parameters.Clear()
command.Parameters.Add("@lastName", SqlDbType.NVarChar).Value = employees(i).LastName
command.Parameters.Add("@firstName", SqlDbType.NVarChar).Value = employees(i).FirstName
' Execute the command.
command.ExecuteNonQuery()
Next i
Finally
connection.Close()
End Try
End Using
End Sub
' Retrieves the number of entries in the sample employee table.
Private Shared Function GetEmployeeCount(ByVal connectionString As String) As Integer
Dim result As Integer = 0
Using sqlConnection As New SqlConnection(connectionString)
Dim sqlCommand As New SqlCommand($"SELECT COUNT(*) FROM [{s_employeeTable}]", sqlConnection)
sqlConnection.Open()
Try
result = CInt(Fix(sqlCommand.ExecuteScalar()))
Finally
sqlConnection.Close()
End Try
End Using
Return result
End Function
' Retrieves the ID of the first employee that has the provided name.
Private Shared Function GetEmployeeID(ByVal lastName As String, ByVal firstName As String, ByVal connectionString As String) As Integer
Using connection As New SqlConnection(connectionString)
Dim query As String = $"SELECT TOP (1) [Employee ID] FROM [{s_employeeTable}] WHERE [Last Name] = @lastName AND [First Name] = @firstName ORDER BY [Employee ID]"
Dim command As New SqlCommand(query, connection)
command.Parameters.Add("@lastName", SqlDbType.NVarChar, 50).Value = lastName
command.Parameters.Add("@firstName", SqlDbType.NVarChar, 50).Value = firstName
connection.Open()
Try
Dim result = command.ExecuteScalar()
If result Is Nothing Then
Throw New KeyNotFoundException("No matching employee.")
End If
Return CInt(result)
Finally
connection.Close()
End Try
End Using
End Function
The InsertEmployees method adds new employee records to the temporary table. The GetEmployeeCount method retrieves the number of entries in that table. The GetEmployeeID method retrieves the identifier of the first employee that has the provided name, or throws KeyNotFoundException if no employee matches. Each method takes the connection string for your development database.
Adding Employee Data to the Database Without Using Buffering
Add to the Program class the AddEmployees and PostRandomEmployees methods.
// Posts random Employee data to the provided target block.
static void PostRandomEmployees(ITargetBlock<Employee> target, int count)
{
Console.WriteLine($"Adding {count} entries to Employee table...");
for (int i = 0; i < count; i++)
{
target.Post(Employee.Random());
}
}
// Adds random employee data to the database by using dataflow.
static void AddEmployees(string connectionString, int count)
{
// Create an ActionBlock<Employee> object that adds a single
// employee entry to the database.
var insertEmployee = new ActionBlock<Employee>(e =>
InsertEmployees(new Employee[] { e }, connectionString));
// Post several random Employee objects to the dataflow block.
PostRandomEmployees(insertEmployee, count);
// Set the dataflow block to the completed state and wait for
// all insert operations to complete.
insertEmployee.Complete();
insertEmployee.Completion.Wait();
}
' Posts random Employee data to the provided target block.
Private Shared Sub PostRandomEmployees(ByVal target As ITargetBlock(Of Employee), ByVal count As Integer)
Console.WriteLine("Adding {0} entries to Employee table...", count)
For i As Integer = 0 To count - 1
target.Post(Employee.Random())
Next i
End Sub
' Adds random employee data to the database by using dataflow.
Private Shared Sub AddEmployees(ByVal connectionString As String, ByVal count As Integer)
' Create an ActionBlock<Employee> object that adds a single
' employee entry to the database.
Dim insertEmployee = New ActionBlock(Of Employee)(Sub(e) InsertEmployees(New Employee() {e}, connectionString))
' Post several random Employee objects to the dataflow block.
PostRandomEmployees(insertEmployee, count)
' Set the dataflow block to the completed state and wait for
' all insert operations to complete.
insertEmployee.Complete()
insertEmployee.Completion.Wait()
End Sub
The AddEmployees method adds random employee data to the database by using dataflow. It creates an ActionBlock<TInput> object that calls the InsertEmployees method to add an employee entry to the database. The AddEmployees method then calls the PostRandomEmployees method to post multiple Employee objects to the ActionBlock<TInput> object. The AddEmployees method then waits for all insert operations to finish.
Using Buffering to Add Employee Data to the Database
Add to the Program class the AddEmployeesBatched method.
// Adds random employee data to the database by using dataflow.
// This method is similar to AddEmployees except that it uses batching
// to add multiple employees to the database at a time.
static void AddEmployeesBatched(string connectionString, int batchSize,
int count)
{
// Create a BatchBlock<Employee> that holds several Employee objects and
// then propagates them out as an array.
var batchEmployees = new BatchBlock<Employee>(batchSize);
// Create an ActionBlock<Employee[]> object that adds multiple
// employee entries to the database.
var insertEmployees = new ActionBlock<Employee[]>(a =>
InsertEmployees(a, connectionString));
// Link the batch block to the action block.
batchEmployees.LinkTo(insertEmployees);
// When the batch block completes, set the action block also to complete.
batchEmployees.Completion.ContinueWith(delegate { insertEmployees.Complete(); });
// Post several random Employee objects to the batch block.
PostRandomEmployees(batchEmployees, count);
// Set the batch block to the completed state and wait for
// all insert operations to complete.
batchEmployees.Complete();
insertEmployees.Completion.Wait();
}
' Adds random employee data to the database by using dataflow.
' This method is similar to AddEmployees except that it uses batching
' to add multiple employees to the database at a time.
Private Shared Sub AddEmployeesBatched(ByVal connectionString As String, ByVal batchSize As Integer, ByVal count As Integer)
' Create a BatchBlock<Employee> that holds several Employee objects and
' then propagates them out as an array.
Dim batchEmployees = New BatchBlock(Of Employee)(batchSize)
' Create an ActionBlock<Employee[]> object that adds multiple
' employee entries to the database.
Dim insertEmployees = New ActionBlock(Of Employee())(Sub(a) Program.InsertEmployees(a, connectionString))
' Link the batch block to the action block.
batchEmployees.LinkTo(insertEmployees)
' When the batch block completes, set the action block also to complete.
batchEmployees.Completion.ContinueWith(Sub() insertEmployees.Complete())
' Post several random Employee objects to the batch block.
PostRandomEmployees(batchEmployees, count)
' Set the batch block to the completed state and wait for
' all insert operations to complete.
batchEmployees.Complete()
insertEmployees.Completion.Wait()
End Sub
This method resembles AddEmployees, except that it also uses the BatchBlock<T> class to buffer multiple Employee objects before it sends those objects to the ActionBlock<TInput> object. Because the BatchBlock<T> class propagates out multiple elements as a collection, the ActionBlock<TInput> object is modified to act on an array of Employee objects. As in the AddEmployees method, AddEmployeesBatched calls the PostRandomEmployees method to post multiple Employee objects; however, AddEmployeesBatched posts these objects to the BatchBlock<T> object. The AddEmployeesBatched method also waits for all insert operations to finish.
Using Buffered Join to Read Employee Data from the Database
Add to the Program class the GetRandomEmployees method.
// Displays information about several random employees to the console.
static void GetRandomEmployees(string connectionString, int batchSize,
int count)
{
// Create a BatchedJoinBlock<Employee, Exception> object that holds
// both employee and exception data.
var selectEmployees = new BatchedJoinBlock<Employee, Exception>(batchSize);
// Holds the total number of exceptions that occurred.
int totalErrors = 0;
// Create an action block that prints employee and error information
// to the console.
var printEmployees =
new ActionBlock<Tuple<IList<Employee>, IList<Exception>>>(data =>
{
// Print information about the employees in this batch.
Console.WriteLine("Received a batch...");
foreach (Employee e in data.Item1)
{
Console.WriteLine($"Last={e.LastName} First={e.FirstName} ID={e.EmployeeID}");
}
// Print the error count for this batch.
Console.WriteLine($"There were {data.Item2.Count} errors in this batch...");
// Update total error count.
totalErrors += data.Item2.Count;
});
// Link the batched join block to the action block.
selectEmployees.LinkTo(printEmployees);
// When the batched join block completes, set the action block also to complete.
selectEmployees.Completion.ContinueWith(delegate { printEmployees.Complete(); });
// Try to retrieve the ID for several random employees.
Console.WriteLine("Selecting random entries from Employees table...");
for (int i = 0; i < count; i++)
{
try
{
// Create a random employee.
Employee e = Employee.Random();
// Try to retrieve the ID for the employee from the database.
e.EmployeeID = GetEmployeeID(e.LastName, e.FirstName, connectionString);
// Post the Employee object to the Employee target of
// the batched join block.
selectEmployees.Target1.Post(e);
}
catch (KeyNotFoundException e)
{
// GetEmployeeID throws KeyNotFoundException when there is
// no such employee with the given name. When this happens,
// post the Exception object to the Exception target of
// the batched join block.
selectEmployees.Target2.Post(e);
}
}
// Set the batched join block to the completed state and wait for
// all retrieval operations to complete.
selectEmployees.Complete();
printEmployees.Completion.Wait();
// Print the total error count.
Console.WriteLine($"Finished. There were {totalErrors} total errors.");
}
' Displays information about several random employees to the console.
Private Shared Sub GetRandomEmployees(ByVal connectionString As String, ByVal batchSize As Integer, ByVal count As Integer)
' Create a BatchedJoinBlock<Employee, Exception> object that holds
' both employee and exception data.
Dim selectEmployees = New BatchedJoinBlock(Of Employee, Exception)(batchSize)
' Holds the total number of exceptions that occurred.
Dim totalErrors As Integer = 0
' Create an action block that prints employee and error information
' to the console.
Dim printEmployees = New ActionBlock(Of Tuple(Of IList(Of Employee), IList(Of Exception)))(
Sub(data)
' Print information about the employees in this batch.
' Print the error count for this batch.
' Update total error count.
Console.WriteLine("Received a batch...")
For Each e As Employee In data.Item1
Console.WriteLine("Last={0} First={1} ID={2}", e.LastName, e.FirstName, e.EmployeeID)
Next e
Console.WriteLine("There were {0} errors in this batch...", data.Item2.Count)
totalErrors += data.Item2.Count
End Sub)
' Link the batched join block to the action block.
selectEmployees.LinkTo(printEmployees)
' When the batched join block completes, set the action block also to complete.
selectEmployees.Completion.ContinueWith(Sub() printEmployees.Complete())
' Try to retrieve the ID for several random employees.
Console.WriteLine("Selecting random entries from Employees table...")
For i As Integer = 0 To count - 1
Try
' Create a random employee.
Dim e As Employee = Employee.Random()
' Try to retrieve the ID for the employee from the database.
e.EmployeeID = GetEmployeeID(e.LastName, e.FirstName, connectionString)
' Post the Employee object to the Employee target of
' the batched join block.
selectEmployees.Target1.Post(e)
Catch e As KeyNotFoundException
' GetEmployeeID throws KeyNotFoundException when there is
' no such employee with the given name. When this happens,
' post the Exception object to the Exception target of
' the batched join block.
selectEmployees.Target2.Post(e)
End Try
Next i
' Set the batched join block to the completed state and wait for
' all retrieval operations to complete.
selectEmployees.Complete()
printEmployees.Completion.Wait()
' Print the total error count.
Console.WriteLine("Finished. There were {0} total errors.", totalErrors)
End Sub
This method prints information about random employees to the console. It creates several random Employee objects and calls the GetEmployeeID method to retrieve the unique identifier for each object. Because the GetEmployeeID method throws an exception if there is no matching employee with the given first and last names, the GetRandomEmployees method uses the BatchedJoinBlock<T1,T2> class to store Employee objects for successful calls to GetEmployeeID and System.Exception objects for calls that fail. The ActionBlock<TInput> object in this example acts on a Tuple<T1,T2> object that holds a list of Employee objects and a list of Exception objects. The BatchedJoinBlock<T1,T2> object propagates out this data when the sum of the received Employee and Exception object counts equals the batch size.
The Complete Example
The following example shows the complete code. The Main method compares the time that is required to perform batched database insertions versus the time to perform non-batched database insertions. It also demonstrates the use of buffered join to read employee data from the database and also report errors.
Both insertion demonstrations start with zero rows and finish with 256 rows. The final demonstration inserts 16 rows and performs 10 lookups. Each batch reuses one connection for its inserts; it still executes one command per employee. Elapsed times depend on your server and network and aren't a general performance guarantee.
using System;
using System.Collections.Generic;
using Microsoft.Data.SqlClient;
using System.Diagnostics;
using System.Threading.Tasks.Dataflow;
// Demonstrates how to use batched dataflow blocks to improve
// the performance of database operations.
namespace DataflowBatchDatabase
{
class Program
{
// The number of employees to add to the database.
// TODO: Change this value to experiment with different numbers of
// employees to insert into the database.
static readonly int s_insertCount = 256;
// The size of a single batch of employees to add to the database.
// TODO: Change this value to experiment with different batch sizes.
static readonly int s_insertBatchSize = 96;
static readonly string s_employeeTable =
"##DataflowEmployees_" + Guid.NewGuid().ToString("N");
// Describes an employee. Each property maps to a
// column in the sample employee table.
class Employee
{
public int EmployeeID { get; set; }
public string LastName { get; set; }
public string FirstName { get; set; }
// A random number generator that helps tp generate
// Employee property values.
static readonly Random s_rand = new Random(42);
// Possible random first names.
static readonly string[] s_firstNames =
{ "Tom", "Mike", "Ruth", "Bob", "John" };
// Possible random last names.
static readonly string[] s_lastNames =
{ "Jones", "Smith", "Johnson", "Walker" };
// Creates an Employee object that contains random
// property values.
public static Employee Random()
{
return new Employee
{
EmployeeID = -1,
LastName = s_lastNames[s_rand.Next() % s_lastNames.Length],
FirstName = s_firstNames[s_rand.Next() % s_firstNames.Length]
};
}
}
// Adds new employee records to the database.
static void InsertEmployees(Employee[] employees, string connectionString)
{
using (SqlConnection connection =
new SqlConnection(connectionString))
{
try
{
// Create the SQL command.
SqlCommand command = new SqlCommand(
$"INSERT INTO [{s_employeeTable}] ([Last Name], [First Name])" +
"VALUES (@lastName, @firstName)",
connection);
connection.Open();
for (int i = 0; i < employees.Length; i++)
{
// Set parameters.
command.Parameters.Clear();
command.Parameters.AddWithValue("@lastName", employees[i].LastName);
command.Parameters.AddWithValue("@firstName", employees[i].FirstName);
// Execute the command.
command.ExecuteNonQuery();
}
}
finally
{
connection.Close();
}
}
}
// Retrieves the number of entries in the sample employee table.
static int GetEmployeeCount(string connectionString)
{
int result = 0;
using (SqlConnection sqlConnection =
new SqlConnection(connectionString))
{
SqlCommand sqlCommand = new SqlCommand(
$"SELECT COUNT(*) FROM [{s_employeeTable}]", sqlConnection);
sqlConnection.Open();
try
{
result = (int)sqlCommand.ExecuteScalar();
}
finally
{
sqlConnection.Close();
}
}
return result;
}
// Retrieves the ID of the first employee that has the provided name.
static int GetEmployeeID(string lastName, string firstName,
string connectionString)
{
using (SqlConnection connection =
new SqlConnection(connectionString))
{
SqlCommand command = new SqlCommand(
$"SELECT TOP (1) [Employee ID] FROM [{s_employeeTable}] " +
"WHERE [Last Name] = @lastName AND [First Name] = @firstName ORDER BY [Employee ID]",
connection);
command.Parameters.Add("@lastName", System.Data.SqlDbType.NVarChar).Value = lastName;
command.Parameters.Add("@firstName", System.Data.SqlDbType.NVarChar).Value = firstName;
connection.Open();
try
{
object result = command.ExecuteScalar();
if (result == null)
throw new KeyNotFoundException("No matching employee.");
return (int)result;
}
finally
{
connection.Close();
}
}
}
// Posts random Employee data to the provided target block.
static void PostRandomEmployees(ITargetBlock<Employee> target, int count)
{
Console.WriteLine($"Adding {count} entries to Employee table...");
for (int i = 0; i < count; i++)
{
target.Post(Employee.Random());
}
}
// Adds random employee data to the database by using dataflow.
static void AddEmployees(string connectionString, int count)
{
// Create an ActionBlock<Employee> object that adds a single
// employee entry to the database.
var insertEmployee = new ActionBlock<Employee>(e =>
InsertEmployees(new Employee[] { e }, connectionString));
// Post several random Employee objects to the dataflow block.
PostRandomEmployees(insertEmployee, count);
// Set the dataflow block to the completed state and wait for
// all insert operations to complete.
insertEmployee.Complete();
insertEmployee.Completion.Wait();
}
// Adds random employee data to the database by using dataflow.
// This method is similar to AddEmployees except that it uses batching
// to add multiple employees to the database at a time.
static void AddEmployeesBatched(string connectionString, int batchSize,
int count)
{
// Create a BatchBlock<Employee> that holds several Employee objects and
// then propagates them out as an array.
var batchEmployees = new BatchBlock<Employee>(batchSize);
// Create an ActionBlock<Employee[]> object that adds multiple
// employee entries to the database.
var insertEmployees = new ActionBlock<Employee[]>(a =>
InsertEmployees(a, connectionString));
// Link the batch block to the action block.
batchEmployees.LinkTo(insertEmployees);
// When the batch block completes, set the action block also to complete.
batchEmployees.Completion.ContinueWith(delegate { insertEmployees.Complete(); });
// Post several random Employee objects to the batch block.
PostRandomEmployees(batchEmployees, count);
// Set the batch block to the completed state and wait for
// all insert operations to complete.
batchEmployees.Complete();
insertEmployees.Completion.Wait();
}
// Displays information about several random employees to the console.
static void GetRandomEmployees(string connectionString, int batchSize,
int count)
{
// Create a BatchedJoinBlock<Employee, Exception> object that holds
// both employee and exception data.
var selectEmployees = new BatchedJoinBlock<Employee, Exception>(batchSize);
// Holds the total number of exceptions that occurred.
int totalErrors = 0;
// Create an action block that prints employee and error information
// to the console.
var printEmployees =
new ActionBlock<Tuple<IList<Employee>, IList<Exception>>>(data =>
{
// Print information about the employees in this batch.
Console.WriteLine("Received a batch...");
foreach (Employee e in data.Item1)
{
Console.WriteLine($"Last={e.LastName} First={e.FirstName} ID={e.EmployeeID}");
}
// Print the error count for this batch.
Console.WriteLine($"There were {data.Item2.Count} errors in this batch...");
// Update total error count.
totalErrors += data.Item2.Count;
});
// Link the batched join block to the action block.
selectEmployees.LinkTo(printEmployees);
// When the batched join block completes, set the action block also to complete.
selectEmployees.Completion.ContinueWith(delegate { printEmployees.Complete(); });
// Try to retrieve the ID for several random employees.
Console.WriteLine("Selecting random entries from Employees table...");
for (int i = 0; i < count; i++)
{
try
{
// Create a random employee.
Employee e = Employee.Random();
// Try to retrieve the ID for the employee from the database.
e.EmployeeID = GetEmployeeID(e.LastName, e.FirstName, connectionString);
// Post the Employee object to the Employee target of
// the batched join block.
selectEmployees.Target1.Post(e);
}
catch (KeyNotFoundException e)
{
// GetEmployeeID throws KeyNotFoundException when there is
// no such employee with the given name. When this happens,
// post the Exception object to the Exception target of
// the batched join block.
selectEmployees.Target2.Post(e);
}
}
// Set the batched join block to the completed state and wait for
// all retrieval operations to complete.
selectEmployees.Complete();
printEmployees.Completion.Wait();
// Print the total error count.
Console.WriteLine($"Finished. There were {totalErrors} total errors.");
}
static void Main()
{
string connectionString = Environment.GetEnvironmentVariable("SQL_CONNECTION_STRING");
if (string.IsNullOrWhiteSpace(connectionString))
throw new InvalidOperationException("Set the SQL_CONNECTION_STRING environment variable.");
// Keep the global temporary table alive while worker connections use it.
using var keeper = new SqlConnection(connectionString);
keeper.Open();
using var setup = new SqlCommand(
$"CREATE TABLE [{s_employeeTable}] " +
"([Employee ID] int IDENTITY PRIMARY KEY, " +
"[Last Name] nvarchar(50) NOT NULL, [First Name] nvarchar(50) NOT NULL)", keeper);
setup.ExecuteNonQuery();
try
{
Stopwatch stopwatch = new Stopwatch();
Console.WriteLine("Demonstrating non-batched database insert operations...");
Console.WriteLine($"Original size of Employee table: {GetEmployeeCount(connectionString)}.");
stopwatch.Start();
AddEmployees(connectionString, s_insertCount);
stopwatch.Stop();
Console.WriteLine("New size of Employee table: {0}; elapsed insert time: {1} ms.",
GetEmployeeCount(connectionString), stopwatch.ElapsedMilliseconds);
Console.WriteLine();
using var reset = new SqlCommand($"TRUNCATE TABLE [{s_employeeTable}]", keeper);
reset.ExecuteNonQuery();
Console.WriteLine("Demonstrating batched database insert operations...");
Console.WriteLine($"Original size of Employee table: {GetEmployeeCount(connectionString)}.");
stopwatch.Restart();
AddEmployeesBatched(connectionString, s_insertBatchSize, s_insertCount);
stopwatch.Stop();
Console.WriteLine("New size of Employee table: {0}; elapsed insert time: {1} ms.",
GetEmployeeCount(connectionString), stopwatch.ElapsedMilliseconds);
Console.WriteLine();
reset.ExecuteNonQuery();
Console.WriteLine("Demonstrating batched join database select operations...");
AddEmployeesBatched(connectionString, s_insertBatchSize, 16);
GetRandomEmployees(connectionString, s_insertBatchSize, 10);
}
finally
{
using var cleanup = new SqlCommand($"DROP TABLE [{s_employeeTable}]", keeper);
cleanup.ExecuteNonQuery();
}
}
}
}
Imports System.Data
Imports Microsoft.Data.SqlClient
Imports System.Collections.Generic
Imports System.Threading.Tasks.Dataflow
' Demonstrates how to use batched dataflow blocks to improve
' the performance of database operations.
Namespace DataflowBatchDatabase
Friend Class Program
' The number of employees to add to the database.
' TODO: Change this value to experiment with different numbers of
' employees to insert into the database.
Private Shared ReadOnly s_insertCount As Integer = 256
' The size of a single batch of employees to add to the database.
' TODO: Change this value to experiment with different batch sizes.
Private Shared ReadOnly s_insertBatchSize As Integer = 96
Private Shared ReadOnly s_employeeTable As String =
"##DataflowEmployees_" & Guid.NewGuid().ToString("N")
' Describes an employee. Each property maps to a
' column in the sample employee table.
Private Class Employee
Public Property EmployeeID() As Integer
Public Property LastName() As String
Public Property FirstName() As String
' A random number generator that helps tp generate
' Employee property values.
Private Shared ReadOnly s_rand As New Random(42)
' Possible random first names.
Private Shared ReadOnly s_firstNames() As String =
{"Tom", "Mike", "Ruth", "Bob", "John"}
' Possible random last names.
Private Shared ReadOnly s_lastNames() As String =
{"Jones", "Smith", "Johnson", "Walker"}
' Creates an Employee object that contains random
' property values.
Public Shared Function Random() As Employee
Return New Employee With {
.EmployeeID = -1,
.LastName = s_lastNames(s_rand.Next() Mod s_lastNames.Length),
.FirstName = s_firstNames(s_rand.Next() Mod s_firstNames.Length)
}
End Function
End Class
' Adds new employee records to the database.
Private Shared Sub InsertEmployees(ByVal employees() As Employee, ByVal connectionString As String)
Using connection As New SqlConnection(connectionString)
Try
' Create the SQL command.
Dim command As New SqlCommand($"INSERT INTO [{s_employeeTable}] ([Last Name], [First Name]) VALUES (@lastName, @firstName)", connection)
connection.Open()
For i As Integer = 0 To employees.Length - 1
' Set parameters.
command.Parameters.Clear()
command.Parameters.Add("@lastName", SqlDbType.NVarChar).Value = employees(i).LastName
command.Parameters.Add("@firstName", SqlDbType.NVarChar).Value = employees(i).FirstName
' Execute the command.
command.ExecuteNonQuery()
Next i
Finally
connection.Close()
End Try
End Using
End Sub
' Retrieves the number of entries in the sample employee table.
Private Shared Function GetEmployeeCount(ByVal connectionString As String) As Integer
Dim result As Integer = 0
Using sqlConnection As New SqlConnection(connectionString)
Dim sqlCommand As New SqlCommand($"SELECT COUNT(*) FROM [{s_employeeTable}]", sqlConnection)
sqlConnection.Open()
Try
result = CInt(Fix(sqlCommand.ExecuteScalar()))
Finally
sqlConnection.Close()
End Try
End Using
Return result
End Function
' Retrieves the ID of the first employee that has the provided name.
Private Shared Function GetEmployeeID(ByVal lastName As String, ByVal firstName As String, ByVal connectionString As String) As Integer
Using connection As New SqlConnection(connectionString)
Dim query As String = $"SELECT TOP (1) [Employee ID] FROM [{s_employeeTable}] WHERE [Last Name] = @lastName AND [First Name] = @firstName ORDER BY [Employee ID]"
Dim command As New SqlCommand(query, connection)
command.Parameters.Add("@lastName", SqlDbType.NVarChar, 50).Value = lastName
command.Parameters.Add("@firstName", SqlDbType.NVarChar, 50).Value = firstName
connection.Open()
Try
Dim result = command.ExecuteScalar()
If result Is Nothing Then
Throw New KeyNotFoundException("No matching employee.")
End If
Return CInt(result)
Finally
connection.Close()
End Try
End Using
End Function
' Posts random Employee data to the provided target block.
Private Shared Sub PostRandomEmployees(ByVal target As ITargetBlock(Of Employee), ByVal count As Integer)
Console.WriteLine("Adding {0} entries to Employee table...", count)
For i As Integer = 0 To count - 1
target.Post(Employee.Random())
Next i
End Sub
' Adds random employee data to the database by using dataflow.
Private Shared Sub AddEmployees(ByVal connectionString As String, ByVal count As Integer)
' Create an ActionBlock<Employee> object that adds a single
' employee entry to the database.
Dim insertEmployee = New ActionBlock(Of Employee)(Sub(e) InsertEmployees(New Employee() {e}, connectionString))
' Post several random Employee objects to the dataflow block.
PostRandomEmployees(insertEmployee, count)
' Set the dataflow block to the completed state and wait for
' all insert operations to complete.
insertEmployee.Complete()
insertEmployee.Completion.Wait()
End Sub
' Adds random employee data to the database by using dataflow.
' This method is similar to AddEmployees except that it uses batching
' to add multiple employees to the database at a time.
Private Shared Sub AddEmployeesBatched(ByVal connectionString As String, ByVal batchSize As Integer, ByVal count As Integer)
' Create a BatchBlock<Employee> that holds several Employee objects and
' then propagates them out as an array.
Dim batchEmployees = New BatchBlock(Of Employee)(batchSize)
' Create an ActionBlock<Employee[]> object that adds multiple
' employee entries to the database.
Dim insertEmployees = New ActionBlock(Of Employee())(Sub(a) Program.InsertEmployees(a, connectionString))
' Link the batch block to the action block.
batchEmployees.LinkTo(insertEmployees)
' When the batch block completes, set the action block also to complete.
batchEmployees.Completion.ContinueWith(Sub() insertEmployees.Complete())
' Post several random Employee objects to the batch block.
PostRandomEmployees(batchEmployees, count)
' Set the batch block to the completed state and wait for
' all insert operations to complete.
batchEmployees.Complete()
insertEmployees.Completion.Wait()
End Sub
' Displays information about several random employees to the console.
Private Shared Sub GetRandomEmployees(ByVal connectionString As String, ByVal batchSize As Integer, ByVal count As Integer)
' Create a BatchedJoinBlock<Employee, Exception> object that holds
' both employee and exception data.
Dim selectEmployees = New BatchedJoinBlock(Of Employee, Exception)(batchSize)
' Holds the total number of exceptions that occurred.
Dim totalErrors As Integer = 0
' Create an action block that prints employee and error information
' to the console.
Dim printEmployees = New ActionBlock(Of Tuple(Of IList(Of Employee), IList(Of Exception)))(
Sub(data)
' Print information about the employees in this batch.
' Print the error count for this batch.
' Update total error count.
Console.WriteLine("Received a batch...")
For Each e As Employee In data.Item1
Console.WriteLine("Last={0} First={1} ID={2}", e.LastName, e.FirstName, e.EmployeeID)
Next e
Console.WriteLine("There were {0} errors in this batch...", data.Item2.Count)
totalErrors += data.Item2.Count
End Sub)
' Link the batched join block to the action block.
selectEmployees.LinkTo(printEmployees)
' When the batched join block completes, set the action block also to complete.
selectEmployees.Completion.ContinueWith(Sub() printEmployees.Complete())
' Try to retrieve the ID for several random employees.
Console.WriteLine("Selecting random entries from Employees table...")
For i As Integer = 0 To count - 1
Try
' Create a random employee.
Dim e As Employee = Employee.Random()
' Try to retrieve the ID for the employee from the database.
e.EmployeeID = GetEmployeeID(e.LastName, e.FirstName, connectionString)
' Post the Employee object to the Employee target of
' the batched join block.
selectEmployees.Target1.Post(e)
Catch e As KeyNotFoundException
' GetEmployeeID throws KeyNotFoundException when there is
' no such employee with the given name. When this happens,
' post the Exception object to the Exception target of
' the batched join block.
selectEmployees.Target2.Post(e)
End Try
Next i
' Set the batched join block to the completed state and wait for
' all retrieval operations to complete.
selectEmployees.Complete()
printEmployees.Completion.Wait()
' Print the total error count.
Console.WriteLine("Finished. There were {0} total errors.", totalErrors)
End Sub
Shared Sub Main()
Dim connectionString = Environment.GetEnvironmentVariable("SQL_CONNECTION_STRING")
If String.IsNullOrWhiteSpace(connectionString) Then
Throw New InvalidOperationException("Set the SQL_CONNECTION_STRING environment variable.")
End If
' Keep the global temporary table alive while worker connections use it.
Using keeper As New SqlConnection(connectionString)
keeper.Open()
Using setup As New SqlCommand(
$"CREATE TABLE [{s_employeeTable}] " &
"([Employee ID] int IDENTITY PRIMARY KEY, " &
"[Last Name] nvarchar(50) NOT NULL, [First Name] nvarchar(50) NOT NULL)", keeper)
setup.ExecuteNonQuery()
End Using
Try
Dim stopwatch As New Stopwatch()
Console.WriteLine("Demonstrating non-batched database insert operations...")
Console.WriteLine("Original size of Employee table: {0}.", GetEmployeeCount(connectionString))
stopwatch.Start()
AddEmployees(connectionString, s_insertCount)
stopwatch.Stop()
Console.WriteLine("New size of Employee table: {0}; elapsed insert time: {1} ms.", GetEmployeeCount(connectionString), stopwatch.ElapsedMilliseconds)
Console.WriteLine()
Using reset As New SqlCommand($"TRUNCATE TABLE [{s_employeeTable}]", keeper)
reset.ExecuteNonQuery()
End Using
Console.WriteLine("Demonstrating batched database insert operations...")
Console.WriteLine("Original size of Employee table: {0}.", GetEmployeeCount(connectionString))
stopwatch.Restart()
AddEmployeesBatched(connectionString, s_insertBatchSize, s_insertCount)
stopwatch.Stop()
Console.WriteLine("New size of Employee table: {0}; elapsed insert time: {1} ms.", GetEmployeeCount(connectionString), stopwatch.ElapsedMilliseconds)
Console.WriteLine()
Using reset As New SqlCommand($"TRUNCATE TABLE [{s_employeeTable}]", keeper)
reset.ExecuteNonQuery()
End Using
Console.WriteLine("Demonstrating batched join database select operations...")
AddEmployeesBatched(connectionString, s_insertBatchSize, 16)
GetRandomEmployees(connectionString, s_insertBatchSize, 10)
Finally
Using cleanup As New SqlCommand($"DROP TABLE [{s_employeeTable}]", keeper)
cleanup.ExecuteNonQuery()
End Using
End Try
End Using
End Sub
End Class
End Namespace
See also
Collaborate with us on GitHub
The source for this content can be found on GitHub, where you can also create and review issues and pull requests. For more information, see our contributor guide.
