Langue

Procédure pas à pas : utilisation de BatchBlock et BatchedJoinBlock pour améliorer l’efficacité

La bibliothèque de flux de données TPL fournit les classes System.Threading.Tasks.Dataflow.BatchBlock<T> et System.Threading.Tasks.Dataflow.BatchedJoinBlock<T1,T2> pour pouvoir recevoir et mettre en mémoire tampon des données d'une ou plusieurs sources, puis transmettre ces données mises en mémoire tampon sous forme d'une seule collection. Ce mécanisme de traitement par lots est utile lorsque vous collectez des données à partir d’une ou plusieurs sources, puis traitez plusieurs éléments de données en tant que lot. Par exemple, considérez une application qui utilise le flux de données pour insérer des enregistrements dans une base de données. Cette opération peut être plus efficace si plusieurs éléments sont insérés en même temps au lieu d’un à la fois séquentiellement. Ce document explique comment utiliser la BatchBlock<T> classe pour améliorer l’efficacité de ces opérations d’insertion de base de données. Il décrit également comment utiliser la BatchedJoinBlock<T1,T2> classe pour capturer les résultats et toutes les exceptions qui se produisent lorsque le programme lit à partir d’une base de données.

Note

La bibliothèque de flux de données TPL (l’espace de noms System.Threading.Tasks.Dataflow) est incluse dans .NET 6 et versions ultérieures. Pour les projets .NET Framework et .NET Standard, vous devez installer le 📦 package NuGet System.Threading.Tasks.Dataflow.

Prerequisites

  1. Lisez la section Blocs de jointure dans le document Dataflow avant de commencer cette procédure pas à pas.

  2. Installez le Kit de développement logiciel (SDK) .NET 10. Utilisez une base de données de développement, et non une base de données de production.

Créer une base de données SQL

Créer ou connecter une base de données SQL sur l’une des plateformes suivantes :

Définissez la variable d’environnement SQL_CONNECTION_STRING sur une chaîne de connexion pour votre base de données de développement. Par exemple, utilisez Server=<server>,<port>;Database=<database>;User ID=<user_id>;Password=<password>;Encrypt=true;TrustServerCertificate=false. Définissez le secret réel dans votre environnement local, et non dans le code source.

Pour un conteneur de développement local avec un certificat auto-signé, utilisez TrustServerCertificate=true uniquement pour cette connexion de développement. Pour les connexions de production, utilisez un certificat de serveur approuvé et activez la validation des certificats.

L’exemple crée une table temporaire globale nommée de manière unique dans tempdb. Une connexion ouverte conserve la table disponible pour les connexions worker. Le programme réinitialise uniquement sa propre table de base de données entre les démonstrations et la supprime lorsque la démonstration se termine. Vous n'avez pas besoin de Northwind ou SQL Server Compact.

Créer l'application console

  1. Dans Visual Studio, créez un projet d’application console Visual C# ou Visual Basic. Dans ce document, le projet est nommé DataflowBatchDatabase.

  2. Ciblez .NET 10 et installez le package NuGet Microsoft.Data.SqlClient. Pour l’authentification Microsoft Entra ID avec Microsoft.Data.SqlClient 7 et versions ultérieures, installez également Microsoft.Data.SqlClient.Extensions.Azure. Les API Dataflow sont incluses dans .NET 10.

  3. Remplacez le code généré dans Program.cs (Program.vb pour Visual Basic) par l’exemple complet ou suivez les étapes décrites dans cette procédure pas à pas pour l’assembler. Commencez par ces using instructions (Importsdans Visual Basic).

    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.Dataflow

  4. Ajoutez les membres de données suivants à la Program classe.

    // 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")

Définition de la classe Employé

Ajoutez à la Program classe la Employee classe.

// 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

La Employee classe contient trois propriétés, EmployeeID, LastNameet FirstName. Ces propriétés correspondent aux colonnes First Name, Employee ID et Last Name dans la table des employés temporaires. Pour cette démonstration, la Employee classe définit également la Random méthode, qui crée un Employee objet qui a des valeurs aléatoires pour ses propriétés.

