Porady: Określanie harmonogramu zadań w bloku przepływu danych

W tym dokumencie pokazano, jak skojarzyć określony harmonogram zadań podczas korzystania z przepływu danych w aplikacji. W przykładzie użyto klasy System.Threading.Tasks.ConcurrentExclusiveSchedulerPair w aplikacji Windows Forms, aby pokazać, kiedy zadania czytelnika są aktywne, a gdy zadanie zapisujące jest aktywne. Używa również metody TaskScheduler.FromCurrentSynchronizationContext, aby umożliwić uruchamianie bloku przepływu danych na wątku interfejsu użytkownika.

Uwaga

Biblioteka przepływów danych TPL ( System.Threading.Tasks.Dataflow przestrzeń nazw) jest zawarta na platformie .NET 6 i nowszych wersjach. W przypadku projektów .NET Framework i .NET Standard należy zainstalować 📦 pakiet NuGet System.Threading.Tasks.Dataflow.

Aby utworzyć aplikację Windows Forms

  1. Utwórz projekt aplikacji Visual C# lub Visual Basic Windows Forms. W poniższych krokach projekt nosi nazwę WriterReadersWinForms.

  2. W projektancie formularzy dla formularza głównego Form1.cs (Form1.vb dla języka Visual Basic) dodaj cztery CheckBox kontrolki. Ustaw właściwość Text na Reader 1 dla checkBox1, Reader 2 dla checkBox2, Reader 3 dla checkBox3, i Pisarz dla checkBox4. Ustaw właściwość Enabled dla każdej kontrolki na False.

  3. Dodaj kontrolkę Timer do formularza. Ustaw właściwość Interval na 2500.

Dodawanie funkcji przepływu danych

W tej sekcji opisano sposób tworzenia bloków przepływu danych, które uczestniczą w aplikacji i jak skojarzyć każdy z nich z harmonogramem zadań.

Aby dodać funkcję przepływu danych do aplikacji

  1. W projekcie dodaj odwołanie do System.Threading.Tasks.Dataflow.dll.

  2. Upewnij się, że Form1.cs (Form1.vb dla języka Visual Basic) zawiera następujące using dyrektywy (Imports w Visual Basic).

    using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using System.Windows.Forms;

    Imports System.Threading
Imports System.Threading.Tasks
Imports System.Threading.Tasks.Dataflow

  3. Dodaj składową BroadcastBlock<T> danych do Form1 klasy.

    // Broadcasts values to an ActionBlock<int> object that is associated
// with each check box.
BroadcastBlock<int> broadcaster = new BroadcastBlock<int>(null);

    ' Broadcasts values to an ActionBlock<int> object that is associated
' with each check box.
Private broadcaster As New BroadcastBlock(Of Integer)(Nothing)

  4. W konstruktorze Form1 po wywołaniu metody InitializeComponent utwórz obiekt ActionBlock<TInput>, który przełącza stan obiektów CheckBox.

    // Create an ActionBlock<CheckBox> object that toggles the state
// of CheckBox objects.
// Specifying the current synchronization context enables the
// action to run on the user-interface thread.
var toggleCheckBox = new ActionBlock<CheckBox>(checkBox =>
{
   checkBox.Checked = !checkBox.Checked;
},
new ExecutionDataflowBlockOptions
{
   TaskScheduler = TaskScheduler.FromCurrentSynchronizationContext()
});

    ' Create an ActionBlock<CheckBox> object that toggles the state
' of CheckBox objects.
' Specifying the current synchronization context enables the 
' action to run on the user-interface thread.
Dim toggleCheckBox = New ActionBlock(Of CheckBox)(Sub(checkBox) checkBox.Checked = Not checkBox.Checked, New ExecutionDataflowBlockOptions With {.TaskScheduler = TaskScheduler.FromCurrentSynchronizationContext()})

  5. W konstruktorze Form1 utwórz ConcurrentExclusiveSchedulerPair obiekt i cztery ActionBlock<TInput> obiekty, jeden ActionBlock<TInput> obiekt dla każdego CheckBox obiektu. Dla każdego obiektu ActionBlock<TInput> określ obiekt ExecutionDataflowBlockOptions, który ma właściwość TaskScheduler ustawioną jako właściwość ConcurrentScheduler dla czytelników oraz właściwość ExclusiveScheduler dla piszącego.

    // Create a ConcurrentExclusiveSchedulerPair object.
// Readers will run on the concurrent part of the scheduler pair.
// The writer will run on the exclusive part of the scheduler pair.
var taskSchedulerPair = new ConcurrentExclusiveSchedulerPair();

