Veri Akışı (Görev Paralel Kitaplığı)
Görev Paralel Kitaplığı (TPL), eşzamanlılık özellikli uygulamaların sağlamlığını artırmaya yardımcı olmak için veri akışı bileşenleri sağlar. Bu veri akışı bileşenleri topluca TPL Veri Akışı Kitaplığı olarak adlandırılır. Bu veri akışı modeli, büyük parçalı veri akışı ve ardışık düzen oluşturma görevleri için işlem içi ileti geçirme sağlayarak aktör temelli programlamayı teşvik eder. Veri akışı bileşenleri, TPL'nin türleri ve zamanlama altyapısını temel alır ve zaman uyumsuz programlama için C#, Visual Basic ve F# dil desteğiyle tümleşir. Bu veri akışı bileşenleri, birbirleriyle zaman uyumsuz olarak iletişim kurması gereken birden fazla işleminiz varsa veya verileri elde ettikçe işlemek istiyorsanız kullanışlıdır. Örneğin, web kamerasından gelen görüntü verilerini işleyen bir uygulamayı düşünün. Veri akışı modelini kullanarak, uygulama görüntü karelerini kullanılabilir oldukça işleyebilir. Uygulama, örneğin ışık düzeltmesi veya kırmızı göz azaltma gerçekleştirerek görüntü çerçevelerini geliştirirse, veri akışı bileşenlerinden oluşan bir işlem hattı oluşturabilirsiniz. Ardışık düzenin her aşaması, görüntüyü dönüştürmek için TPL tarafından sağlanan işlevsellik gibi daha büyük parçalı paralellik işlevlerini kullanabilir.
Bu belge, TPL Veri Akışı Kitaplığı'nın ana hatlarını sunmaktadır. Programlama modelini, önceden tanımlı veri akışı bloğu türlerini ve veri akışı bloklarının uygulamanızın özel gereksinimlerini karşılaması için nasıl yapılandırılabileceğini açıklar.
Not
TPL Veri Akışı Kitaplığı ( System.Threading.Tasks.Dataflow ad alanı) .NET ile dağıtılmaz. Ad alanını System.Threading.Tasks.Dataflow Visual Studio'ya yüklemek için projenizi açın, Proje menüsünden NuGet Paketlerini Yönet'i seçin ve çevrimiçi ortamda System.Threading.Tasks.Dataflow paketi arayın. Alternatif olarak, .NET Core CLI kullanarak yüklemek için komutunu çalıştırındotnet add package System.Threading.Tasks.Dataflow.
Programlama Modeli
TPL Veri Akışı Kitaplığı, ileti geçirmenin yanı sıra yüksek veri hacmine ve düşük gecikmeye sahip CPU yoğun ve I/O yoğun uygulamalarını paralelleştirmek için bir temel sağlar. Ayrıca, verilerin arabelleğe nasıl alınacağı ve sistemde nasıl hareket edeceği konusunda doğrudan denetim olanağı sağlar. Veri akışı programlama modelini daha iyi anlamak için, diskten zaman uyumsuz olarak görüntü yükleyen ve bu görüntülerin bileşimini oluşturan bir uygulamayı düşünün. Geleneksel programlama modelleri, görevleri koordine etmek ve paylaşılan verilere erişmek için genellikle kilitler gibi eşitleme nesneleri ve geri çağrılar kullanmanızı gerektirir. Veri akışı programlama modelini kullanarak, görüntüleri diskten okundukça işleyen veri akışı nesneleri oluşturabilirsiniz. Veri akışı modelinde, verinin kullanılabilir olduğunda nasıl işleneceğini ve varsa veriler arsındaki bağımlılıkları siz belirtirsiniz. Veriler arasındaki bağımlılıkları çalışma zamanı yönettiği için, paylaşılan verilere erişimi eşitleme gereksinimini genellikle önleyebilirsiniz. Ek olarak, çalışma programları işi verilerinin zaman uyumsuz gelişine göre belirlendiği için, veri akışı temel iş parçacıklarını etkin olarak yöneterek uygulamanın hassaslığını ve veri hacmini artırabilir. Windows Forms uygulamasında görüntü işleme uygulamak için veri akışı programlama modelini kullanan bir örnek için bkz . İzlenecek Yol: Windows Forms Uygulamasında Veri Akışını Kullanma.
Kaynaklar ve Hedefler
TPL Veri Akışı Kitaplığı, verileri arabelleğe alan ve işleyen veri yapıları olan veri akışı bloklarından oluşur. TPL üç tür veri akışı bloğu tanımlar: kaynak bloklar, hedef bloklar ve yayıcı bloklar. Bir kaynak blok, veriler için kaynak görevi görür ve okunabilir. Bir hedef blok, veriler için alıcı görevi görür ve yazılabilir. Bir yayıcı blok, hem kaynak blok hem de hedef blok olarak görev görür ve hem okunabilir, hem de yazılabilir. TPL, kaynakları temsil temek için System.Threading.Tasks.Dataflow.ISourceBlock<TOutput> arabirimini, hedefleri temsil etmek için System.Threading.Tasks.Dataflow.ITargetBlock<TInput> arabirimini ve yayıcıları temsil etmek için System.Threading.Tasks.Dataflow.IPropagatorBlock<TInput,TOutput> arabirimini tanımlar. IPropagatorBlock<TInput,TOutput>, hem ISourceBlock<TOutput> hem de ITargetBlock<TInput> arabirimlerinden devralır.
TPL Veri Akışı Kitaplığı ISourceBlock<TOutput>, ITargetBlock<TInput> ve IPropagatorBlock<TInput,TOutput> arabirimlerini uygulayan önceden tanımlı birkaç veri akışı bloğu türü sağlar. Bu veri akışı bloğu türleri, bu belgede Önceden Tanımlanmış Veri Akışı Blok Türleri bölümünde açıklanmıştır.
Blokları Bağlama
Veri akışı bloklarını, veri akışı bloklarının doğrusal dizileri olan form işlem hatlarına veya veri akışı bloklarının grafikleri olan ağlara bağlayabilirsiniz. Ardışık düzen, bir ağ türüdür. Bir ardışık düzen veya ağda, kaynaklar, veriler kullanılabilir oldukça zaman uyumsuz olarak hedeflere yayarlar. ISourceBlock<TOutput>.LinkTo yöntemi, bir kaynak veri akışı bloğunu hedef bloğa bağlar. Bir kaynak sıfır veya daha fazla hedefe bağlanabilir; hedeflere sıfır veya daha fazla kaynaktan bağlanılabilir. Bir ardışık düzen veya ağda eşzamanlı olarak veri akışı bloğu ekleme ve kaldırma işlemleri yapabilirsiniz. Önceden tanımlı veri akışı bloğu türleri bağlama ve bağlantıyı kaldırma işlemlerinin iş parçacığı güvenliğini tüm boyutlarıyla işler.
Temel bir işlem hattı oluşturmak için veri akışı bloklarını bağlayan bir örnek için bkz . İzlenecek Yol: Veri Akışı İşlem Hattı Oluşturma. Daha karmaşık bir ağ oluşturmak için veri akışı bloklarını bağlayan bir örnek için bkz . İzlenecek Yol: Windows Forms Uygulamasında Veri Akışını Kullanma. Kaynak bir iletiyi sunduğunda bir hedefle bağlantıyı kaldıran bir örnek için bkz . Nasıl yapılır: Veri Akışı Bloklarının Bağlantısını Kaldırma.
Filtreleme
Bir kaynağı hedefe bağlamak için ISourceBlock<TOutput>.LinkTo yöntemini çağırdığınızda, hedef bloğun bir iletiyi değerine göre kabul edeceğini veya reddedeceğini belirleyen bir temsilci sağlayabilirsiniz. Bu filtreleme mekanizması, bir veri akışı bloğunun yalnızca belirli değerleri almasını sağlamak için kullanışlı bir yoldur. Çoğu önceden tanımlı veri akışı bloğu türünde, eğer bir kaynak blok birden çok hedef bloğa bağlı ise, bir hedef blok iletiyi reddettiğinde kaynak o iletiyi sonraki hedefe sunar. Kaynağın hedeflere iletileri hangi sırayla sunacağı kaynak tarafından tanımlanır ve kaynağın türüne göre farklılık gösterebilir. Çoğu kaynak blok türü, bir hedef iletiyi kabul ettiğinde o iletiyi sunmayı bırakır. Bazı hedefler iletiyi reddetse her iletiyi tüm hedeflere sunan BroadcastBlock<T> sınıfı, bu kural için bir istisnadır. Yalnızca belirli iletileri işlemek için filtreleme kullanan bir örnek için bkz . İzlenecek Yol: Windows Forms Uygulamasında Veri Akışını Kullanma.
Önemli
Önceden tanımlı veri akışı bloklarının her biri iletilerin alındıkları sırayla gönderilmesini sağladığından, kaynak bloğun sonraki iletiyi işleyebilmesi için kaynak bloktaki her ileti okunmalıdır. Bu nedenle, birden çok hedefi tek bir kaynağa bağlamak için filtreleme kullandığınızda her iletiyi en az bir hedef bloğun aldığından emin olun. Aksi halde, uygulamanız kilitlenebilir.
İleti Geçirme
Veri akışı programlama modeli, bir programın bağımsız bileşenlerinin ileti göndererek birbirleriyle iletişim kurduğu ileti geçirme kavramıyla ilgilidir. İletileri uygulama bileşenleri arasında yayma yollarından biri, hedef veri akışı bloklarına ileti göndermek için (zaman uyumlu) ve SendAsync (zaman uyumsuz) yöntemlerini, Receivekaynak bloklardan ileti almak için ise , ReceiveAsyncve TryReceive yöntemlerini çağırmaktır Post . Giriş verilerini baş düğüme (hedef blok) göndererek ve işlem hattının terminal düğümünden veya ağın terminal düğümlerinden (bir veya daha fazla kaynak blok) çıkış verileri alarak bu yöntemleri veri akışı işlem hatları veya ağlarla birleştirebilirsiniz. Ayrıca, verilere sahip olan kullanılabilir kaynakların ilkinden okuyup o veri üzerinde işlem yapabilmek için Choose yöntemini de kullanabilirsiniz.
Kaynak bloklar yöntemini çağırarak ITargetBlock<TInput>.OfferMessage hedef bloklara veri sunar. Hedef blok, sunulan bir iletiye üç şekilde cevap verebilir: İletiyi kabul edebilir, iletiyi reddedebilir veya iletiyi erteleyebilir. Hedef iletiyi kabul ettiğinde, OfferMessage yöntemi döndürülürAccepted. Hedef iletiyi reddettiğinde, OfferMessage yöntemi döndürülürDeclined. Hedef, kaynaktan daha fazla ileti almak istemediğinde OfferMessage değeri döndürülürDecliningPermanently. Önceden tanımlı kaynak blok türleri bu tür bir dönüş değeri alındığında bağlantılı hedeflere ileti sunmaz ve bu tür hedeflerle otomatik olarak bağlantılarını keserler.
Bir hedef bloğu iletiyi daha sonra kullanmak üzere ertelerse, OfferMessage yöntemi döndürülürPostponed. Bir iletiyi erteleyen bir hedef blok, daha sonra sunulan iletiyi ayırmaya çalışmak için yöntemini çağırabilir ISourceBlock<TOutput>.ReserveMessage . Bu noktada, ileti ya hala geçerlidir ve hedef blok tarafından kullanılabilir ya da başka bir hedef tarafından alınmıştır. Hedef blok daha sonra iletiyi gerektirdiğinde veya artık iletiye ihtiyaç duymadığında sırasıyla veya ReleaseReservation yöntemini çağırırISourceBlock<TOutput>.ConsumeMessage. İleti ayırma işlemi, genellikle doyumsuz olmayan modda çalışan veri akışı bloğu türleri tarafından kullanılır. Doyumsuz olmayan mod, bu belgenin ilerleyen bölümlerinde açıklanmıştır. Bir hedef blok, ertelenmiş bir iletiyi ayırmak yerine, ertelenmiş iletiyi doğrudan tüketmeyi denemek için ISourceBlock<TOutput>.ConsumeMessage yöntemini de kullanabilir.
Veri Akışı Bloğunun Tamamlanması
Veri akışı blokları tamamlama kavramını da destekler. Tamamlanmış durumdaki bir veri akışı bloğu, daha fazla iş yapmaz. Her veri akışı bloğunun, bloğun tamamlanma durumunu temsil eden, tamamlama görevi olarak bilinen ilişkili System.Threading.Tasks.Task bir nesnesi vardır. Bir Task nesnesinin tamamlanmasını bekleyebildiğinizden, tamamlanma görevlerini kullanarak bir veri akışı düğümündeki bir veya daha fazla terminal düğümün bitmesini de bekleyebilirsiniz. IDataflowBlock arabirimi, veri akışı bloğuna tamamlanması yönünde bir istek olduğunu bildiren Complete yöntemini ve veri akışı bloğunu ilgilendiren tamamlanma görevini döndüren Completion özelliğini tanımlar. ISourceBlock<TOutput> ve ITargetBlock<TInput> öğelerinin her ikisi de IDataflowBlock arabiriminden devralır.
Bir veri akışı bloğunun hatasız olarak tamamlanıp tamamlanmadığını, bir veya daha fazla hatayla karşılaşıp karşılaşmadığını veya iptal edilip edilmediğini belirlemek için iki yol vardır. İlk yol, bir bloktaki tamamlama görevinde (Catch-TryVisual Basic'tetrycatch-) yöntemini çağırmaktır.Task.Wait Aşağıdaki örnek, giriş değeri sıfırdan daha azsa bir ActionBlock<TInput> oluşturan bir ArgumentOutOfRangeException nesnesi oluşturur. Bu örnek tamamlanma görevinde AggregateException yöntemini çağırınca Wait oluşturulur. ArgumentOutOfRangeException öğesine, InnerExceptions nesnesinin AggregateException özelliği aracılığıyla erişilir.
// Create an ActionBlock<int> object that prints its input
// and throws ArgumentOutOfRangeException if the input
// is less than zero.
var throwIfNegative = new ActionBlock<int>(n =>
{
Console.WriteLine("n = {0}", n);
if (n < 0)
{
throw new ArgumentOutOfRangeException();
}
});
// Post values to the block.
throwIfNegative.Post(0);
throwIfNegative.Post(-1);
throwIfNegative.Post(1);
throwIfNegative.Post(-2);
throwIfNegative.Complete();
// Wait for completion in a try/catch block.
try
{
throwIfNegative.Completion.Wait();
}
catch (AggregateException ae)
{
// If an unhandled exception occurs during dataflow processing, all
// exceptions are propagated through an AggregateException object.
ae.Handle(e =>
{
Console.WriteLine("Encountered {0}: {1}",
e.GetType().Name, e.Message);
return true;
});
}
/* Output:
n = 0
n = -1
Encountered ArgumentOutOfRangeException: Specified argument was out of the range
of valid values.
*/
' Create an ActionBlock<int> object that prints its input
' and throws ArgumentOutOfRangeException if the input
' is less than zero.
Dim throwIfNegative = New ActionBlock(Of Integer)(Sub(n)
Console.WriteLine("n = {0}", n)
If n < 0 Then
Throw New ArgumentOutOfRangeException()
End If
End Sub)
' Post values to the block.
throwIfNegative.Post(0)
throwIfNegative.Post(-1)
throwIfNegative.Post(1)
throwIfNegative.Post(-2)
throwIfNegative.Complete()
' Wait for completion in a try/catch block.
Try
throwIfNegative.Completion.Wait()
Catch ae As AggregateException
' If an unhandled exception occurs during dataflow processing, all
' exceptions are propagated through an AggregateException object.
ae.Handle(Function(e)
Console.WriteLine("Encountered {0}: {1}", e.GetType().Name, e.Message)
Return True
End Function)
End Try
' Output:
' n = 0
' n = -1
' Encountered ArgumentOutOfRangeException: Specified argument was out of the range
' of valid values.
'
Bu örnekte, özel bir durumunun işlenmediği bir yürütme veri akışı bloğu temsilcisi gösterilmektedir. Bu tür blokların gövdelerindeki özel durumları işlemenizi öneririz. Ancak, bunu yapmanız mümkün değilse, blok iptal edilmiş gibi davranır ve gelen iletileri işlemez.
Bir veri akışı bloğu açıkça iptal edildiğinde, AggregateException nesnesi, OperationCanceledException özelliğinde InnerExceptions öğesini içerir. Veri akışı iptali hakkında daha fazla bilgi için İptali Etkinleştirme bölümüne bakın.
Veri akışı bloğunun tamamlanma durumunu belirlemenin ikinci yolu, tamamlanma görevinin devamını kullanmak veya zaman uyumsuz olarak tamamlanma görevini beklemek için C# ve Visual Basic'in zaman uyumsuz dil özelliklerini kullanmaktır. Task.ContinueWith yöntemine sağladığınız temsilci, öncül görevi temsil eden bir Task nesnesi alır. Completion özelliğinde, devamlılık için atanan temsilci, tamamlanma görevinin kendisini alır. Aşağıdaki örnek öncekine benzer, ancak ContinueWith yöntemini kullanarak genel veri akışı işleminin durumunu yazdıran bir devamlılık görevi oluşturur.
// Create an ActionBlock<int> object that prints its input
// and throws ArgumentOutOfRangeException if the input
// is less than zero.
var throwIfNegative = new ActionBlock<int>(n =>
{
Console.WriteLine("n = {0}", n);
if (n < 0)
{
throw new ArgumentOutOfRangeException();
}
});
// Create a continuation task that prints the overall
// task status to the console when the block finishes.
throwIfNegative.Completion.ContinueWith(task =>
{
Console.WriteLine("The status of the completion task is '{0}'.",
task.Status);
});
// Post values to the block.
throwIfNegative.Post(0);
throwIfNegative.Post(-1);
throwIfNegative.Post(1);
throwIfNegative.Post(-2);
throwIfNegative.Complete();
// Wait for completion in a try/catch block.
try
{
throwIfNegative.Completion.Wait();
}
catch (AggregateException ae)
{
// If an unhandled exception occurs during dataflow processing, all
// exceptions are propagated through an AggregateException object.
ae.Handle(e =>
{
Console.WriteLine("Encountered {0}: {1}",
e.GetType().Name, e.Message);
return true;
});
}
/* Output:
n = 0
n = -1
The status of the completion task is 'Faulted'.
Encountered ArgumentOutOfRangeException: Specified argument was out of the range
of valid values.
*/
' Create an ActionBlock<int> object that prints its input
' and throws ArgumentOutOfRangeException if the input
' is less than zero.
Dim throwIfNegative = New ActionBlock(Of Integer)(Sub(n)
Console.WriteLine("n = {0}", n)
If n < 0 Then
Throw New ArgumentOutOfRangeException()
End If
End Sub)
' Create a continuation task that prints the overall
' task status to the console when the block finishes.
throwIfNegative.Completion.ContinueWith(Sub(task) Console.WriteLine("The status of the completion task is '{0}'.", task.Status))
' Post values to the block.
throwIfNegative.Post(0)
throwIfNegative.Post(-1)
throwIfNegative.Post(1)
throwIfNegative.Post(-2)
throwIfNegative.Complete()
' Wait for completion in a try/catch block.
Try
throwIfNegative.Completion.Wait()
Catch ae As AggregateException
' If an unhandled exception occurs during dataflow processing, all
' exceptions are propagated through an AggregateException object.
ae.Handle(Function(e)
Console.WriteLine("Encountered {0}: {1}", e.GetType().Name, e.Message)
Return True
End Function)
End Try
' Output:
' n = 0
' n = -1
' The status of the completion task is 'Faulted'.
' Encountered ArgumentOutOfRangeException: Specified argument was out of the range
' of valid values.
'
Bir veri akışı bloğunun tamamlanma durumu hakkında ek bilgi edinmek için devamlılık görevinin gövdesinde IsCanceled gibi özellikleri de kullanabilirsiniz. Devamlılık görevleri ve bunların iptal ve hata işlemeyle ilişkisi hakkında daha fazla bilgi için bkz . Devamlılık Görevlerini Kullanarak Görevleri Zincirleme, Görev İptali ve Özel Durum İşleme.
Önceden Tanımlı Veri Akışı Bloğu Türleri
TPL Veri Akışı Kitaplığı, birkaç önceden tanımlı veri akışı bloğu türü sağlar. Bu türler üç kategoriye ayrılır: arabelleğe alma blokları, yürütme blokları ve gruplandırma blokları. Aşağıdaki bölümlerde, bu kategorileri oluşturan blok türleri açıklanmaktadır.
Arabellek Blokları
Arabellek blokları, veri tüketicilerinin kullanılmak üzere veri tutar. TPL Veri Akışı Kitaplığı üç arabellek bloğu türü sağlar: System.Threading.Tasks.Dataflow.BufferBlock<T>, System.Threading.Tasks.Dataflow.BroadcastBlock<T> ve System.Threading.Tasks.Dataflow.WriteOnceBlock<T>.
BufferBlock<T>
BufferBlock<T> sınıfı, genel amaçlı bir zaman uyumsuz ileti yapısını temsil eder. Bu sınıf, iletilerin birden çok kaynak tarafından yazılabilen veya birden çok hedef tarafından okunabilen bir ilk giren ilk çıkar (FIFO) sırasını tutar. Bir hedef bir BufferBlock<T> nesnesinden ileti aldığında, o ileti, ileti sırasından kaldırılır. Bu nedenle, bir BufferBlock<T> nesnesi birden çok hedefe sahip olabilse de her iletiyi yalnızca bir hedef alır. BufferBlock<T> sınıfı, başka bir bileşene birden çok ileti geçirmek istediğinizde ve bu bileşenin her iletiyi alması gerektiğinde kullanışlıdır.
Aşağıdaki temel örnek, birInt32 nesnesine birkaç BufferBlock<T> değeri gönderir ve o değerleri bu nesneden geri okur.
// Create a BufferBlock<int> object.
var bufferBlock = new BufferBlock<int>();
// Post several messages to the block.
for (int i = 0; i < 3; i++)
{
bufferBlock.Post(i);
}
// Receive the messages back from the block.
for (int i = 0; i < 3; i++)
{
Console.WriteLine(bufferBlock.Receive());
}
/* Output:
0
1
2
*/
' Create a BufferBlock<int> object.
Dim bufferBlock = New BufferBlock(Of Integer)()
' Post several messages to the block.
For i As Integer = 0 To 2
bufferBlock.Post(i)
Next i
' Receive the messages back from the block.
For i As Integer = 0 To 2
Console.WriteLine(bufferBlock.Receive())
Next i
' Output:
' 0
' 1
' 2
'
Bir nesneye ileti yazmayı ve nesneden iletileri okumayı gösteren eksiksiz bir BufferBlock<T> örnek için bkz . Nasıl yapılır: Veri Akışı Bloğuna İleti Yazma ve İletileri Okuma.
BroadcastBlock<T>
BroadcastBlock<T> sınıfı, başka bir bileşene birden çok ileti geçirmeniz ancak bu bileşenin yalnızca en son değeri okuması gerektiğinde kullanışlıdır. Bu sınıf, bir iletiyi birden çok bileşene yayınlamak istediğinizde de kullanışlıdır.
Aşağıdaki temel örnek, bir Double nesnesine bir BroadcastBlock<T> değeri gönderir ve değeri o nesneden birkaç kez geri okur. BroadcastBlock<T> nesnelerindeki değerler okunduktan sonra kaldırılmadığından, her seferinde aynı değer okunabilir.
// Create a BroadcastBlock<double> object.
var broadcastBlock = new BroadcastBlock<double>(null);
// Post a message to the block.
broadcastBlock.Post(Math.PI);
// Receive the messages back from the block several times.
for (int i = 0; i < 3; i++)
{
Console.WriteLine(broadcastBlock.Receive());
}
/* Output:
3.14159265358979
3.14159265358979
3.14159265358979
*/
' Create a BroadcastBlock<double> object.
Dim broadcastBlock = New BroadcastBlock(Of Double)(Nothing)
' Post a message to the block.
broadcastBlock.Post(Math.PI)
' Receive the messages back from the block several times.
For i As Integer = 0 To 2
Console.WriteLine(broadcastBlock.Receive())
Next i
' Output:
' 3.14159265358979
' 3.14159265358979
' 3.14159265358979
'
İletiyi birden çok hedef bloğuna yayınlamak için nasıl kullanılacağını BroadcastBlock<T> gösteren eksiksiz bir örnek için bkz . Nasıl yapılır: Veri Akışı Bloğunda Görev Zamanlayıcı Belirtme.
WriteOnceBlock<T>
WriteOnceBlock<T> sınıfı, BroadcastBlock<T> nesnesine yalnızca bir kez yazılabilmesi dışında WriteOnceBlock<T> sınıfına benzer. C# readonly (Visual Basic'te ReadOnly) anahtar sözcüğüne benzer olduğunu düşünebilirsinizWriteOnceBlock<T>, ancak nesne WriteOnceBlock<T> oluşturma sırasında değil değer aldıktan sonra sabit hale gelir. BroadcastBlock<T> sınıfı gibi, hedef de bir WriteOnceBlock<T> nesnesinden ileti aldığında, o ileti nesneden kaldırılmaz. Bu nedenle, birden çok hedef iletinin bir kopyasını alır. WriteOnceBlock<T> sınıfı, birden çok iletinin yalnızca ilkini yaymak istediğinizde kullanışlıdır.
Aşağıdaki temel örnek, bir String nesnesine birden çok WriteOnceBlock<T> değeri gönderir ve nesneden değeri geri okur. Bir WriteOnceBlock<T> nesnesine yalnızca bir kez yazılabildiğinden, WriteOnceBlock<T> nesnesi bir ileti aldıktan sonra, onun iletinin ardından gelen iletileri atar.
// Create a WriteOnceBlock<string> object.
var writeOnceBlock = new WriteOnceBlock<string>(null);
// Post several messages to the block in parallel. The first
// message to be received is written to the block.
// Subsequent messages are discarded.
Parallel.Invoke(
() => writeOnceBlock.Post("Message 1"),
() => writeOnceBlock.Post("Message 2"),
() => writeOnceBlock.Post("Message 3"));
// Receive the message from the block.
Console.WriteLine(writeOnceBlock.Receive());
/* Sample output:
Message 2
*/
' Create a WriteOnceBlock<string> object.
Dim writeOnceBlock = New WriteOnceBlock(Of String)(Nothing)
' Post several messages to the block in parallel. The first
' message to be received is written to the block.
' Subsequent messages are discarded.
Parallel.Invoke(Function() writeOnceBlock.Post("Message 1"), Function() writeOnceBlock.Post("Message 2"), Function() writeOnceBlock.Post("Message 3"))
' Receive the message from the block.
Console.WriteLine(writeOnceBlock.Receive())
' Sample output:
' Message 2
'
Tamamlanan ilk işlemin değerini almak için nasıl kullanılacağını WriteOnceBlock<T> gösteren eksiksiz bir örnek için bkz . Nasıl yapılır: Veri Akışı Bloklarının Bağlantısını Kaldırma.
Yürütme Blokları
Yürütme blokları, alınan her veri parçası için kullanıcı tarafından sağlanan bir temsilciyi çağırır. TPL Veri Akışı Kitaplığı, üç yürütme bloğu türü sağlar: ActionBlock<TInput>, System.Threading.Tasks.Dataflow.TransformBlock<TInput,TOutput> ve System.Threading.Tasks.Dataflow.TransformManyBlock<TInput,TOutput>.
ActionBlock<T>
ActionBlock<TInput> sınıfı, veri aldığında bir temsilci çağıran bir hedef bloktur. Bir ActionBlock<TInput> nesnesini, veri kullanılabilir olduğunda zaman uyumsuz olarak çalışan bir temsilci gibi düşünün. Bir ActionBlock<TInput> nesnesine sağladığınız temsilcinin türü, Action<T> veya System.Func<TInput, Task> olabilir. Bir ActionBlock<TInput> nesnesini Action<T> ile kullandığınızda, her giriş öğesinin işlenmesi temsilci döndüğünde tamamlanmış sayılır. Bir ActionBlock<TInput> nesnesini System.Func<TInput, Task> ile kullandığınızda, her giriş öğesinin işlenmesi ancak döndürülen Task nesnesi tamamlandığında tamamlanmış sayılır. Bu iki mekanizmayı kullanarak, ActionBlock<TInput> sınıfını her giriş öğesinin hem zaman uyumlu hem de zaman uyumsuz işlenmesi için kullanabilirsiniz.
Aşağıdaki temel örnek, bir Int32 nesnesine birden çok ActionBlock<TInput> değeri gönderir. ActionBlock<TInput> nesnesi, bu değerleri konsola yazdırır. Ardından, bu örnek, bloğu tamamlanmış duruma ayarlar ve tüm veri akışı görevlerinin bitmesini bekler.
// Create an ActionBlock<int> object that prints values
// to the console.
var actionBlock = new ActionBlock<int>(n => Console.WriteLine(n));
// Post several messages to the block.
for (int i = 0; i < 3; i++)
{
actionBlock.Post(i * 10);
}
// Set the block to the completed state and wait for all
// tasks to finish.
actionBlock.Complete();
actionBlock.Completion.Wait();
/* Output:
0
10
20
*/
' Create an ActionBlock<int> object that prints values
' to the console.
Dim actionBlock = New ActionBlock(Of Integer)(Function(n) WriteLine(n))
' Post several messages to the block.
For i As Integer = 0 To 2
actionBlock.Post(i * 10)
Next i
' Set the block to the completed state and wait for all
' tasks to finish.
actionBlock.Complete()
actionBlock.Completion.Wait()
' Output:
' 0
' 10
' 20
'
Sınıfıyla ActionBlock<TInput> temsilcilerin nasıl kullanılacağını gösteren eksiksiz örnekler için bkz . Nasıl yapılır: Veri Akışı Bloğu Veri Aldığında Eylem Gerçekleştirme.
TransformBlock<TInput, TOutput>
TransformBlock<TInput,TOutput> sınıfı, hem bir kaynak hem de bir hedef olarak davranması dışında ActionBlock<TInput> sınıfına benzer. Bir TransformBlock<TInput,TOutput> nesnesine geçirdiğiniz temsilci TOutput türünde bir değer döndürür. Bir TransformBlock<TInput,TOutput> nesnesine sağladığınız temsilcinin türü, System.Func<TInput, TOutput> veya System.Func<TInput, Task<TOutput>> olabilir. Bir TransformBlock<TInput,TOutput> nesnesini System.Func<TInput, TOutput> ile kullandığınızda, her giriş öğesinin işlenmesi temsilci döndüğünde tamamlanmış sayılır. TransformBlock<TInput,TOutput> ile kullanılan bir System.Func<TInput, Task<TOutput>> nesnesini kullandığınızda, her giriş öğesinin işlenmesi ancak döndürülen Task<TResult> nesnesi tamamlandığında tamamlanmış sayılır. ActionBlock<TInput> sınıfında olduğu gibi, bu iki mekanizmayı kullanarak TransformBlock<TInput,TOutput> sınıfını her giriş öğesinin hem zaman uyumlu hem de zaman uyumsuz işlenmesi için kullanabilirsiniz.
Aşağıdaki temel örnek, giriş değerinin kare kökünü hesaplayan bir TransformBlock<TInput,TOutput> nesnesi oluşturur. TransformBlock<TInput,TOutput> nesnesi, giriş olarak Int32 değerlerini alır ve çıktı olarak Double değerlerini üretir.
// Create a TransformBlock<int, double> object that
// computes the square root of its input.
var transformBlock = new TransformBlock<int, double>(n => Math.Sqrt(n));
// Post several messages to the block.
transformBlock.Post(10);
transformBlock.Post(20);
transformBlock.Post(30);
// Read the output messages from the block.
for (int i = 0; i < 3; i++)
{
Console.WriteLine(transformBlock.Receive());
}
/* Output:
3.16227766016838
4.47213595499958
5.47722557505166
*/
' Create a TransformBlock<int, double> object that
' computes the square root of its input.
Dim transformBlock = New TransformBlock(Of Integer, Double)(Function(n) Math.Sqrt(n))
' Post several messages to the block.
transformBlock.Post(10)
transformBlock.Post(20)
transformBlock.Post(30)
' Read the output messages from the block.
For i As Integer = 0 To 2
Console.WriteLine(transformBlock.Receive())
Next i
' Output:
' 3.16227766016838
' 4.47213595499958
' 5.47722557505166
'
Windows Forms uygulamasında görüntü işleme gerçekleştiren bir veri akışı blokları ağında kullanılan TransformBlock<TInput,TOutput> tam örnekler için bkz . İzlenecek Yol: Windows Forms Uygulamasında Veri Akışını Kullanma.
TransformManyBlock<TInput, TOutput>
TransformManyBlock<TInput,TOutput> sınıfı, TransformBlock<TInput,TOutput> sınıfına benzer, ancak TransformManyBlock<TInput,TOutput> sınıfı her giriş değeri için tek bir çıktı değeri üretmek yerine her giriş değeri için sıfır veya daha fazla çıktı değeri üretir. Bir TransformManyBlock<TInput,TOutput> nesnesine sağladığınız temsilcinin türü, System.Func<TInput, IEnumerable<TOutput>> veya System.Func<TInput, Task<IEnumerable<TOutput>>> olabilir. Bir TransformManyBlock<TInput,TOutput> nesnesini System.Func<TInput, IEnumerable<TOutput>> ile kullandığınızda, her giriş öğesinin işlenmesi temsilci döndüğünde tamamlanmış sayılır. Bir TransformManyBlock<TInput,TOutput> nesnesini System.Func<TInput, Task<IEnumerable<TOutput>>> ile kullandığınızda, her giriş öğesinin işlenmesi ancak döndürülen System.Threading.Tasks.Task<IEnumerable<TOutput>> nesnesi tamamlandığında tamamlanmış sayılır.
Aşağıdaki temel örnek, dizeleri karakter sıralarına ayıran bir TransformManyBlock<TInput,TOutput> nesnesi oluşturur. TransformManyBlock<TInput,TOutput> nesnesi, giriş olarak String değerlerini alır ve çıktı olarak Char değerlerini üretir.
// Create a TransformManyBlock<string, char> object that splits
// a string into its individual characters.
var transformManyBlock = new TransformManyBlock<string, char>(
s => s.ToCharArray());
// Post two messages to the first block.
transformManyBlock.Post("Hello");
transformManyBlock.Post("World");
// Receive all output values from the block.
for (int i = 0; i < ("Hello" + "World").Length; i++)
{
Console.WriteLine(transformManyBlock.Receive());
}
/* Output:
H
e
l
l
o
W
o
r
l
d
*/
' Create a TransformManyBlock<string, char> object that splits
' a string into its individual characters.
Dim transformManyBlock = New TransformManyBlock(Of String, Char)(Function(s) s.ToCharArray())
' Post two messages to the first block.
transformManyBlock.Post("Hello")
transformManyBlock.Post("World")
' Receive all output values from the block.
For i As Integer = 0 To ("Hello" & "World").Length - 1
Console.WriteLine(transformManyBlock.Receive())
Next i
' Output:
' H
' e
' l
' l
' o
' W
' o
' r
' l
' d
'
Bir veri akışı işlem hattındaki her giriş için birden çok bağımsız çıkış üretmek için kullanılan TransformManyBlock<TInput,TOutput> tam örnekler için bkz . İzlenecek Yol: Veri Akışı İşlem Hattı Oluşturma.
Paralellik Derecesi
Her ActionBlock<TInput>, TransformBlock<TInput,TOutput> ve TransformManyBlock<TInput,TOutput> nesnesi, blok onları işlemeye hazır olana kadar giriş iletilerini arabelleğe alır. Varsayılan olarak, bu sınıflar iletileri alındıkları sırada, teker teker işler. ActionBlock<TInput>, TransformBlock<TInput,TOutput> ve TransformManyBlock<TInput,TOutput> nesnelerinin aynı anda birden çok ileti işlemesini mümkün kılmak için paralellik derecesini de belirtebilirsiniz. Eş zamanlı yürütme hakkında daha fazla bilgi edinmek için, bu belgenin ilerleyen bölümlerindeki Paralellik Derecesini Belirtme bölümüne bakın. Yürütme veri akışı bloğunun aynı anda birden fazla iletiyi işlemesini sağlamak için paralellik derecesini ayarlayan bir örnek için bkz . Nasıl yapılır: Veri Akışı Bloğunda Paralellik Derecesini Belirtme.
Temsilci Türlerinin Özeti
Aşağıdaki tabloda ActionBlock<TInput>, TransformBlock<TInput,TOutput> ve TransformManyBlock<TInput,TOutput> nesnelerine sağlayabileceğiniz temsilci türleri özetlenmektedir. Bu tabloda, temsilci türünün zaman uyumlu mu yoksa zaman uyumsuz mu çalıştığı da belirtilir.
| Tür | Zaman Uyumlu Temsilci Türü | Zaman Uyumsuz Temsilci Türü |
|---|---|---|
| ActionBlock<TInput> | System.Action |
System.Func<TInput, Task> |
| TransformBlock<TInput,TOutput> | System.Func<TInput, TOutput> |
System.Func<TInput, Task<TOutput>> |
| TransformManyBlock<TInput,TOutput> | System.Func<TInput, IEnumerable<TOutput>> |
System.Func<TInput, Task<IEnumerable<TOutput>>> |
Yürütme bloğu türleriyle çalışırken lambda ifadeleri de kullanabilirsiniz. Yürütme bloğuyla lambda ifadesinin nasıl kullanılacağını gösteren bir örnek için bkz . Nasıl yapılır: Veri Akışı Bloğu Veri Aldığında Eylem Gerçekleştirme.
Gruplandırma Blokları
Gruplandırma blokları, bir veya birden fazla kaynaktaki verileri çeşitli kısıtlamalar altında birleştirir. TPL Veri Akışı Kitaplığı, üç birleştirme bloğu türü sağlar: BatchBlock<T>, JoinBlock<T1,T2> ve BatchedJoinBlock<T1,T2>.
BatchBlock<T>
BatchBlock<T> sınıfı, toplu iş olarak bilinen giriş veri kümelerini birleştirerek çıktı veri dizileri oluşturur. Bir BatchBlock<T> nesnesi oluşturduğunuzda her toplu işin boyutunu siz belirtirsiniz. BatchBlock<T> nesnesi belirtilen sayıda giriş öğesi aldığında, zaman uyumsuz olarak bu öğeleri içeren bir dizi yayar. Eğer bir BatchBlock<T> nesnesi tamamlanmış duruma ayarlanmış olmasına rağmen toplu iş oluşturmak için yeterli öğeye sahip değilse, kalan giriş öğelerini içeren son bir dizi yayınlar.
BatchBlock<T> sınıfı doyumsuz veya doyumsuz olmayan modda çalışır. Varsayılan olan doyumsuz modda, bir BatchBlock<T> nesnesi sunulan her iletiyi kabul eder ve belirtilen sayıda öğeyi kabul ettikten sonra bir dizi yayınlar. Doyumsuz olmayan modda, bir BatchBlock<T> öğesi, yeterli sayıda kaynak tarafından bir toplu iş oluşturacak kadar ileti sunulana kadar tüm iletileri erteler. Doyumsuz mod genellikle doyumsuz olmayan moddan daha iyi çalışır, çünkü daha az işlem ek yükü gerektirir. Ancak, birden çok kaynaktan yapılan tüketimi atomik bir düzeyde koordine etmeniz gerektiğinde, doyumsuz olmayan modu kullanabilirsiniz. Doyumsuz olmayan modu belirtmek için Greedy oluşturucusundaki False parametresinin dataflowBlockOptions özelliğini BatchBlock<T> olarak ayarlayın.
Aşağıdaki temel örnek, bir toplu işte on öğe tutan bir Int32 nesnesine birden çok BatchBlock<T> değeri gönderir. BatchBlock<T> nesnesindeki tüm değerlerin yayınlandığından emin olmak için, bu örnek Complete yöntemini çağırır. Complete yöntemi, BatchBlock<T> nesnesinin tamamlanmış duruma ayarlar ve bu yüzden, BatchBlock<T> nesnesi kalan öğeleri son bir toplu iş olarak yayınlar.
// Create a BatchBlock<int> object that holds ten
// elements per batch.
var batchBlock = new BatchBlock<int>(10);
// Post several values to the block.
for (int i = 0; i < 13; i++)
{
batchBlock.Post(i);
}
// Set the block to the completed state. This causes
// the block to propagate out any remaining
// values as a final batch.
batchBlock.Complete();
// Print the sum of both batches.
Console.WriteLine("The sum of the elements in batch 1 is {0}.",
batchBlock.Receive().Sum());
Console.WriteLine("The sum of the elements in batch 2 is {0}.",
batchBlock.Receive().Sum());
/* Output:
The sum of the elements in batch 1 is 45.
The sum of the elements in batch 2 is 33.
*/
' Create a BatchBlock<int> object that holds ten
' elements per batch.
Dim batchBlock = New BatchBlock(Of Integer)(10)
' Post several values to the block.
For i As Integer = 0 To 12
batchBlock.Post(i)
Next i
' Set the block to the completed state. This causes
' the block to propagate out any remaining
' values as a final batch.
batchBlock.Complete()
' Print the sum of both batches.
Console.WriteLine("The sum of the elements in batch 1 is {0}.", batchBlock.Receive().Sum())
Console.WriteLine("The sum of the elements in batch 2 is {0}.", batchBlock.Receive().Sum())
' Output:
' The sum of the elements in batch 1 is 45.
' The sum of the elements in batch 2 is 33.
'
Veritabanı ekleme işlemlerinin verimliliğini artırmak için kullanan BatchBlock<T> eksiksiz bir örnek için bkz . İzlenecek Yol: Verimliliği Artırmak için BatchBlock ve BatchedJoinBlock Kullanma.
JoinBlock<T1, T2, ...>
JoinBlock<T1,T2> ve JoinBlock<T1,T2,T3> sınıfları, giriş öğelerini toplar ve bu öğeleri içeren System.Tuple<T1,T2> veya System.Tuple<T1,T2,T3> nesnelerini yayınlar. JoinBlock<T1,T2> ve JoinBlock<T1,T2,T3> sınıfları, ITargetBlock<TInput> öğesinden devralmaz. Bunun yerine, Target1 öğesini uygulayan Target2, Target3 ve ITargetBlock<TInput> özelliklerini sağlarlar.
BatchBlock<T> gibi, JoinBlock<T1,T2> ve JoinBlock<T1,T2,T3> öğeleri de doyumsuz veya doyumsuz olmayan modda çalışırlar. Varsayılan olan doyumsuz modda, bir JoinBlock<T1,T2> veya JoinBlock<T1,T2,T3> nesnesi, sunulan tüm iletileri kabul eder ve hedeflerinden her biri en az bir ileti aldıktan sonra bir kayıt düzeni yayınlar. Doyumsuz olmayan modda, bir JoinBlock<T1,T2> veya JoinBlock<T1,T2,T3> nesnesi tüm hedeflere kayıt düzeni oluşturmak için gereken veriler sunulana kadar tüm iletileri erteler. Bu noktada, blok, kaynaklardan gereken tüm öğeleri atomik olarak almak için iki aşamalı bir kaydetme protokolü uygular. Bu erteleme, o esnada başka bir varlığın veriyi tüketmesini mümkün kılarak tüm sistemin ilerlemesini sağlar.
Aşağıdaki temel örnek, bir JoinBlock<T1,T2,T3> nesnesinin bir değeri hesaplamak için birden çok veriye ihtiyaç duyduğu bir durumu gösterir. Bu örnek, bir aritmetik işlemi gerçekleştirmek için iki JoinBlock<T1,T2,T3> değeri ve bir Int32 değeri gerektiren bir Char nesnesi oluşturur.
// Create a JoinBlock<int, int, char> object that requires
// two numbers and an operator.
var joinBlock = new JoinBlock<int, int, char>();
// Post two values to each target of the join.
joinBlock.Target1.Post(3);
joinBlock.Target1.Post(6);
joinBlock.Target2.Post(5);
joinBlock.Target2.Post(4);
joinBlock.Target3.Post('+');
joinBlock.Target3.Post('-');
// Receive each group of values and apply the operator part
// to the number parts.
for (int i = 0; i < 2; i++)
{
var data = joinBlock.Receive();
switch (data.Item3)
{
case '+':
Console.WriteLine("{0} + {1} = {2}",
data.Item1, data.Item2, data.Item1 + data.Item2);
break;
case '-':
Console.WriteLine("{0} - {1} = {2}",
data.Item1, data.Item2, data.Item1 - data.Item2);
break;
default:
Console.WriteLine("Unknown operator '{0}'.", data.Item3);
break;
}
}
/* Output:
3 + 5 = 8
6 - 4 = 2
*/
' Create a JoinBlock<int, int, char> object that requires
' two numbers and an operator.
Dim joinBlock = New JoinBlock(Of Integer, Integer, Char)()
' Post two values to each target of the join.
joinBlock.Target1.Post(3)
joinBlock.Target1.Post(6)
joinBlock.Target2.Post(5)
joinBlock.Target2.Post(4)
joinBlock.Target3.Post("+"c)
joinBlock.Target3.Post("-"c)
' Receive each group of values and apply the operator part
' to the number parts.
For i As Integer = 0 To 1
Dim data = joinBlock.Receive()
Select Case data.Item3
Case "+"c
Console.WriteLine("{0} + {1} = {2}", data.Item1, data.Item2, data.Item1 + data.Item2)
Case "-"c
Console.WriteLine("{0} - {1} = {2}", data.Item1, data.Item2, data.Item1 - data.Item2)
Case Else
Console.WriteLine("Unknown operator '{0}'.", data.Item3)
End Select
Next i
' Output:
' 3 + 5 = 8
' 6 - 4 = 2
'
Doyumsuz olmayan modda bir kaynağı işbirliğiyle paylaşmak için nesneleri kullanan JoinBlock<T1,T2> eksiksiz bir örnek için bkz . Nasıl yapılır: Birden Çok Kaynaktan Veri Okumak için JoinBlock Kullanma.
BatchedJoinBlock<T1, T2, ...>
BatchedJoinBlock<T1,T2> ve BatchedJoinBlock<T1,T2,T3> sınıfları, giriş öğelerini toplu bir halde toplar ve bu öğeleri içeren System.Tuple(IList(T1), IList(T2)) veya System.Tuple(IList(T1), IList(T2), IList(T3)) nesneleri yayınlar. BatchedJoinBlock<T1,T2> öğesini BatchBlock<T> ve JoinBlock<T1,T2> öğelerinin birleşimi olarak düşünebilirsiniz. Bir BatchedJoinBlock<T1,T2> nesnesi oluşturduğunuzda her toplu işin boyutunu belirtin. Ayrıca, BatchedJoinBlock<T1,T2>Target1 öğesini uygulayan Target2 ve ITargetBlock<TInput> özelliklerini de sağlar. Tüm hedeflerden belirtilen sayıda giriş öğesi alındığında, BatchedJoinBlock<T1,T2> nesnesi zaman uyumsuz olarak bu öğeleri içeren bir System.Tuple(IList(T1), IList(T2)) nesnesi yayar.
Aşağıdaki temel örnek, sonuçları, BatchedJoinBlock<T1,T2> değerlerini ve birer Int32 nesnesi olanları tutan bir Exception nesnesi oluşturur. Bu örnek birden çok işlem gerçekleştirir ve sonuçları Target1 nesnesinin Target2 özelliğine, hataları ise BatchedJoinBlock<T1,T2> özelliğine yazar. Başarılı ve başarısız işlemlerin sayısı önceden bilinmediği için, IList<T> nesneleri her hedefin sıfır veya daha fazla değer alabilmesini sağlar.
// For demonstration, create a Func<int, int> that
// returns its argument, or throws ArgumentOutOfRangeException
// if the argument is less than zero.
Func<int, int> DoWork = n =>
{
if (n < 0)
throw new ArgumentOutOfRangeException();
return n;
};
// Create a BatchedJoinBlock<int, Exception> object that holds
// seven elements per batch.
var batchedJoinBlock = new BatchedJoinBlock<int, Exception>(7);
// Post several items to the block.
foreach (int i in new int[] { 5, 6, -7, -22, 13, 55, 0 })
{
try
{
// Post the result of the worker to the
// first target of the block.
batchedJoinBlock.Target1.Post(DoWork(i));
}
catch (ArgumentOutOfRangeException e)
{
// If an error occurred, post the Exception to the
// second target of the block.
batchedJoinBlock.Target2.Post(e);
}
}
// Read the results from the block.
var results = batchedJoinBlock.Receive();
// Print the results to the console.
// Print the results.
foreach (int n in results.Item1)
{
Console.WriteLine(n);
}
// Print failures.
foreach (Exception e in results.Item2)
{
Console.WriteLine(e.Message);
}
/* Output:
5
6
13
55
0
Specified argument was out of the range of valid values.
Specified argument was out of the range of valid values.
*/
' For demonstration, create a Func<int, int> that
' returns its argument, or throws ArgumentOutOfRangeException
' if the argument is less than zero.
Dim DoWork As Func(Of Integer, Integer) = Function(n)
If n < 0 Then
Throw New ArgumentOutOfRangeException()
End If
Return n
End Function
' Create a BatchedJoinBlock<int, Exception> object that holds
' seven elements per batch.
Dim batchedJoinBlock = New BatchedJoinBlock(Of Integer, Exception)(7)
' Post several items to the block.
For Each i As Integer In New Integer() {5, 6, -7, -22, 13, 55, 0}
Try
' Post the result of the worker to the
' first target of the block.
batchedJoinBlock.Target1.Post(DoWork(i))
Catch e As ArgumentOutOfRangeException
' If an error occurred, post the Exception to the
' second target of the block.
batchedJoinBlock.Target2.Post(e)
End Try
Next i
' Read the results from the block.
Dim results = batchedJoinBlock.Receive()
' Print the results to the console.
' Print the results.
For Each n As Integer In results.Item1
Console.WriteLine(n)
Next n
' Print failures.
For Each e As Exception In results.Item2
Console.WriteLine(e.Message)
Next e
' Output:
' 5
' 6
' 13
' 55
' 0
' Specified argument was out of the range of valid values.
' Specified argument was out of the range of valid values.
'
Program bir veritabanından okurken hem sonuçları hem de oluşan özel durumları yakalamak için kullanan BatchedJoinBlock<T1,T2> eksiksiz bir örnek için bkz . Verimliliği Artırmak için İzlenecek Yol: BatchBlock ve BatchedJoinBlock Kullanma.
Veri Akışı Bloğu Davranışını Yapılandırma
Veri akışı bloğu türlerinin oluşturucusuna bir System.Threading.Tasks.Dataflow.DataflowBlockOptions sağlayarak ek seçenekleri etkinleştirebilirsiniz. Bu seçenekler, arkaplandaki görevi yöneten zamanlayıcı ve paralellik derecesi gibi davranışları denetler. DataflowBlockOptions, belirli veri akışı bloğu türlerine özel davranışları belirtmek için devralınmış özelliklere de sahiptir. Aşağıdaki tabloda, her veri akışı bloğu türüyle hangi seçenek türünün ilişkili olduğunu özetlenmektedir.
Sonraki bölümlerde, System.Threading.Tasks.Dataflow.DataflowBlockOptions, System.Threading.Tasks.Dataflow.ExecutionDataflowBlockOptions ve System.Threading.Tasks.Dataflow.GroupingDataflowBlockOptions sınıfları aracılığıyla kullanılabilen önemli veri akışı bloğu seçenekleri türleri hakkında ek bilgiler sağlar.
Görev Zamanlayıcısı'ı Belirtme
Her önceden tanımlı veri akışı bloğu, veriyi bir hedefe yayma, bir kaynaktan veri alma ve veri kullanılabilir oldukça kullanıcı tanımlı temsilcileri çalıştırma gibi işlemleri gerçekleştirmek için TPL görev zamanlama mekanizmasını kullanır. TaskScheduler, iş parçacıklarının görevlerini sıralayan görev zamanlayıcısını temsil eden bir soyut sınıftır. Varsayılan görev zamanlayıcısı, Default, işleri sıralamak ve yürütmek için ThreadPool sınıfını kullanır. Bir veri akışı bloğu nesnesi oluştururken TaskScheduler özelliğini ayarlayarak varsayılan görev zamanlayıcısını geçersiz kılabilirsiniz.
Görev zamanlayıcısı birden çok veri akışı bloğunu yönettiğinde, aralarında ilkeler uygulayabilir. Örneğin, birden çok veri akışı bloğunun her biri aynı ConcurrentExclusiveSchedulerPair nesnesinin özel zamanlayıcısını hedef alacak şekilde yapılandırılırsa, bu bloklarda çalışan tüm işler seri hale getirilir. Benzer şekilde, bu bloklar aynı ConcurrentExclusiveSchedulerPair nesnesinin eşzamanlı zamanlayıcısını hedef alacak şekilde, bu zamanlayıcı da en fazla eşzamanlılık düzeyine sahip olacak şekilde yapılandırılırsa, bu bloklardaki tüm işler o eşzamanlı işlem sayısıyla sınırlandırılır. Okuma işlemlerinin ConcurrentExclusiveSchedulerPair paralel olarak gerçekleşmesini, ancak yazma işlemlerinin yalnızca diğer tüm işlemlerden oluşmasını sağlamak için sınıfını kullanan bir örnek için bkz . Nasıl yapılır: Veri Akışı Bloğunda Görev Zamanlayıcı Belirtme. TPL'deki görev zamanlayıcıları hakkında daha fazla bilgi için sınıf konusuna TaskScheduler bakın.
Paralellik Derecesini Belirtme
Varsayılan olarak, TPL Veri Akışı Kitaplığının sağladığı üç yürütme bloğu türü olan ActionBlock<TInput>, TransformBlock<TInput,TOutput> ve TransformManyBlock<TInput,TOutput>, iletileri tek tek işler. Ayrıca, bu veri akışı blokları iletileri alındıkları sırada işler. Bu veri akışı bloklarının iletileri eşzamanlı olarak işlemesini mümkün kılmak için, veri akışı bloğu nesnesini oluştururken ExecutionDataflowBlockOptions.MaxDegreeOfParallelism özelliğini ayarlayın.
Varsayılan MaxDegreeOfParallelism değeri 1'dir ve tüm veri akışı boklarının aynı anda tek bir ileti işlemesini sağlar. Bu özelliği 1'den daha büyük bir değere ayarlayarak veri akışı bloğunun birden çok iletiyi eşzamanlı olarak işlemesini sağlayabilirsiniz. Bu özelliği DataflowBlockOptions.Unbounded olarak ayarlamak, temelde yatan görev zamanlayıcısının en büyük eşzamanlılık derecesini yönetmesini sağlar.
Önemli
En fazla 1'den büyük bir paralellik derecesi belirttiğinizde, birden çok ileti aynı anda işlenir ve bu nedenle iletiler alındıkları sırayla işlenmeyebilir. Ancak, iletilerin bloktan çıkış sırası, alındıkları sıradır.
MaxDegreeOfParallelism özelliği en büyük paralellik derecesini temsil ettiğinden, veri akışı bloğu belirttiğinizden daha düşük derecede paralellikle çalışabilir. Veri akışı bloğu işlevsel gereksinimlerini karşılamak için veya yeterli kullanılabilir sistem kaynağı olmaması nedeniyle daha düşük bir paralellik derecesi kullanabilir Bir veri akışı bloğu asla belirttiğinizden daha fazla paralellik seçmez.
MaxDegreeOfParallelism özelliğinin değeri, her veri akışı bloğu nesnesi için o nesneye özeldir. Örneğin, dört veri akışı bloğu nesnesinin her biri en büyük paralellik derecesi için 1 belirtirse, dört veri akışı bloğu nesnesinin hepsi paralel çalışma potansiyeline sahiptir.
Uzun işlemlerin paralel gerçekleşmesini sağlamak için en yüksek paralellik derecesini ayarlayan bir örnek için bkz . Nasıl yapılır: Veri Akışı Bloğunda Paralellik Derecesini Belirtme.
Görev Başına İleti Sayısını Belirtme
Önceden tanımlı veri akışı bloğu türleri, görevleri kullanarak birden çok giriş öğesini işler. Bu veriyi işlemek için gereken görev nesnesi sayısını az sayıda tutmaya yardımcı olur ve bu sayede, uygulamaların daha verimli çalışmasını sağlanır. Ancak, bir veri akışı bloğu kümesindeki görevler veri işlerken, başka veri akışı bloklarındaki görevlerin iletileri sıraya alarak işlemci zamanı için beklemesi gerekebilir. Veri akışı görevleri arasında daha iyi eşitlik sağlamak için, MaxMessagesPerTask özelliğini ayarlayın. MaxMessagesPerTask özelliği varsayılan değer olan DataflowBlockOptions.Unbounded değerine ayarlandığında, bir veri akışı bloğu tarafından kullanılan görev, mevcut iletilerin hepsini işler. MaxMessagesPerTask özelliği Unbounded dışında bir değere ayarlandığında, veri akışı bloğu Task nesnesi başına en fazla o sayı kadar ileti işler. MaxMessagesPerTask özelliğini ayarlamak görevler arası eşitliği artırsa da, sistemin gerektiğinden fazla görev oluşturmasına sebep olup performansı azaltabilir.
İptal Etkinleştirme
TPL, görevlerin karşılıklı olarak iptal işlemini koordine etmelerine olanak sağlayan bir mekanizma sunar. Veri akışı bloklarının bu iptal mekanizmasına katılımını etkinleştirmek için CancellationToken özelliğini ayarlayın. Bu CancellationToken nesnesi iptal edilmiş duruma ayarlandığında, bu belirteci izleyen tüm veri akışı blokları geçerli öğelerini yürütmeyi bitirir ama sonraki öğeleri işlemeye başlamaz. Bu veri akışı blokları, ayrıca, arabelleğe alınmış tüm iletileri temizler, kaynak ve hedef bloklara olan bağlantılarını bırakır ve iptal edilmiş duruma geçer. Completion özelliği iptal edilmiş duruma geçerek, işlem sırasında bir özel durum oluşmadığı sürece, Status özelliğinin Canceled olarak ayarlanmasını sağlar. Böyle bir durum oluştuysa, Status özelliği Faulted olarak ayarlanır.
Windows Forms uygulamasında iptalin nasıl kullanılacağını gösteren bir örnek için bkz . Nasıl yapılır: Veri Akışı Bloğunu İptal Etme. TPL'de iptal hakkında daha fazla bilgi için bkz . Görev İptali.
Doyumsuz veya Doyumsuz Olmayan Davranış Belirtme
Çeşitli gruplama veri akışı bloğu türleri doyumsuz veya doyumsuz olmayan modda çalışabilir. Varsayılan olarak, önceden tanımlı veri akışı bloğu türleri doyumsuz modda çalışır.
JoinBlock<T1,T2> gibi birleştirme bloğu türleri için, doyumsuz mod, blok tarafından verilerin kendilerine karşılık gelen birleştirilecek veriler mevcut olmasa bile hemen kabul edildiği anlamına gelir. Doyumsuz olmayan mod ise, bloğun her hedefinde birleştirme işlemini tamamlayacak bir ileti olana kadar gelen tüm iletileri ertelemesi anlamına gelir. Ertelenen iletilerden herhangi biri artık kullanılabilir değilse, birleştirme bloğu ertelenen tüm iletileri bırakır ve işlemi yeniden başlatır. BatchBlock<T> sınıfı için doyumsuz ve doyumsuz olmayan davranış benzerdir, ancak doyumsuz olmayan modda bir BatchBlock<T> nesnesi, ayrı kaynaklardan bir toplu işi tamamlayacak sayıda ileti gelene kadar tüm gelen iletileri erteler.
Bir veri akışı bloğunda doyumsuz olmayan modu ayarlamak için, Greedy özelliğini False olarak ayarlayın. Bir veri kaynağını daha verimli bir şekilde paylaşmak üzere birden çok birleştirme bloğunu etkinleştirmek için doyumsuz olmayan modun nasıl kullanılacağını gösteren bir örnek için bkz . Nasıl yapılır: Birden Çok Kaynaktan Veri Okumak için JoinBlock Kullanma.
Özel Veri Akışı Blokları
TPL Veri Akışı Kitaplığı pek çok önceden tanımlı blok türü sunsa da özel davranışları olan ek blok türleri oluşturabilirsiniz. ISourceBlock<TOutput> veya ITargetBlock<TInput> arabirimlerini doğrudan uygulayın veya yöntemini kullanarak Encapsulate mevcut blok türlerinin davranışını kapsülleyen karmaşık bir blok oluşturun. Özel veri akışı bloğu işlevselliğinin nasıl uygulandığını gösteren örnekler için bkz . İzlenecek Yol: Özel Veri Akışı Blok Türü Oluşturma.
İlgili Konular'a
| Ünvan | Açıklama |
|---|---|
| Nasıl yapılır: Veri Akışı Bloğuna İletiler Yazma ve Veri Akışı Bloğundan İletiler Okuma | Bir BufferBlock<T> nesnesinde iletilerin nasıl yazılacağını ve okunacağını gösterir. |
| Nasıl yapılır: Üretici-Tüketici Veri Akışı Düzeni Uygulama | Üreticinin bir veri akışı bloğuna ileti gönderdiği ve tüketicinin o bloktan iletileri okuduğu şekilde bir üretici - tüketici modeli uygulamak için veri akışı modelinin nasıl kullanılacağını açıklar. |
| Nasıl yapılır: Veri Akışı Bloğu Veri Aldığında İşlem Gerçekleştirme | Yürütme veri akışı bloğu türleri ActionBlock<TInput>, TransformBlock<TInput,TOutput> ve TransformManyBlock<TInput,TOutput> için nasıl temsilci sağlanabileceğini açıklar. |
| İzlenecek Yol: Veri Akışı İşlem Hattı Oluşturma | İnternet'ten metin indiren ve o metin üzerinde çeşitli işlemler gerçekleştiren bir veri akışı ardışık düzeninin nasıl oluşturulduğunu açıklar. |
| Nasıl yapılır: Veri Akışı Bloklarının Bağlantısını Kaldırma | Kaynak hedefe bir ileti verdikten sonra hedef bloğun kaynağıyla bağlantısını kesmek için yönteminin nasıl kullanılacağını LinkTo gösterir. |
| İzlenecek yol: Windows Forms Uygulaması'nda Veri Akışı Kullanma | Bir Windows Forms uygulamasında görüntü işleme gerçekleştiren bir veri akışı bloğu ağının nasıl oluşturulduğunu gösterir. |
| Nasıl yapılır: Veri Akışı Bloğunu İptal Etme | Bir Windows Forms uygulamasında iptal işleminin nasıl kullanıldığını gösterir. |
| Nasıl yapılır: Birden Fazla Kaynaktan Okumak için JoinBlock'u Kullanma | Birden çok kaynaktan veri kullanılabilir olduğunda bir işlemi gerçekleştirmek içn JoinBlock<T1,T2> sınıfının nasıl kullanıldığını, ve birden çok birleştirme bloğunun bir veri kaynağını daha etkin olarak kullanabilmesini sağlamak için doyumsuz olmayan modun nasıl kullanıldığını açıklar. |
| Nasıl yapılır: Veri Akışı Bloğunda Paralellik Derecesini Belirtme | Bir yürütme veri akışı bloğunun aynı anda birden çok letiyi işlemesi için MaxDegreeOfParallelism özelliğinin nasıl ayarlandığını açıklar. |
| Nasıl yapılır: Veri Akışı Bloğunda Görev Zamanlayıcı Belirtme | Uygulamanızda veri akışı kullandığınızda belirli bir görev zamanlayıcının nasıl ilişkilendirildiğini gösterir. |
| İzlenecek yol: Verimliliği Artırmak için BatchBlock ve BatchedJoinBlock'u Kullanma | Veritabanı ekleme işlemlerinin verimini artırmak için BatchBlock<T> sınıfının; uygulama veritabanından okurken hem sonuçları hem de oluşan özel durumları yakalamak içinse BatchedJoinBlock<T1,T2> sınıfının nasıl kullanıldığını açıklar. |
| İzlenecek Yol: Özel bir Veri Akışı Blok Türü Oluşturma | Özel davranış uygulayan bir veri akışı bloğu türünü oluşturmanın iki yolunu gösterir. |
| Görev Paralel Kitaplığı (TPL) | .NET Framework uygulamalarında paralel ve eşzamanlı programlamayı basitleştiren bir kitaplık olan TPL'yi tanıtır. |
Geri Bildirim
Çok yakında: 2024 boyunca, içerik için geri bildirim mekanizması olarak GitHub Sorunları’nı kullanımdan kaldıracak ve yeni bir geri bildirim sistemiyle değiştireceğiz. Daha fazla bilgi için bkz. https://aka.ms/ContentUserFeedback.
Gönderin ve geri bildirimi görüntüleyin