Passo a passo: usando BatchBlock e BatchedJoinBlock para melhorar a eficiência
A Biblioteca de Fluxo de dados TPL fornece as classes System.Threading.Tasks.Dataflow.BatchBlock<T> e System.Threading.Tasks.Dataflow.BatchedJoinBlock<T1,T2> para que você possa receber e armazenar em buffer os dados de uma ou mais fontes e, depois, propagar esses dados armazenados em buffer como uma coleção. Este mecanismo de envio em lote é útil quando você coleta dados de uma ou mais fontes e, em seguida, processa vários elementos de dados como um lote. Por exemplo, considere um aplicativo que usa o fluxo de dados para inserir registros em um banco de dados. Essa operação pode ser mais eficiente se vários itens forem inseridos ao mesmo tempo, em vez de um de cada vez sequencialmente. Este documento descreve como usar a classe BatchBlock<T> para melhorar a eficiência dessas operações de inserção de banco de dados. Também descreve como usar a classe BatchedJoinBlock<T1,T2> para capturar os resultados e todas as exceções que ocorrem quando o programa lê de um banco de dados.
Observação
A Biblioteca de Fluxo de Dados TPL (o namespace System.Threading.Tasks.Dataflow) não é distribuída com o .NET. Para instalar o namespace System.Threading.Tasks.Dataflow no Visual Studio, abra o projeto, escolha Gerenciar Pacotes NuGet no menu Projeto e pesquise online o pacote System.Threading.Tasks.Dataflow. Como alternativa, instale-o usando a CLI do .NET Core e execute dotnet add package System.Threading.Tasks.Dataflow.
Pré-requisitos
Leia a seção Blocos de ingresso no documento Fluxo de dados antes de começar este passo a passo.
Certifique-se de que você tenha uma cópia do banco de dados Northwind, Northwind.sdf, disponível em seu computador. Normalmente, esse arquivo está localizado na pasta %Arquivos de Programas%\Microsoft SQL Server Compact Edition\v3.5\Samples\.
Importante
Em algumas versões do Windows, não é possível se conectar ao Northwind.sdf quando o Visual Studio está em execução em um modo não administrador. Para conectar-se ao Northwind.sdf, inicie o Visual Studio ou um Prompt de Comando do Desenvolvedor para Visual Studio no modo Executar como administrador.
Este passo a passo contém as seguintes seções:
Adicionar dados de funcionário ao banco de dados sem o uso de buffer
Usar o armazenamento em buffer para adicionar dados de funcionário ao banco de dados
Usar o ingresso em buffer para ler dados de funcionário do banco de dados
Criando o Aplicativo de Console
No Visual Studio, crie um projeto de aplicativo de console do Visual Basic ou do Visual C#. Neste documento, o projeto é chamado
DataflowBatchDatabase.Em seu projeto, adicione uma referência ao System.Data.SqlServerCe.dll e uma referência a System.Threading.Tasks.Dataflow.dll.
Verifique se o Form1.cs (Form1.vb para Visual Basic) contém as seguintes instruções
using(Importsem Visual Basic).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.DataflowAdicione os membros de dados a seguir à classe
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";' 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"
Definir a classe do funcionário
Adicione a classe Program à classe Employee.
// 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
A classe Employee contém três propriedades: EmployeeID, LastName e FirstName. Essas propriedades correspondem às colunas Employee ID, Last Name e First Name na tabela Employees no banco de dados Northwind. Para esta demonstração, a classe Employee também define o método Random, que cria um objeto Employee com valores aleatórios para suas propriedades.
Definir as operações de banco de dados do funcionário
Adicione à classe Program os métodos InsertEmployees, GetEmployeeCount e GetEmployeeID.
// 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
O método InsertEmployees adiciona novos registros de funcionário ao banco de dados. O método GetEmployeeCount recupera o número de entradas na tabela Employees. O método GetEmployeeID recupera o identificador do primeiro funcionário que tem o nome fornecido. Cada um desses métodos usa uma cadeia de caracteres de conexão para o banco de dados Northwind e usa a funcionalidade no namespace System.Data.SqlServerCe para se comunicar com o banco de dados.
Adicionar dados de funcionário ao banco de dados sem o uso de buffer
Adicione à classe Program os métodos AddEmployees e PostRandomEmployees.
// 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
O método AddEmployees adiciona dados aleatórios de funcionários ao banco de dados usando o fluxo de dados. Ele cria um objeto ActionBlock<TInput> que chama o método InsertEmployees para adicionar uma entrada de funcionário ao banco de dados. Em seguida, o método AddEmployees chama o método PostRandomEmployees para postar vários objetos Employee para o objeto ActionBlock<TInput>. Depois, o método AddEmployees espera a conclusão de todas as operações de inserção.
Usar o armazenamento em buffer para adicionar dados de funcionário ao banco de dados
Adicione à classe Program o método AddEmployeesBatched.
// 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
Esse método é semelhante a AddEmployees, exceto que ele também usa a classe BatchBlock<T> para armazenar em buffer vários objetos Employee antes de enviar esses objetos ao objeto ActionBlock<TInput>. Como a classe BatchBlock<T> propaga vários elementos como uma coleção, o objeto ActionBlock<TInput> é modificado a fim de funcionar em uma matriz de objetos Employee. Assim como no método AddEmployees, AddEmployeesBatched chama o método PostRandomEmployees para postar vários objetos Employee; no entanto, AddEmployeesBatchedposta esses objetos no objeto BatchBlock<T>. O método AddEmployeesBatched também espera a conclusão de todas as operações de inserção.
Usar o ingresso em buffer para ler dados de funcionário do banco de dados
Adicione à classe Program o método GetRandomEmployees.
// 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
Esse método imprime as informações sobre funcionários aleatórios no console. Ele cria vários objetos Employee aleatórios e chama o método GetEmployeeID para recuperar o identificador exclusivo de cada objeto. Como o método GetEmployeeID lança uma exceção se nenhum funcionário corresponder aos nomes e sobrenomes indicados, o método GetRandomEmployees usa a classe BatchedJoinBlock<T1,T2> para armazenar objetos Employee para chamada bem-sucedida a GetEmployeeID, e objetos System.Exception para chamadas com falha. O objeto ActionBlock<TInput> neste exemplo age em um objeto Tuple<T1,T2> que contém uma lista de objetos Employee e uma lista de objetos Exception. O objeto BatchedJoinBlock<T1,T2> propaga esse dados quando a soma dos objetos Employee e Exception recebidos for igual ao tamanho do lote.
O Exemplo Completo
O exemplo a seguir mostra todo o código. O método Main compara o tempo necessário para executar inserções em lotes do banco de dados em comparação com o tempo para executar as inserções do banco de dados sem lote. Ele também demonstra o uso do ingresso em buffer para ler dados de funcionário do banco de dados e também informar erros.
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.
'
Confira também
Colaborar conosco no GitHub
A fonte deste conteúdo pode ser encontrada no GitHub, onde você também pode criar e revisar problemas e solicitações de pull. Para obter mais informações, confira o nosso guia para colaboradores.