Définition des opérations de base de données des employés

Ajoutez à la classe Program les méthodes InsertEmployees, GetEmployeeCount et GetEmployeeID.

// 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

La InsertEmployees méthode ajoute de nouveaux enregistrements d’employés à la table temporaire. La GetEmployeeCount méthode récupère le nombre d’entrées dans cette table. La GetEmployeeID méthode récupère l’identificateur du premier employé qui a le nom fourni ou lève KeyNotFoundException si aucun employé ne correspond. Chaque méthode prend le chaîne de connexion de votre base de données de développement.

Ajout de données d’employé à la base de données sans utiliser la mise en mémoire tampon

Ajoutez à la classe Program les méthodes AddEmployees et PostRandomEmployees.

// 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

La AddEmployees méthode ajoute des données d’employé aléatoires à la base de données à l’aide du flux de données. Il crée un ActionBlock<TInput> objet qui appelle la InsertEmployees méthode pour ajouter une entrée d’employé à la base de données. La AddEmployees méthode appelle ensuite la PostRandomEmployees méthode pour publier plusieurs Employee objets à l’objet ActionBlock<TInput> . La AddEmployees méthode attend ensuite que toutes les opérations d’insertion se terminent.

Utilisation de la mise en mémoire tampon pour ajouter des données d’employé à la base de données

Ajoutez à la classe Program la méthode 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

Cette méthode ressemble AddEmployees, sauf qu’elle utilise également la BatchBlock<T> classe pour mettre en mémoire tampon plusieurs Employee objets avant d’envoyer ces objets à l’objet ActionBlock<TInput> . Étant donné que la BatchBlock<T> classe propage plusieurs éléments en tant que collection, l’objet ActionBlock<TInput> est modifié pour agir sur un tableau d’objets Employee . Comme dans la AddEmployees méthode, AddEmployeesBatched appelle la PostRandomEmployees méthode pour publier plusieurs Employee objets ; toutefois, AddEmployeesBatched publie ces objets sur l’objet BatchBlock<T> . La AddEmployeesBatched méthode attend également que toutes les opérations d’insertion se terminent.

Utilisation de la jointure en mémoire tampon pour extraire les données des employés de la base de données

Ajoutez à la classe Program la méthode 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={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

Cette méthode imprime des informations sur les employés aléatoires dans la console. Il crée plusieurs objets aléatoires Employee et appelle la GetEmployeeID méthode pour récupérer l’identificateur unique de chaque objet. Étant donné que la GetEmployeeID méthode lève une exception s’il n’existe aucun employé correspondant avec le prénom et le nom donnés, la GetRandomEmployees méthode utilise la BatchedJoinBlock<T1,T2> classe pour stocker des objets pour les Employee appels réussis vers GetEmployeeID et System.Exception les objets pour les appels qui échouent. L’objet ActionBlock<TInput> de cet exemple agit sur un Tuple<T1,T2> objet qui contient une liste d’objets Employee et une liste d’objets Exception . L'objet BatchedJoinBlock<T1,T2> propage ces données lorsque la somme des nombres d'objets reçus Employee et Exception est égale à la taille du lot.

Exemple complet

L’exemple suivant montre le code complet. La Main méthode compare le temps nécessaire pour effectuer des insertions de base de données par lots et le temps nécessaire pour effectuer des insertions de base de données non lotées. Il illustre également l’utilisation de la jointure mise en mémoire tampon pour lire les données des employés à partir de la base de données et signaler également des erreurs.

Les deux démonstrations d’insertion commencent par zéro ligne et se terminent par 256 lignes. La démonstration finale insère 16 lignes et effectue 10 recherches. Chaque lot réutilise une connexion pour ses insertions ; il exécute toujours une commande par employé. Les temps écoulés dépendent de votre serveur et de votre réseau et ne sont pas une garantie de performances générale.

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

Voir aussi

  • Dernière mise à jour :