Exemplarische Vorgehensweise: Effizienzverbesserung durch Verwendung von BatchBlock und BatchedJoinBlock
Die TPL-Datenflussbibliothek stellt die System.Threading.Tasks.Dataflow.BatchBlock<T>- und die System.Threading.Tasks.Dataflow.BatchedJoinBlock<T1,T2>-Klasse bereit, sodass Sie Daten aus einer oder mehreren Quellen empfangen und puffern und diese Daten dann als Auflistung weitergeben können. Dieser Batchverarbeitungsmechanismus ist hilfreich, wenn Sie Daten aus einer oder mehreren Quellen sammeln und dann mehrere Datenelemente als Batch verarbeiten. Stellen Sie sich beispielsweise eine Anwendung vor, die Datensätze mithilfe von Datenfluss in eine Datenbank einfügt. Dieser Vorgang kann effizienter werden, wenn mehrere Elemente gleichzeitig statt hintereinander eingefügt werden. In diesem Dokument wird beschrieben, wie mit der BatchBlock<T>-Klasse die Effizienz solcher Datenbankeinfügevorgänge verbessert wird. Außerdem wird beschrieben, wie mit der BatchedJoinBlock<T1,T2>-Klasse sowohl die Ergebnisse als auch Ausnahmen erfasst werden, die auftreten, während das Programm aus einer Datenbank liest.
Hinweis
Die TPL-Datenflussbibliothek (System.Threading.Tasks.Dataflow-Namespace) wird nicht mit .NET ausgeliefert. Öffnen Sie zum Installieren des System.Threading.Tasks.Dataflow-Namespace in Visual Studio Ihr Projekt, wählen Sie im Menü Projekt die Option NuGet-Pakete verwalten aus, und suchen Sie online nach dem System.Threading.Tasks.Dataflow-Paket. Alternativ können Sie es mithilfe der .NET Core-CLI installieren und dazu dotnet add package System.Threading.Tasks.Dataflow ausführen.
Voraussetzungen
Lesen Sie den Abschnitt „Gruppierungsblöcke“ im Dokument Datenfluss, bevor Sie mit dieser exemplarischen Vorgehensweise beginnen.
Stellen Sie sicher, dass Sie über eine Kopie der Northwind-Datenbank (Northwind.sdf) auf Ihrem Computer verfügen. Diese Datei befindet sich normalerweise im Ordner %Programme%\Microsoft SQL Server Compact Edition\v3.5\Samples\.
Wichtig
In einigen Versionen von Windows können Sie keine Verbindung zu „Northwind.sdf“ herstellen, wenn Visual Studio nicht im Administratormodus ausgeführt wird. Starten Sie zum Herstellen einer Verbindung mit „Northwind.sdf“ Visual Studio oder eine Developer-Eingabeaufforderung für Visual Studio im Modus Als Administrator ausführen.
Diese exemplarische Vorgehensweise enthält folgende Abschnitte:
Hinzufügen von Mitarbeiterdaten zur Datenbank ohne Pufferung
Verwenden von Pufferung beim Hinzufügen von Mitarbeiterdaten zur Datenbank
Verwenden einer gepufferten Gruppierung zum Lesen von Mitarbeiterdaten aus der Datenbank
Erstellen der Konsolenanwendung
Erstellen Sie in Visual Studio ein Visual C#- oder Visual Basic-Projekt des Typs Konsolenanwendung. In diesem Dokument hat das Projekt den Namen
DataflowBatchDatabase.Fügen Sie im Projekt einen Verweis auf „System.Data.SqlServerCe.dll“ und einen Verweis auf „System.Threading.Tasks.Dataflow.dll“ hinzu.
Stellen Sie sicher, dass „Form1.cs“ („Form1.vb“ für Visual Basic) die folgenden
using-Anweisungen (Importsin Visual Basic) enthält.using System; using System.Collections.Generic; using System.Data.SqlServerCe; using System.Diagnostics; using System.IO; using System.Threading.Tasks.Dataflow;Imports System.Collections.Generic Imports System.Data.SqlServerCe Imports System.Diagnostics Imports System.IO Imports System.Threading.Tasks.DataflowFügen Sie der
Program-Klasse die folgenden Datenmember hinzu.// 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 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 insertBatchSize = 96; // The source database file. // TODO: Change this value if Northwind.sdf is at a different location // on your computer. static readonly string sourceDatabase = @"C:\Program Files\Microsoft SQL Server Compact Edition\v3.5\Samples\Northwind.sdf"; // TODO: Change this value if you require a different temporary location. static readonly string scratchDatabase = @"C:\Temp\Northwind.sdf";' 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 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 insertBatchSize As Integer = 96 ' The source database file. ' TODO: Change this value if Northwind.sdf is at a different location ' on your computer. Private Shared ReadOnly sourceDatabase As String = "C:\Program Files\Microsoft SQL Server Compact Edition\v3.5\Samples\Northwind.sdf" ' TODO: Change this value if you require a different temporary location. Private Shared ReadOnly scratchDatabase As String = "C:\Temp\Northwind.sdf"
Definieren der Klasse für Mitarbeiter
Fügen Sie der Program-Klasse die Employee-Klasse hinzu.
// Describes an employee. Each property maps to a
// column in the Employees table in the Northwind database.
// For brevity, the Employee class does not contain
// all columns from the Employees 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 Random rand = new Random(42);
// Possible random first names.
static readonly string[] firstNames = { "Tom", "Mike", "Ruth", "Bob", "John" };
// Possible random last names.
static readonly string[] lastNames = { "Jones", "Smith", "Johnson", "Walker" };
// Creates an Employee object that contains random
// property values.
public static Employee Random()
{
return new Employee
{
EmployeeID = -1,
LastName = lastNames[rand.Next() % lastNames.Length],
FirstName = firstNames[rand.Next() % firstNames.Length]
};
}
}
' Describes an employee. Each property maps to a
' column in the Employees table in the Northwind database.
' For brevity, the Employee class does not contain
' all columns from the Employees 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 rand As New Random(42)
' Possible random first names.
Private Shared ReadOnly firstNames() As String = {"Tom", "Mike", "Ruth", "Bob", "John"}
' Possible random last names.
Private Shared ReadOnly 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 = lastNames(rand.Next() Mod lastNames.Length), .FirstName = firstNames(rand.Next() Mod firstNames.Length)}
End Function
End Class
Die Employee-Klasse enthält die drei Eigenschaften EmployeeID, LastName und FirstName. Diese Eigenschaften entsprechen den Spalten Employee ID, Last Name und First Name in der Tabelle Employees der Northwind-Datenbank. Bei dieser Demonstration wird von der Employee-Klasse auch die Random-Methode definiert, die ein Employee-Objekt erstellt, das Zufallswerte für die zugehörigen Eigenschaften enthält.
Definieren der Datenbankvorgänge für Mitarbeiter
Fügen Sie der Program-Klasse die Methoden InsertEmployees, GetEmployeeCount und GetEmployeeID hinzu.
// Adds new employee records to the database.
static void InsertEmployees(Employee[] employees, string connectionString)
{
using (SqlCeConnection connection =
new SqlCeConnection(connectionString))
{
try
{
// Create the SQL command.
SqlCeCommand command = new SqlCeCommand(
"INSERT INTO Employees ([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.Add("@lastName", employees[i].LastName);
command.Parameters.Add("@firstName", employees[i].FirstName);
// Execute the command.
command.ExecuteNonQuery();
}
}
finally
{
connection.Close();
}
}
}
// Retrieves the number of entries in the Employees table in
// the Northwind database.
static int GetEmployeeCount(string connectionString)
{
int result = 0;
using (SqlCeConnection sqlConnection =
new SqlCeConnection(connectionString))
{
SqlCeCommand sqlCommand = new SqlCeCommand(
"SELECT COUNT(*) FROM Employees", 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 (SqlCeConnection connection =
new SqlCeConnection(connectionString))
{
SqlCeCommand command = new SqlCeCommand(
string.Format(
"SELECT [Employee ID] FROM Employees " +
"WHERE [Last Name] = '{0}' AND [First Name] = '{1}'",
lastName, firstName),
connection);
connection.Open();
try
{
return (int)command.ExecuteScalar();
}
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 SqlCeConnection(connectionString)
Try
' Create the SQL command.
Dim command As New SqlCeCommand("INSERT INTO Employees ([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", employees(i).LastName)
command.Parameters.Add("@firstName", 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 Employees table in
' the Northwind database.
Private Shared Function GetEmployeeCount(ByVal connectionString As String) As Integer
Dim result As Integer = 0
Using sqlConnection As New SqlCeConnection(connectionString)
Dim sqlCommand As New SqlCeCommand("SELECT COUNT(*) FROM Employees", 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 SqlCeConnection(connectionString)
Dim command As New SqlCeCommand(String.Format("SELECT [Employee ID] FROM Employees " & "WHERE [Last Name] = '{0}' AND [First Name] = '{1}'", lastName, firstName), connection)
connection.Open()
Try
Return CInt(Fix(command.ExecuteScalar()))
Finally
connection.Close()
End Try
End Using
End Function
Die InsertEmployees-Methode fügt der Datenbank neue Mitarbeiterdatensätze hinzu. Die GetEmployeeCount-Methode ruft die Anzahl von Einträgen in der Tabelle Employees ab. Die GetEmployeeID-Methode ruft den Bezeichner des ersten Mitarbeiters mit dem angegebenen Namen ab. Jede dieser Methoden nimmt eine Verbindungszeichenfolge zur Northwind-Datenbank an und verwendet Funktionen im System.Data.SqlServerCe-Namespace, um mit der Datenbank zu kommunizieren.
Hinzufügen von Mitarbeiterdaten zur Datenbank ohne Pufferung
Fügen Sie der Program-Klasse die Methoden AddEmployees und PostRandomEmployees hinzu.
// Posts random Employee data to the provided target block.
static void PostRandomEmployees(ITargetBlock<Employee> target, int count)
{
Console.WriteLine("Adding {0} entries to Employee table...", count);
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
Die AddEmployees-Methode fügt der Datenbank mithilfe von Datenfluss zufällige Mitarbeiterdaten hinzu. Sie erstellt ein ActionBlock<TInput>-Objekt, das die InsertEmployees-Methode aufruft, um der Datenbank einen Mitarbeitereintrag hinzuzufügen. Die AddEmployees-Methode ruft dann die PostRandomEmployees-Methode auf, um mehrere Employee-Objekte an das ActionBlock<TInput>-Objekt zu senden. Die AddEmployees-Methode wartet dann, bis alle Einfügevorgänge abgeschlossen sind.
Verwenden von Pufferung beim Hinzufügen von Mitarbeiterdaten zur Datenbank
Fügen Sie der Program-Klasse die AddEmployeesBatched-Methode hinzu.
// 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
Diese Methode ähnelt AddEmployees, außer dass sie auch die BatchBlock<T>-Klasse verwendet, um mehrere Employee-Objekte zu puffern, bevor diese Objekte an das ActionBlock<TInput>-Objekt gesendet werden. Da die BatchBlock<T>-Klasse mehrere Elemente als Auflistung weitergibt, wird das ActionBlock<TInput>-Objekt so geändert, dass es auf ein Array von Employee-Objekten angewendet wird. Wie bei der Methode AddEmployees ruft AddEmployeesBatched die PostRandomEmployees-Methode auf, um mehrere Employee-Objekte zu senden. AddEmployeesBatched sendet diese Objekte jedoch an das BatchBlock<T>-Objekt. Die AddEmployeesBatched-Methode wartet ebenfalls, bis alle Einfügevorgänge abgeschlossen sind.
Verwenden einer gepufferten Gruppierung zum Lesen von Mitarbeiterdaten aus der Datenbank
Fügen Sie der Program-Klasse die GetRandomEmployees-Methode hinzu.
// 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={0} First={1} ID={2}",
e.LastName, e.FirstName, e.EmployeeID);
}
// Print the error count for this batch.
Console.WriteLine("There were {0} errors in this batch...",
data.Item2.Count);
// 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 (NullReferenceException e)
{
// GetEmployeeID throws NullReferenceException 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 {0} total errors.", totalErrors);
}
' 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 NullReferenceException
' GetEmployeeID throws NullReferenceException 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
Diese Methode gibt Informationen über zufällige Mitarbeiter in der Konsole aus. Sie erstellt mehrere zufällige Employee-Objekte ruft die GetEmployeeID-Methode auf, um den eindeutigen Bezeichner für die einzelnen Objekte abzurufen. Da die GetEmployeeID-Methode eine Ausnahme auslöst, wenn kein entsprechender Mitarbeiter mit dem angegebenen Vor- und Nachnamen vorhanden ist, verwendet die GetRandomEmployees-Methode die BatchedJoinBlock<T1,T2>-Klasse, um Employee-Objekte für erfolgreiche Aufrufe von GetEmployeeID und System.Exception-Objekte für fehlgeschlagene Aufrufe zu speichern. Das ActionBlock<TInput>-Objekt in diesem Beispiel wird auf ein Tuple<T1,T2>-Objekt angewendet, das eine Liste von Employee-Objekten und eine Liste von Exception-Objekten enthält. Das BatchedJoinBlock<T1,T2>-Objekt gibt diese Daten weiter, wenn die Summe empfangenen Employee- und Exception-Objekte die Batchgröße erreicht.
Vollständiges Beispiel
Das folgende Beispiel enthält den vollständigen Code. Die Main-Methode vergleicht die Zeitspanne, die erforderlich ist, um Datenbankeinfügungen im Batch durchzuführen, mit der Zeitspanne zum Durchführen von Datenbankeinfügungen ohne Verwendung von Batches. Außerdem wird damit die Verwendung einer gepufferten Gruppierung demonstriert, um Mitarbeiterdaten aus der Datenbank zu lesen und Fehler zu melden.
using System;
using System.Collections.Generic;
using System.Data.SqlServerCe;
using System.Diagnostics;
using System.IO;
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 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 insertBatchSize = 96;
// The source database file.
// TODO: Change this value if Northwind.sdf is at a different location
// on your computer.
static readonly string sourceDatabase =
@"C:\Program Files\Microsoft SQL Server Compact Edition\v3.5\Samples\Northwind.sdf";
// TODO: Change this value if you require a different temporary location.
static readonly string scratchDatabase =
@"C:\Temp\Northwind.sdf";
// Describes an employee. Each property maps to a
// column in the Employees table in the Northwind database.
// For brevity, the Employee class does not contain
// all columns from the Employees 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 Random rand = new Random(42);
// Possible random first names.
static readonly string[] firstNames = { "Tom", "Mike", "Ruth", "Bob", "John" };
// Possible random last names.
static readonly string[] lastNames = { "Jones", "Smith", "Johnson", "Walker" };
// Creates an Employee object that contains random
// property values.
public static Employee Random()
{
return new Employee
{
EmployeeID = -1,
LastName = lastNames[rand.Next() % lastNames.Length],
FirstName = firstNames[rand.Next() % firstNames.Length]
};
}
}
// Adds new employee records to the database.
static void InsertEmployees(Employee[] employees, string connectionString)
{
using (SqlCeConnection connection =
new SqlCeConnection(connectionString))
{
try
{
// Create the SQL command.
SqlCeCommand command = new SqlCeCommand(
"INSERT INTO Employees ([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.Add("@lastName", employees[i].LastName);
command.Parameters.Add("@firstName", employees[i].FirstName);
// Execute the command.
command.ExecuteNonQuery();
}
}
finally
{
connection.Close();
}
}
}
// Retrieves the number of entries in the Employees table in
// the Northwind database.
static int GetEmployeeCount(string connectionString)
{
int result = 0;
using (SqlCeConnection sqlConnection =
new SqlCeConnection(connectionString))
{
SqlCeCommand sqlCommand = new SqlCeCommand(
"SELECT COUNT(*) FROM Employees", 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 (SqlCeConnection connection =
new SqlCeConnection(connectionString))
{
SqlCeCommand command = new SqlCeCommand(
string.Format(
"SELECT [Employee ID] FROM Employees " +
"WHERE [Last Name] = '{0}' AND [First Name] = '{1}'",
lastName, firstName),
connection);
connection.Open();
try
{
return (int)command.ExecuteScalar();
}
finally
{
connection.Close();
}
}
}
// Posts random Employee data to the provided target block.
static void PostRandomEmployees(ITargetBlock<Employee> target, int count)
{
Console.WriteLine("Adding {0} entries to Employee table...", count);
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={0} First={1} ID={2}",
e.LastName, e.FirstName, e.EmployeeID);
}
// Print the error count for this batch.
Console.WriteLine("There were {0} errors in this batch...",
data.Item2.Count);
// 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 (NullReferenceException e)
{
// GetEmployeeID throws NullReferenceException 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 {0} total errors.", totalErrors);
}
static void Main(string[] args)
{
// Create a connection string for accessing the database.
// The connection string refers to the temporary database location.
string connectionString = string.Format(@"Data Source={0}",
scratchDatabase);
// Create a Stopwatch object to time database insert operations.
Stopwatch stopwatch = new Stopwatch();
// Start with a clean database file by copying the source database to
// the temporary location.
File.Copy(sourceDatabase, scratchDatabase, true);
// Demonstrate multiple insert operations without batching.
Console.WriteLine("Demonstrating non-batched database insert operations...");
Console.WriteLine("Original size of Employee table: {0}.",
GetEmployeeCount(connectionString));
stopwatch.Start();
AddEmployees(connectionString, insertCount);
stopwatch.Stop();
Console.WriteLine("New size of Employee table: {0}; elapsed insert time: {1} ms.",
GetEmployeeCount(connectionString), stopwatch.ElapsedMilliseconds);
Console.WriteLine();
// Start again with a clean database file.
File.Copy(sourceDatabase, scratchDatabase, true);
// Demonstrate multiple insert operations, this time with batching.
Console.WriteLine("Demonstrating batched database insert operations...");
Console.WriteLine("Original size of Employee table: {0}.",
GetEmployeeCount(connectionString));
stopwatch.Restart();
AddEmployeesBatched(connectionString, insertBatchSize, insertCount);
stopwatch.Stop();
Console.WriteLine("New size of Employee table: {0}; elapsed insert time: {1} ms.",
GetEmployeeCount(connectionString), stopwatch.ElapsedMilliseconds);
Console.WriteLine();
// Start again with a clean database file.
File.Copy(sourceDatabase, scratchDatabase, true);
// Demonstrate multiple retrieval operations with error reporting.
Console.WriteLine("Demonstrating batched join database select operations...");
// Add a small number of employees to the database.
AddEmployeesBatched(connectionString, insertBatchSize, 16);
// Query for random employees.
GetRandomEmployees(connectionString, insertBatchSize, 10);
}
}
}
/* Sample output:
Demonstrating non-batched database insert operations...
Original size of Employee table: 15.
Adding 256 entries to Employee table...
New size of Employee table: 271; elapsed insert time: 11035 ms.
Demonstrating batched database insert operations...
Original size of Employee table: 15.
Adding 256 entries to Employee table...
New size of Employee table: 271; elapsed insert time: 197 ms.
Demonstrating batched join database insert operations...
Adding 16 entries to Employee table...
Selecting items from Employee table...
Received a batch...
Last=Jones First=Tom ID=21
Last=Jones First=John ID=24
Last=Smith First=Tom ID=26
Last=Jones First=Tom ID=21
There were 4 errors in this batch...
Received a batch...
Last=Smith First=Tom ID=26
Last=Jones First=Mike ID=28
There were 0 errors in this batch...
Finished. There were 4 total errors.
*/
Imports System.Collections.Generic
Imports System.Data.SqlServerCe
Imports System.Diagnostics
Imports System.IO
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 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 insertBatchSize As Integer = 96
' The source database file.
' TODO: Change this value if Northwind.sdf is at a different location
' on your computer.
Private Shared ReadOnly sourceDatabase As String = "C:\Program Files\Microsoft SQL Server Compact Edition\v3.5\Samples\Northwind.sdf"
' TODO: Change this value if you require a different temporary location.
Private Shared ReadOnly scratchDatabase As String = "C:\Temp\Northwind.sdf"
' Describes an employee. Each property maps to a
' column in the Employees table in the Northwind database.
' For brevity, the Employee class does not contain
' all columns from the Employees 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 rand As New Random(42)
' Possible random first names.
Private Shared ReadOnly firstNames() As String = {"Tom", "Mike", "Ruth", "Bob", "John"}
' Possible random last names.
Private Shared ReadOnly 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 = lastNames(rand.Next() Mod lastNames.Length), .FirstName = firstNames(rand.Next() Mod 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 SqlCeConnection(connectionString)
Try
' Create the SQL command.
Dim command As New SqlCeCommand("INSERT INTO Employees ([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", employees(i).LastName)
command.Parameters.Add("@firstName", 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 Employees table in
' the Northwind database.
Private Shared Function GetEmployeeCount(ByVal connectionString As String) As Integer
Dim result As Integer = 0
Using sqlConnection As New SqlCeConnection(connectionString)
Dim sqlCommand As New SqlCeCommand("SELECT COUNT(*) FROM Employees", 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 SqlCeConnection(connectionString)
Dim command As New SqlCeCommand(String.Format("SELECT [Employee ID] FROM Employees " & "WHERE [Last Name] = '{0}' AND [First Name] = '{1}'", lastName, firstName), connection)
connection.Open()
Try
Return CInt(Fix(command.ExecuteScalar()))
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 NullReferenceException
' GetEmployeeID throws NullReferenceException 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(ByVal args() As String)
' Create a connection string for accessing the database.
' The connection string refers to the temporary database location.
Dim connectionString As String = String.Format("Data Source={0}", scratchDatabase)
' Create a Stopwatch object to time database insert operations.
Dim stopwatch As New Stopwatch()
' Start with a clean database file by copying the source database to
' the temporary location.
File.Copy(sourceDatabase, scratchDatabase, True)
' Demonstrate multiple insert operations without batching.
Console.WriteLine("Demonstrating non-batched database insert operations...")
Console.WriteLine("Original size of Employee table: {0}.", GetEmployeeCount(connectionString))
stopwatch.Start()
AddEmployees(connectionString, insertCount)
stopwatch.Stop()
Console.WriteLine("New size of Employee table: {0}; elapsed insert time: {1} ms.", GetEmployeeCount(connectionString), stopwatch.ElapsedMilliseconds)
Console.WriteLine()
' Start again with a clean database file.
File.Copy(sourceDatabase, scratchDatabase, True)
' Demonstrate multiple insert operations, this time with batching.
Console.WriteLine("Demonstrating batched database insert operations...")
Console.WriteLine("Original size of Employee table: {0}.", GetEmployeeCount(connectionString))
stopwatch.Restart()
AddEmployeesBatched(connectionString, insertBatchSize, insertCount)
stopwatch.Stop()
Console.WriteLine("New size of Employee table: {0}; elapsed insert time: {1} ms.", GetEmployeeCount(connectionString), stopwatch.ElapsedMilliseconds)
Console.WriteLine()
' Start again with a clean database file.
File.Copy(sourceDatabase, scratchDatabase, True)
' Demonstrate multiple retrieval operations with error reporting.
Console.WriteLine("Demonstrating batched join database select operations...")
' Add a small number of employees to the database.
AddEmployeesBatched(connectionString, insertBatchSize, 16)
' Query for random employees.
GetRandomEmployees(connectionString, insertBatchSize, 10)
End Sub
End Class
End Namespace
' Sample output:
'Demonstrating non-batched database insert operations...
'Original size of Employee table: 15.
'Adding 256 entries to Employee table...
'New size of Employee table: 271; elapsed insert time: 11035 ms.
'
'Demonstrating batched database insert operations...
'Original size of Employee table: 15.
'Adding 256 entries to Employee table...
'New size of Employee table: 271; elapsed insert time: 197 ms.
'
'Demonstrating batched join database insert operations...
'Adding 16 entries to Employee table...
'Selecting items from Employee table...
'Received a batch...
'Last=Jones First=Tom ID=21
'Last=Jones First=John ID=24
'Last=Smith First=Tom ID=26
'Last=Jones First=Tom ID=21
'There were 4 errors in this batch...
'Received a batch...
'Last=Smith First=Tom ID=26
'Last=Jones First=Mike ID=28
'There were 0 errors in this batch...
'Finished. There were 4 total errors.
'
Siehe auch
Zusammenarbeit auf GitHub
Die Quelle für diesen Inhalt finden Sie auf GitHub, wo Sie auch Issues und Pull Requests erstellen und überprüfen können. Weitere Informationen finden Sie in unserem Leitfaden für Mitwirkende.