// Create an ActionBlock<int> object for each reader CheckBox object.
// Each ActionBlock<int> object represents an action that can read
// from a resource in parallel to other readers.
// Specifying the concurrent part of the scheduler pair enables the
// reader to run in parallel to other actions that are managed by
// that scheduler.
var readerActions =
   from checkBox in new CheckBox[] {checkBox1, checkBox2, checkBox3}
   select new ActionBlock<int>(milliseconds =>
   {
      // Toggle the check box to the checked state.
      toggleCheckBox.Post(checkBox);

      // Perform the read action. For demonstration, suspend the current
      // thread to simulate a lengthy read operation.
      Thread.Sleep(milliseconds);

      // Toggle the check box to the unchecked state.
      toggleCheckBox.Post(checkBox);
   },
   new ExecutionDataflowBlockOptions
   {
      TaskScheduler = taskSchedulerPair.ConcurrentScheduler
   });

// Create an ActionBlock<int> object for the writer CheckBox object.
// This ActionBlock<int> object represents an action that writes to
// a resource, but cannot run in parallel to readers.
// Specifying the exclusive part of the scheduler pair enables the
// writer to run in exclusively with respect to other actions that are
// managed by the scheduler pair.
var writerAction = new ActionBlock<int>(milliseconds =>
{
   // Toggle the check box to the checked state.
   toggleCheckBox.Post(checkBox4);

   // Perform the write action. For demonstration, suspend the current
   // thread to simulate a lengthy write operation.
   Thread.Sleep(milliseconds);

   // Toggle the check box to the unchecked state.
   toggleCheckBox.Post(checkBox4);
},
new ExecutionDataflowBlockOptions
{
   TaskScheduler = taskSchedulerPair.ExclusiveScheduler
});

// Link the broadcaster to each reader and writer block.
// The BroadcastBlock<T> class propagates values that it
// receives to all connected targets.
foreach (var readerAction in readerActions)
{
   broadcaster.LinkTo(readerAction);
}
broadcaster.LinkTo(writerAction);

    ' Create a ConcurrentExclusiveSchedulerPair object.
' Readers will run on the concurrent part of the scheduler pair.
' The writer will run on the exclusive part of the scheduler pair.
Dim taskSchedulerPair = New ConcurrentExclusiveSchedulerPair()

' Create an ActionBlock<int> object for each reader CheckBox object.
' Each ActionBlock<int> object represents an action that can read 
' from a resource in parallel to other readers.
' Specifying the concurrent part of the scheduler pair enables the 
' reader to run in parallel to other actions that are managed by 
' that scheduler.
Dim readerActions = From checkBox In New CheckBox() {checkBox1, checkBox2, checkBox3} _
                    Select New ActionBlock(Of Integer)(Sub(milliseconds)
                                           ' Toggle the check box to the checked state.
                                           ' Perform the read action. For demonstration, suspend the current
                                           ' thread to simulate a lengthy read operation.
                                           ' Toggle the check box to the unchecked state.
                                           toggleCheckBox.Post(checkBox)
                                                           Thread.Sleep(milliseconds)
                                                           toggleCheckBox.Post(checkBox)
                                                       End Sub, New ExecutionDataflowBlockOptions With {.TaskScheduler = taskSchedulerPair.ConcurrentScheduler})

' Create an ActionBlock<int> object for the writer CheckBox object.
' This ActionBlock<int> object represents an action that writes to 
' a resource, but cannot run in parallel to readers.
' Specifying the exclusive part of the scheduler pair enables the 
' writer to run in exclusively with respect to other actions that are 
' managed by the scheduler pair.
Dim writerAction = New ActionBlock(Of Integer)(Sub(milliseconds)
                                                   ' Toggle the check box to the checked state.
                                                   ' Perform the write action. For demonstration, suspend the current
                                                   ' thread to simulate a lengthy write operation.
                                                   ' Toggle the check box to the unchecked state.
                                                   toggleCheckBox.Post(checkBox4)
                                                   Thread.Sleep(milliseconds)
                                                   toggleCheckBox.Post(checkBox4)
                                               End Sub, New ExecutionDataflowBlockOptions With {.TaskScheduler = taskSchedulerPair.ExclusiveScheduler})

' Link the broadcaster to each reader and writer block.
' The BroadcastBlock<T> class propagates values that it 
' receives to all connected targets.
For Each readerAction In readerActions
    broadcaster.LinkTo(readerAction)
Next readerAction
broadcaster.LinkTo(writerAction)

  6. W konstruktorze Form1 uruchom obiekt Timer.

    // Start the timer.
timer1.Start();

    ' Start the timer.
timer1.Start()

  7. W projektancie formularzy dla formularza głównego utwórz procedurę obsługi zdarzeń dla Tick zdarzenia dla czasomierza.

  8. Zaimplementuj Tick zdarzenie dla czasomierza.

    // Event handler for the timer.
