연습: 사용자 지정 데이터 흐름 블록 형식 만들기
TPL 데이터 흐름 라이브러리는 다양한 기능을 구현하는 여러 데이터 흐름 블록 형식을 제공하지만 사용자 지정 블록 형식을 만들 수도 있습니다. 이 문서에서는 사용자 지정 동작을 구현하는 데이터 흐름 블록 형식을 만드는 방법을 설명합니다.
필수 구성 요소
이 문서를 읽기 전에 데이터 흐름을 읽어 보세요.
참고
TPL 데이터 흐름 라이브러리(System.Threading.Tasks.Dataflow 네임스페이스)는 .NET과 함께 배포되지 않습니다. Visual Studio에서 System.Threading.Tasks.Dataflow 네임스페이스를 설치하려면 프로젝트를 열고, 프로젝트 메뉴에서 NuGet 패키지 관리를 선택한 후, System.Threading.Tasks.Dataflow 패키지를 온라인으로 검색합니다. 또는 .NET Core CLI를 사용하여 설치하려면 dotnet add package System.Threading.Tasks.Dataflow를 실행합니다.
슬라이딩 윈도우 데이터 흐름 블록 정의
입력 값을 버퍼링한 다음, 슬라이딩 윈도우 방식으로 출력해야 하는 데이터 흐름 애플리케이션을 고려합니다. 예를 들어 입력 값 {0, 1, 2, 3, 4, 5} 및 윈도우 크기가 3인 경우 슬라이딩 윈도우 데이터 흐름 블록은 출력 배열 {0, 1, 2}, {1, 2, 3}, {2, 3, 4}, {3, 4, 5}를 생성합니다. 다음 섹션에서는 이 사용자 지정 동작을 구현하는 데이터 흐름 블록 형식을 만드는 두 가지 방법을 보여줍니다. 첫 번째 방법은 Encapsulate 메서드를 사용하여 ISourceBlock<TOutput> 개체와 ITargetBlock<TInput> 개체의 기능을 하나의 전파자 블록으로 결합합니다. 두 번째 방법은 IPropagatorBlock<TInput,TOutput>에서 파생되는 클래스를 정의하고 기존 기능을 결합하여 사용자 지정 동작을 수행합니다.
Encapsulate 메서드를 사용하여 슬라이딩 윈도우 데이터 흐름 블록 정의
다음 예제에서는 Encapsulate 메서드를 사용하여 대상과 소스에서 전파자 블록을 만듭니다. 전파자 블록을 사용하면 소스 블록과 대상 블록이 데이터 수신자 및 송신자 역할을 할 수 있습니다.
이 방법은 사용자 지정 데이터 흐름 기능이 필요하지만 추가 메서드, 속성 또는 필드를 제공하는 형식이 필요하지 않은 경우 유용합니다.
// Creates a IPropagatorBlock<T, T[]> object propagates data in a
// sliding window fashion.
public static IPropagatorBlock<T, T[]> CreateSlidingWindow<T>(int windowSize)
{
// Create a queue to hold messages.
var queue = new Queue<T>();
// The source part of the propagator holds arrays of size windowSize
// and propagates data out to any connected targets.
var source = new BufferBlock<T[]>();
// The target part receives data and adds them to the queue.
var target = new ActionBlock<T>(item =>
{
// Add the item to the queue.
queue.Enqueue(item);
// Remove the oldest item when the queue size exceeds the window size.
if (queue.Count > windowSize)
queue.Dequeue();
// Post the data in the queue to the source block when the queue size
// equals the window size.
if (queue.Count == windowSize)
source.Post(queue.ToArray());
});
// When the target is set to the completed state, propagate out any
// remaining data and set the source to the completed state.
target.Completion.ContinueWith(delegate
{
if (queue.Count > 0 && queue.Count < windowSize)
source.Post(queue.ToArray());
source.Complete();
});
// Return a IPropagatorBlock<T, T[]> object that encapsulates the
// target and source blocks.
return DataflowBlock.Encapsulate(target, source);
}
' Creates a IPropagatorBlock<T, T[]> object propagates data in a
' sliding window fashion.
Public Shared Function CreateSlidingWindow(Of T)(ByVal windowSize As Integer) As IPropagatorBlock(Of T, T())
' Create a queue to hold messages.
Dim queue = New Queue(Of T)()
' The source part of the propagator holds arrays of size windowSize
' and propagates data out to any connected targets.
Dim source = New BufferBlock(Of T())()
' The target part receives data and adds them to the queue.
Dim target = New ActionBlock(Of T)(Sub(item)
' Add the item to the queue.
' Remove the oldest item when the queue size exceeds the window size.
' Post the data in the queue to the source block when the queue size
' equals the window size.
queue.Enqueue(item)
If queue.Count > windowSize Then
queue.Dequeue()
End If
If queue.Count = windowSize Then
source.Post(queue.ToArray())
End If
End Sub)
' When the target is set to the completed state, propagate out any
' remaining data and set the source to the completed state.
target.Completion.ContinueWith(Sub()
If queue.Count > 0 AndAlso queue.Count < windowSize Then
source.Post(queue.ToArray())
End If
source.Complete()
End Sub)
' Return a IPropagatorBlock<T, T[]> object that encapsulates the
' target and source blocks.
Return DataflowBlock.Encapsulate(target, source)
End Function
IPropagatorBlock에서 파생시켜 슬라이딩 윈도우 데이터 흐름 블록 정의
다음 예제에서는 SlidingWindowBlock 클래스를 보여줍니다. 이 클래스는 IPropagatorBlock<TInput,TOutput>에서 파생되므로 데이터의 소스 및 대상 역할을 할 수 있습니다. 이전 예제와 같이 SlidingWindowBlock 클래스는 기존 데이터 흐름 블록 형식에 빌드됩니다. 그러나 SlidingWindowBlock 클래스는 ISourceBlock<TOutput>, ITargetBlock<TInput> 및 IDataflowBlock 인터페이스에 필요한 메서드도 구현합니다. 이러한 메서드는 모두 미리 정의된 데이터 흐름 블록 형식 멤버에 대한 작업을 전달합니다. 예를 들어 Post 메서드는 작업을 ITargetBlock<TInput> 개체이기도 한 m_target 데이터 멤버로 전달합니다.
이 방법은 사용자 지정 데이터 흐름 기능이 필요하고 추가 메서드, 속성 또는 필드를 제공하는 형식도 필요한 경우 유용합니다. 예를 들어 SlidingWindowBlock 클래스는 IReceivableSourceBlock<TOutput>에서 파생되므로 TryReceive 및 TryReceiveAll 메서드를 제공할 수 있습니다. 또한 SlidingWindowBlock 클래스는 슬라이딩 윈도우에서 요소 수를 검색하는 WindowSize 속성을 제공하여 확장성을 보여줍니다.
// Propagates data in a sliding window fashion.
public class SlidingWindowBlock<T> : IPropagatorBlock<T, T[]>,
IReceivableSourceBlock<T[]>
{
// The size of the window.
private readonly int m_windowSize;
// The target part of the block.
private readonly ITargetBlock<T> m_target;
// The source part of the block.
private readonly IReceivableSourceBlock<T[]> m_source;
// Constructs a SlidingWindowBlock object.
public SlidingWindowBlock(int windowSize)
{
// Create a queue to hold messages.
var queue = new Queue<T>();
// The source part of the propagator holds arrays of size windowSize
// and propagates data out to any connected targets.
var source = new BufferBlock<T[]>();
// The target part receives data and adds them to the queue.
var target = new ActionBlock<T>(item =>
{
// Add the item to the queue.
queue.Enqueue(item);
// Remove the oldest item when the queue size exceeds the window size.
if (queue.Count > windowSize)
queue.Dequeue();
// Post the data in the queue to the source block when the queue size
// equals the window size.
if (queue.Count == windowSize)
source.Post(queue.ToArray());
});
// When the target is set to the completed state, propagate out any
// remaining data and set the source to the completed state.
target.Completion.ContinueWith(delegate
{
if (queue.Count > 0 && queue.Count < windowSize)
source.Post(queue.ToArray());
source.Complete();
});
m_windowSize = windowSize;
m_target = target;
m_source = source;
}
// Retrieves the size of the window.
public int WindowSize { get { return m_windowSize; } }
#region IReceivableSourceBlock<TOutput> members
// Attempts to synchronously receive an item from the source.
public bool TryReceive(Predicate<T[]> filter, out T[] item)
{
return m_source.TryReceive(filter, out item);
}
// Attempts to remove all available elements from the source into a new
// array that is returned.
public bool TryReceiveAll(out IList<T[]> items)
{
return m_source.TryReceiveAll(out items);
}
#endregion
#region ISourceBlock<TOutput> members
// Links this dataflow block to the provided target.
public IDisposable LinkTo(ITargetBlock<T[]> target, DataflowLinkOptions linkOptions)
{
return m_source.LinkTo(target, linkOptions);
}
// Called by a target to reserve a message previously offered by a source
// but not yet consumed by this target.
bool ISourceBlock<T[]>.ReserveMessage(DataflowMessageHeader messageHeader,
ITargetBlock<T[]> target)
{
return m_source.ReserveMessage(messageHeader, target);
}
// Called by a target to consume a previously offered message from a source.
T[] ISourceBlock<T[]>.ConsumeMessage(DataflowMessageHeader messageHeader,
ITargetBlock<T[]> target, out bool messageConsumed)
{
return m_source.ConsumeMessage(messageHeader,
target, out messageConsumed);
}
// Called by a target to release a previously reserved message from a source.
void ISourceBlock<T[]>.ReleaseReservation(DataflowMessageHeader messageHeader,
ITargetBlock<T[]> target)
{
m_source.ReleaseReservation(messageHeader, target);
}
#endregion
#region ITargetBlock<TInput> members
// Asynchronously passes a message to the target block, giving the target the
// opportunity to consume the message.
DataflowMessageStatus ITargetBlock<T>.OfferMessage(DataflowMessageHeader messageHeader,
T messageValue, ISourceBlock<T> source, bool consumeToAccept)
{
return m_target.OfferMessage(messageHeader,
messageValue, source, consumeToAccept);
}
#endregion
#region IDataflowBlock members
// Gets a Task that represents the completion of this dataflow block.
public Task Completion { get { return m_source.Completion; } }
// Signals to this target block that it should not accept any more messages,
// nor consume postponed messages.
public void Complete()
{
m_target.Complete();
}
public void Fault(Exception error)
{
m_target.Fault(error);
}
#endregion
}
' Propagates data in a sliding window fashion.
Public Class SlidingWindowBlock(Of T)
Implements IPropagatorBlock(Of T, T()), IReceivableSourceBlock(Of T())
' The size of the window.
Private ReadOnly m_windowSize As Integer
' The target part of the block.
Private ReadOnly m_target As ITargetBlock(Of T)
' The source part of the block.
Private ReadOnly m_source As IReceivableSourceBlock(Of T())
' Constructs a SlidingWindowBlock object.
Public Sub New(ByVal windowSize As Integer)
' Create a queue to hold messages.
Dim queue = New Queue(Of T)()
' The source part of the propagator holds arrays of size windowSize
' and propagates data out to any connected targets.
Dim source = New BufferBlock(Of T())()
' The target part receives data and adds them to the queue.
Dim target = New ActionBlock(Of T)(Sub(item)
' Add the item to the queue.
' Remove the oldest item when the queue size exceeds the window size.
' Post the data in the queue to the source block when the queue size
' equals the window size.
queue.Enqueue(item)
If queue.Count > windowSize Then
queue.Dequeue()
End If
If queue.Count = windowSize Then
source.Post(queue.ToArray())
End If
End Sub)
' When the target is set to the completed state, propagate out any
' remaining data and set the source to the completed state.
target.Completion.ContinueWith(Sub()
If queue.Count > 0 AndAlso queue.Count < windowSize Then
source.Post(queue.ToArray())
End If
source.Complete()
End Sub)
m_windowSize = windowSize
m_target = target
m_source = source
End Sub
' Retrieves the size of the window.
Public ReadOnly Property WindowSize() As Integer
Get
Return m_windowSize
End Get
End Property
'#Region "IReceivableSourceBlock<TOutput> members"
' Attempts to synchronously receive an item from the source.
Public Function TryReceive(ByVal filter As Predicate(Of T()), <System.Runtime.InteropServices.Out()> ByRef item() As T) As Boolean Implements IReceivableSourceBlock(Of T()).TryReceive
Return m_source.TryReceive(filter, item)
End Function
' Attempts to remove all available elements from the source into a new
' array that is returned.
Public Function TryReceiveAll(<System.Runtime.InteropServices.Out()> ByRef items As IList(Of T())) As Boolean Implements IReceivableSourceBlock(Of T()).TryReceiveAll
Return m_source.TryReceiveAll(items)
End Function
'#End Region
#Region "ISourceBlock<TOutput> members"
' Links this dataflow block to the provided target.
Public Function LinkTo(ByVal target As ITargetBlock(Of T()), ByVal linkOptions As DataflowLinkOptions) As IDisposable Implements ISourceBlock(Of T()).LinkTo
Return m_source.LinkTo(target, linkOptions)
End Function
' Called by a target to reserve a message previously offered by a source
' but not yet consumed by this target.
Private Function ReserveMessage(ByVal messageHeader As DataflowMessageHeader, ByVal target As ITargetBlock(Of T())) As Boolean Implements ISourceBlock(Of T()).ReserveMessage
Return m_source.ReserveMessage(messageHeader, target)
End Function
' Called by a target to consume a previously offered message from a source.
Private Function ConsumeMessage(ByVal messageHeader As DataflowMessageHeader, ByVal target As ITargetBlock(Of T()), ByRef messageConsumed As Boolean) As T() Implements ISourceBlock(Of T()).ConsumeMessage
Return m_source.ConsumeMessage(messageHeader, target, messageConsumed)
End Function
' Called by a target to release a previously reserved message from a source.
Private Sub ReleaseReservation(ByVal messageHeader As DataflowMessageHeader, ByVal target As ITargetBlock(Of T())) Implements ISourceBlock(Of T()).ReleaseReservation
m_source.ReleaseReservation(messageHeader, target)
End Sub
#End Region
#Region "ITargetBlock<TInput> members"
' Asynchronously passes a message to the target block, giving the target the
' opportunity to consume the message.
Private Function OfferMessage(ByVal messageHeader As DataflowMessageHeader, ByVal messageValue As T, ByVal source As ISourceBlock(Of T), ByVal consumeToAccept As Boolean) As DataflowMessageStatus Implements ITargetBlock(Of T).OfferMessage
Return m_target.OfferMessage(messageHeader, messageValue, source, consumeToAccept)
End Function
#End Region
#Region "IDataflowBlock members"
' Gets a Task that represents the completion of this dataflow block.
Public ReadOnly Property Completion() As Task Implements IDataflowBlock.Completion
Get
Return m_source.Completion
End Get
End Property
' Signals to this target block that it should not accept any more messages,
' nor consume postponed messages.
Public Sub Complete() Implements IDataflowBlock.Complete
m_target.Complete()
End Sub
Public Sub Fault(ByVal [error] As Exception) Implements IDataflowBlock.Fault
m_target.Fault([error])
End Sub
#End Region
End Class
전체 예제
다음 예제에서는 이 연습의 전체 코드를 보여 줍니다. 또한 블록에 쓰고, 블록에서 읽고, 콘솔에 결과를 인쇄하는 메서드에서 두 슬라이딩 윈도우 블록을 모두 사용하는 방법을 보여줍니다.
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
// Demonstrates how to create a custom dataflow block type.
class Program
{
// Creates a IPropagatorBlock<T, T[]> object propagates data in a
// sliding window fashion.
public static IPropagatorBlock<T, T[]> CreateSlidingWindow<T>(int windowSize)
{
// Create a queue to hold messages.
var queue = new Queue<T>();
// The source part of the propagator holds arrays of size windowSize
// and propagates data out to any connected targets.
var source = new BufferBlock<T[]>();
// The target part receives data and adds them to the queue.
var target = new ActionBlock<T>(item =>
{
// Add the item to the queue.
queue.Enqueue(item);
// Remove the oldest item when the queue size exceeds the window size.
if (queue.Count > windowSize)
queue.Dequeue();
// Post the data in the queue to the source block when the queue size
// equals the window size.
if (queue.Count == windowSize)
source.Post(queue.ToArray());
});
// When the target is set to the completed state, propagate out any
// remaining data and set the source to the completed state.
target.Completion.ContinueWith(delegate
{
if (queue.Count > 0 && queue.Count < windowSize)
source.Post(queue.ToArray());
source.Complete();
});
// Return a IPropagatorBlock<T, T[]> object that encapsulates the
// target and source blocks.
return DataflowBlock.Encapsulate(target, source);
}
// Propagates data in a sliding window fashion.
public class SlidingWindowBlock<T> : IPropagatorBlock<T, T[]>,
IReceivableSourceBlock<T[]>
{
// The size of the window.
private readonly int m_windowSize;
// The target part of the block.
private readonly ITargetBlock<T> m_target;
// The source part of the block.
private readonly IReceivableSourceBlock<T[]> m_source;
// Constructs a SlidingWindowBlock object.
public SlidingWindowBlock(int windowSize)
{
// Create a queue to hold messages.
var queue = new Queue<T>();
// The source part of the propagator holds arrays of size windowSize
// and propagates data out to any connected targets.
var source = new BufferBlock<T[]>();
// The target part receives data and adds them to the queue.
var target = new ActionBlock<T>(item =>
{
// Add the item to the queue.
queue.Enqueue(item);
// Remove the oldest item when the queue size exceeds the window size.
if (queue.Count > windowSize)
queue.Dequeue();
// Post the data in the queue to the source block when the queue size
// equals the window size.
if (queue.Count == windowSize)
source.Post(queue.ToArray());
});
// When the target is set to the completed state, propagate out any
// remaining data and set the source to the completed state.
target.Completion.ContinueWith(delegate
{
if (queue.Count > 0 && queue.Count < windowSize)
source.Post(queue.ToArray());
source.Complete();
});
m_windowSize = windowSize;
m_target = target;
m_source = source;
}
// Retrieves the size of the window.
public int WindowSize { get { return m_windowSize; } }
#region IReceivableSourceBlock<TOutput> members
// Attempts to synchronously receive an item from the source.
public bool TryReceive(Predicate<T[]> filter, out T[] item)
{
return m_source.TryReceive(filter, out item);
}
// Attempts to remove all available elements from the source into a new
// array that is returned.
public bool TryReceiveAll(out IList<T[]> items)
{
return m_source.TryReceiveAll(out items);
}
#endregion
#region ISourceBlock<TOutput> members
// Links this dataflow block to the provided target.
public IDisposable LinkTo(ITargetBlock<T[]> target, DataflowLinkOptions linkOptions)
{
return m_source.LinkTo(target, linkOptions);
}
// Called by a target to reserve a message previously offered by a source
// but not yet consumed by this target.
bool ISourceBlock<T[]>.ReserveMessage(DataflowMessageHeader messageHeader,
ITargetBlock<T[]> target)
{
return m_source.ReserveMessage(messageHeader, target);
}
// Called by a target to consume a previously offered message from a source.
T[] ISourceBlock<T[]>.ConsumeMessage(DataflowMessageHeader messageHeader,
ITargetBlock<T[]> target, out bool messageConsumed)
{
return m_source.ConsumeMessage(messageHeader,
target, out messageConsumed);
}
// Called by a target to release a previously reserved message from a source.
void ISourceBlock<T[]>.ReleaseReservation(DataflowMessageHeader messageHeader,
ITargetBlock<T[]> target)
{
m_source.ReleaseReservation(messageHeader, target);
}
#endregion
#region ITargetBlock<TInput> members
// Asynchronously passes a message to the target block, giving the target the
// opportunity to consume the message.
DataflowMessageStatus ITargetBlock<T>.OfferMessage(DataflowMessageHeader messageHeader,
T messageValue, ISourceBlock<T> source, bool consumeToAccept)
{
return m_target.OfferMessage(messageHeader,
messageValue, source, consumeToAccept);
}
#endregion
#region IDataflowBlock members
// Gets a Task that represents the completion of this dataflow block.
public Task Completion { get { return m_source.Completion; } }
// Signals to this target block that it should not accept any more messages,
// nor consume postponed messages.
public void Complete()
{
m_target.Complete();
}
public void Fault(Exception error)
{
m_target.Fault(error);
}
#endregion
}
// Demonstrates usage of the sliding window block by sending the provided
// values to the provided propagator block and printing the output of
// that block to the console.
static void DemonstrateSlidingWindow<T>(IPropagatorBlock<T, T[]> slidingWindow,
IEnumerable<T> values)
{
// Create an action block that prints arrays of data to the console.
string windowComma = string.Empty;
var printWindow = new ActionBlock<T[]>(window =>
{
Console.Write(windowComma);
Console.Write("{");
string comma = string.Empty;
foreach (T item in window)
{
Console.Write(comma);
Console.Write(item);
comma = ",";
}
Console.Write("}");
windowComma = ", ";
});
// Link the printer block to the sliding window block.
slidingWindow.LinkTo(printWindow);
// Set the printer block to the completed state when the sliding window
// block completes.
slidingWindow.Completion.ContinueWith(delegate { printWindow.Complete(); });
// Print an additional newline to the console when the printer block completes.
var completion = printWindow.Completion.ContinueWith(delegate { Console.WriteLine(); });
// Post the provided values to the sliding window block and then wait
// for the sliding window block to complete.
foreach (T value in values)
{
slidingWindow.Post(value);
}
slidingWindow.Complete();
// Wait for the printer to complete and perform its final action.
completion.Wait();
}
static void Main(string[] args)
{
Console.Write("Using the DataflowBlockExtensions.Encapsulate method ");
Console.WriteLine("(T=int, windowSize=3):");
DemonstrateSlidingWindow(CreateSlidingWindow<int>(3), Enumerable.Range(0, 10));
Console.WriteLine();
var slidingWindow = new SlidingWindowBlock<char>(4);
Console.Write("Using SlidingWindowBlock<T> ");
Console.WriteLine("(T=char, windowSize={0}):", slidingWindow.WindowSize);
DemonstrateSlidingWindow(slidingWindow, from n in Enumerable.Range(65, 10)
select (char)n);
}
}
/* Output:
Using the DataflowBlockExtensions.Encapsulate method (T=int, windowSize=3):
{0,1,2}, {1,2,3}, {2,3,4}, {3,4,5}, {4,5,6}, {5,6,7}, {6,7,8}, {7,8,9}
Using SlidingWindowBlock<T> (T=char, windowSize=4):
{A,B,C,D}, {B,C,D,E}, {C,D,E,F}, {D,E,F,G}, {E,F,G,H}, {F,G,H,I}, {G,H,I,J}
*/
Imports System.Collections.Generic
Imports System.Linq
Imports System.Threading.Tasks
Imports System.Threading.Tasks.Dataflow
' Demonstrates how to create a custom dataflow block type.
Friend Class Program
' Creates a IPropagatorBlock<T, T[]> object propagates data in a
' sliding window fashion.
Public Shared Function CreateSlidingWindow(Of T)(ByVal windowSize As Integer) As IPropagatorBlock(Of T, T())
' Create a queue to hold messages.
Dim queue = New Queue(Of T)()
' The source part of the propagator holds arrays of size windowSize
' and propagates data out to any connected targets.
Dim source = New BufferBlock(Of T())()
' The target part receives data and adds them to the queue.
Dim target = New ActionBlock(Of T)(Sub(item)
' Add the item to the queue.
' Remove the oldest item when the queue size exceeds the window size.
' Post the data in the queue to the source block when the queue size
' equals the window size.
queue.Enqueue(item)
If queue.Count > windowSize Then
queue.Dequeue()
End If
If queue.Count = windowSize Then
source.Post(queue.ToArray())
End If
End Sub)
' When the target is set to the completed state, propagate out any
' remaining data and set the source to the completed state.
target.Completion.ContinueWith(Sub()
If queue.Count > 0 AndAlso queue.Count < windowSize Then
source.Post(queue.ToArray())
End If
source.Complete()
End Sub)
' Return a IPropagatorBlock<T, T[]> object that encapsulates the
' target and source blocks.
Return DataflowBlock.Encapsulate(target, source)
End Function
' Propagates data in a sliding window fashion.
Public Class SlidingWindowBlock(Of T)
Implements IPropagatorBlock(Of T, T()), IReceivableSourceBlock(Of T())
' The size of the window.
Private ReadOnly m_windowSize As Integer
' The target part of the block.
Private ReadOnly m_target As ITargetBlock(Of T)
' The source part of the block.
Private ReadOnly m_source As IReceivableSourceBlock(Of T())
' Constructs a SlidingWindowBlock object.
Public Sub New(ByVal windowSize As Integer)
' Create a queue to hold messages.
Dim queue = New Queue(Of T)()
' The source part of the propagator holds arrays of size windowSize
' and propagates data out to any connected targets.
Dim source = New BufferBlock(Of T())()
' The target part receives data and adds them to the queue.
Dim target = New ActionBlock(Of T)(Sub(item)
' Add the item to the queue.
' Remove the oldest item when the queue size exceeds the window size.
' Post the data in the queue to the source block when the queue size
' equals the window size.
queue.Enqueue(item)
If queue.Count > windowSize Then
queue.Dequeue()
End If
If queue.Count = windowSize Then
source.Post(queue.ToArray())
End If
End Sub)
' When the target is set to the completed state, propagate out any
' remaining data and set the source to the completed state.
target.Completion.ContinueWith(Sub()
If queue.Count > 0 AndAlso queue.Count < windowSize Then
source.Post(queue.ToArray())
End If
source.Complete()
End Sub)
m_windowSize = windowSize
m_target = target
m_source = source
End Sub
' Retrieves the size of the window.
Public ReadOnly Property WindowSize() As Integer
Get
Return m_windowSize
End Get
End Property
'#Region "IReceivableSourceBlock<TOutput> members"
' Attempts to synchronously receive an item from the source.
Public Function TryReceive(ByVal filter As Predicate(Of T()), <System.Runtime.InteropServices.Out()> ByRef item() As T) As Boolean Implements IReceivableSourceBlock(Of T()).TryReceive
Return m_source.TryReceive(filter, item)
End Function
' Attempts to remove all available elements from the source into a new
' array that is returned.
Public Function TryReceiveAll(<System.Runtime.InteropServices.Out()> ByRef items As IList(Of T())) As Boolean Implements IReceivableSourceBlock(Of T()).TryReceiveAll
Return m_source.TryReceiveAll(items)
End Function
'#End Region
#Region "ISourceBlock<TOutput> members"
' Links this dataflow block to the provided target.
Public Function LinkTo(ByVal target As ITargetBlock(Of T()), ByVal linkOptions As DataflowLinkOptions) As IDisposable Implements ISourceBlock(Of T()).LinkTo
Return m_source.LinkTo(target, linkOptions)
End Function
' Called by a target to reserve a message previously offered by a source
' but not yet consumed by this target.
Private Function ReserveMessage(ByVal messageHeader As DataflowMessageHeader, ByVal target As ITargetBlock(Of T())) As Boolean Implements ISourceBlock(Of T()).ReserveMessage
Return m_source.ReserveMessage(messageHeader, target)
End Function
' Called by a target to consume a previously offered message from a source.
Private Function ConsumeMessage(ByVal messageHeader As DataflowMessageHeader, ByVal target As ITargetBlock(Of T()), ByRef messageConsumed As Boolean) As T() Implements ISourceBlock(Of T()).ConsumeMessage
Return m_source.ConsumeMessage(messageHeader, target, messageConsumed)
End Function
' Called by a target to release a previously reserved message from a source.
Private Sub ReleaseReservation(ByVal messageHeader As DataflowMessageHeader, ByVal target As ITargetBlock(Of T())) Implements ISourceBlock(Of T()).ReleaseReservation
m_source.ReleaseReservation(messageHeader, target)
End Sub
#End Region
#Region "ITargetBlock<TInput> members"
' Asynchronously passes a message to the target block, giving the target the
' opportunity to consume the message.
Private Function OfferMessage(ByVal messageHeader As DataflowMessageHeader, ByVal messageValue As T, ByVal source As ISourceBlock(Of T), ByVal consumeToAccept As Boolean) As DataflowMessageStatus Implements ITargetBlock(Of T).OfferMessage
Return m_target.OfferMessage(messageHeader, messageValue, source, consumeToAccept)
End Function
#End Region
#Region "IDataflowBlock members"
' Gets a Task that represents the completion of this dataflow block.
Public ReadOnly Property Completion() As Task Implements IDataflowBlock.Completion
Get
Return m_source.Completion
End Get
End Property
' Signals to this target block that it should not accept any more messages,
' nor consume postponed messages.
Public Sub Complete() Implements IDataflowBlock.Complete
m_target.Complete()
End Sub
Public Sub Fault(ByVal [error] As Exception) Implements IDataflowBlock.Fault
m_target.Fault([error])
End Sub
#End Region
End Class
' Demonstrates usage of the sliding window block by sending the provided
' values to the provided propagator block and printing the output of
' that block to the console.
Private Shared Sub DemonstrateSlidingWindow(Of T)(ByVal slidingWindow As IPropagatorBlock(Of T, T()), ByVal values As IEnumerable(Of T))
' Create an action block that prints arrays of data to the console.
Dim windowComma As String = String.Empty
Dim printWindow = New ActionBlock(Of T())(Sub(window)
Console.Write(windowComma)
Console.Write("{")
Dim comma As String = String.Empty
For Each item As T In window
Console.Write(comma)
Console.Write(item)
comma = ","
Next item
Console.Write("}")
windowComma = ", "
End Sub)
' Link the printer block to the sliding window block.
slidingWindow.LinkTo(printWindow)
' Set the printer block to the completed state when the sliding window
' block completes.
slidingWindow.Completion.ContinueWith(Sub() printWindow.Complete())
' Print an additional newline to the console when the printer block completes.
Dim completion = printWindow.Completion.ContinueWith(Sub() Console.WriteLine())
' Post the provided values to the sliding window block and then wait
' for the sliding window block to complete.
For Each value As T In values
slidingWindow.Post(value)
Next value
slidingWindow.Complete()
' Wait for the printer to complete and perform its final action.
completion.Wait()
End Sub
Shared Sub Main(ByVal args() As String)
Console.Write("Using the DataflowBlockExtensions.Encapsulate method ")
Console.WriteLine("(T=int, windowSize=3):")
DemonstrateSlidingWindow(CreateSlidingWindow(Of Integer)(3), Enumerable.Range(0, 10))
Console.WriteLine()
Dim slidingWindow = New SlidingWindowBlock(Of Char)(4)
Console.Write("Using SlidingWindowBlock<T> ")
Console.WriteLine("(T=char, windowSize={0}):", slidingWindow.WindowSize)
DemonstrateSlidingWindow(slidingWindow, _
From n In Enumerable.Range(65, 10) _
Select ChrW(n))
End Sub
End Class
' Output:
'Using the DataflowBlockExtensions.Encapsulate method (T=int, windowSize=3):
'{0,1,2}, {1,2,3}, {2,3,4}, {3,4,5}, {4,5,6}, {5,6,7}, {6,7,8}, {7,8,9}
'
'Using SlidingWindowBlock<T> (T=char, windowSize=4):
'{A,B,C,D}, {B,C,D,E}, {C,D,E,F}, {D,E,F,G}, {E,F,G,H}, {F,G,H,I}, {G,H,I,J}
'
참조
GitHub에서 Microsoft와 공동 작업
이 콘텐츠의 원본은 GitHub에서 찾을 수 있으며, 여기서 문제와 끌어오기 요청을 만들고 검토할 수도 있습니다. 자세한 내용은 참여자 가이드를 참조하세요.
.NET
피드백
출시 예정: 2024년 내내 콘텐츠에 대한 피드백 메커니즘으로 GitHub 문제를 단계적으로 폐지하고 이를 새로운 피드백 시스템으로 바꿀 예정입니다. 자세한 내용은 다음을 참조하세요. https://aka.ms/ContentUserFeedback
다음에 대한 사용자 의견 제출 및 보기