private void timer1_Tick(object sender, EventArgs e)
{
   // Post a value to the broadcaster. The broadcaster
   // sends this message to each target.
   broadcaster.Post(1000);
}

    ' Event handler for the timer.
Private Sub timer1_Tick(ByVal sender As Object, ByVal e As EventArgs) Handles timer1.Tick
    ' Post a value to the broadcaster. The broadcaster
    ' sends this message to each target. 
    broadcaster.Post(1000)
End Sub

toggleCheckBox Ponieważ blok przepływu danych działa w interfejsie użytkownika, ważne jest, aby ta akcja wystąpiła w wątku interfejsu użytkownika. W tym celu, podczas konstruowania, ten obiekt udostępnia obiekt ExecutionDataflowBlockOptions, który ma właściwość TaskScheduler ustawioną na TaskScheduler.FromCurrentSynchronizationContext. Metoda FromCurrentSynchronizationContext tworzy obiekt TaskScheduler, który realizuje zadania w bieżącym kontekście synchronizacji. Form1 Ponieważ konstruktor jest wywoływany z wątku interfejsu użytkownika, akcja w toggleCheckBox bloku przepływu danych również działa w wątku interfejsu użytkownika.

W tym przykładzie ConcurrentExclusiveSchedulerPair użyto również klasy , aby umożliwić współbieżne działanie niektórych bloków przepływu danych, a inny blok przepływu danych działa wyłącznie w odniesieniu do wszystkich innych bloków przepływu danych uruchamianych w tym samym ConcurrentExclusiveSchedulerPair obiekcie. Ta technika jest przydatna, gdy wiele bloków przepływu danych współdzieli zasób, a niektóre z nich wymagają wyłącznego dostępu do tego zasobu, ponieważ eliminuje to konieczność ręcznego synchronizowania dostępu do tego zasobu. Eliminacja synchronizacji ręcznej może zwiększyć wydajność kodu.

Przykład

Poniższy przykład przedstawia kompletny kod dla Form1.cs (Form1.vb dla języka Visual Basic).

using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using System.Windows.Forms;

namespace WriterReadersWinForms
{
   public partial class Form1 : Form
   {
      // Broadcasts values to an ActionBlock<int> object that is associated
      // with each check box.
      BroadcastBlock<int> broadcaster = new BroadcastBlock<int>(null);

      public Form1()
      {
         InitializeComponent();

         // Create an ActionBlock<CheckBox> object that toggles the state
         // of CheckBox objects.
         // Specifying the current synchronization context enables the
         // action to run on the user-interface thread.
         var toggleCheckBox = new ActionBlock<CheckBox>(checkBox =>
         {
            checkBox.Checked = !checkBox.Checked;
         },
         new ExecutionDataflowBlockOptions
         {
            TaskScheduler = TaskScheduler.FromCurrentSynchronizationContext()
         });

         // Create a ConcurrentExclusiveSchedulerPair object.
         // Readers will run on the concurrent part of the scheduler pair.
         // The writer will run on the exclusive part of the scheduler pair.
         var taskSchedulerPair = new ConcurrentExclusiveSchedulerPair();

         // Create an ActionBlock<int> object for each reader CheckBox object.
         // Each ActionBlock<int> object represents an action that can read
         // from a resource in parallel to other readers.
         // Specifying the concurrent part of the scheduler pair enables the
         // reader to run in parallel to other actions that are managed by
         // that scheduler.
         var readerActions =
            from checkBox in new CheckBox[] {checkBox1, checkBox2, checkBox3}
            select new ActionBlock<int>(milliseconds =>
            {
               // Toggle the check box to the checked state.
               toggleCheckBox.Post(checkBox);

               // Perform the read action. For demonstration, suspend the current
               // thread to simulate a lengthy read operation.
               Thread.Sleep(milliseconds);

               // Toggle the check box to the unchecked state.
               toggleCheckBox.Post(checkBox);
            },
            new ExecutionDataflowBlockOptions
            {
               TaskScheduler = taskSchedulerPair.ConcurrentScheduler
            });

         // Create an ActionBlock<int> object for the writer CheckBox object.
         // This ActionBlock<int> object represents an action that writes to
         // a resource, but cannot run in parallel to readers.
         // Specifying the exclusive part of the scheduler pair enables the
         // writer to run in exclusively with respect to other actions that are
         // managed by the scheduler pair.
         var writerAction = new ActionBlock<int>(milliseconds =>
         {
            // Toggle the check box to the checked state.
            toggleCheckBox.Post(checkBox4);

            // Perform the write action. For demonstration, suspend the current
            // thread to simulate a lengthy write operation.
            Thread.Sleep(milliseconds);

            // Toggle the check box to the unchecked state.
            toggleCheckBox.Post(checkBox4);
         },
         new ExecutionDataflowBlockOptions
         {
            TaskScheduler = taskSchedulerPair.ExclusiveScheduler
         });

         // Link the broadcaster to each reader and writer block.
         // The BroadcastBlock<T> class propagates values that it
         // receives to all connected targets.
         foreach (var readerAction in readerActions)
         {
            broadcaster.LinkTo(readerAction);
         }
         broadcaster.LinkTo(writerAction);

         // Start the timer.
         timer1.Start();
      }

      // Event handler for the timer.
      private void timer1_Tick(object sender, EventArgs e)
      {
         // Post a value to the broadcaster. The broadcaster
         // sends this message to each target.
         broadcaster.Post(1000);
      }
   }
}

Imports System.Threading
Imports System.Threading.Tasks
Imports System.Threading.Tasks.Dataflow


Namespace WriterReadersWinForms
    Partial Public Class Form1
        Inherits Form
        ' Broadcasts values to an ActionBlock<int> object that is associated
        ' with each check box.
        Private broadcaster As New BroadcastBlock(Of Integer)(Nothing)

        Public Sub New()
            InitializeComponent()

            ' Create an ActionBlock<CheckBox> object that toggles the state
            ' of CheckBox objects.
            ' Specifying the current synchronization context enables the 
            ' action to run on the user-interface thread.
            Dim toggleCheckBox = New ActionBlock(Of CheckBox)(Sub(checkBox) checkBox.Checked = Not checkBox.Checked, New ExecutionDataflowBlockOptions With {.TaskScheduler = TaskScheduler.FromCurrentSynchronizationContext()})

            ' Create a ConcurrentExclusiveSchedulerPair object.
            ' Readers will run on the concurrent part of the scheduler pair.
            ' The writer will run on the exclusive part of the scheduler pair.
            Dim taskSchedulerPair = New ConcurrentExclusiveSchedulerPair()

            ' Create an ActionBlock<int> object for each reader CheckBox object.
            ' Each ActionBlock<int> object represents an action that can read 
            ' from a resource in parallel to other readers.
            ' Specifying the concurrent part of the scheduler pair enables the 
            ' reader to run in parallel to other actions that are managed by 
            ' that scheduler.
            Dim readerActions = From checkBox In New CheckBox() {checkBox1, checkBox2, checkBox3} _
                                Select New ActionBlock(Of Integer)(Sub(milliseconds)
                                                       ' Toggle the check box to the checked state.
                                                       ' Perform the read action. For demonstration, suspend the current
                                                       ' thread to simulate a lengthy read operation.
                                                       ' Toggle the check box to the unchecked state.
                                                       toggleCheckBox.Post(checkBox)
                                                                       Thread.Sleep(milliseconds)
                                                                       toggleCheckBox.Post(checkBox)
                                                                   End Sub, New ExecutionDataflowBlockOptions With {.TaskScheduler = taskSchedulerPair.ConcurrentScheduler})

            ' Create an ActionBlock<int> object for the writer CheckBox object.
            ' This ActionBlock<int> object represents an action that writes to 
            ' a resource, but cannot run in parallel to readers.
            ' Specifying the exclusive part of the scheduler pair enables the 
            ' writer to run in exclusively with respect to other actions that are 
            ' managed by the scheduler pair.
            Dim writerAction = New ActionBlock(Of Integer)(Sub(milliseconds)
                                                               ' Toggle the check box to the checked state.
                                                               ' Perform the write action. For demonstration, suspend the current
                                                               ' thread to simulate a lengthy write operation.
                                                               ' Toggle the check box to the unchecked state.
                                                               toggleCheckBox.Post(checkBox4)
                                                               Thread.Sleep(milliseconds)
                                                               toggleCheckBox.Post(checkBox4)
                                                           End Sub, New ExecutionDataflowBlockOptions With {.TaskScheduler = taskSchedulerPair.ExclusiveScheduler})

            ' Link the broadcaster to each reader and writer block.
            ' The BroadcastBlock<T> class propagates values that it 
            ' receives to all connected targets.
            For Each readerAction In readerActions
                broadcaster.LinkTo(readerAction)
            Next readerAction
            broadcaster.LinkTo(writerAction)

            ' Start the timer.
            timer1.Start()
        End Sub

        ' Event handler for the timer.
        Private Sub timer1_Tick(ByVal sender As Object, ByVal e As EventArgs) Handles timer1.Tick
            ' Post a value to the broadcaster. The broadcaster
            ' sends this message to each target. 
            broadcaster.Post(1000)
        End Sub
    End Class
End Namespace

Zobacz też

  • Last updated on