屬性TPL Dataflow是微軟面向高並發應用而推出的一個類庫。借助於異步消息傳遞與管道，它可以提供比線程池更好的控制，也比手工線程方式具備更好的性能。我們常常可以消息傳遞，生產-消費模式或Actor-Agent模式中使用。在TDF是構建於Task Parallel Library (TPL)之上的，它是我們開發高性能，高並發的應用程序的又一利器。您可以在NuGet中下載使用，目前最新的版本只支持.net framework 4.5。最早支持.net framework 4.0是作為Microsoft Visual Studio Async CTP中的一部分發布的，你可以在這里下載到。
TDP的主要作用就是Buffering Data和Processing Data，在TDF中，有兩個非常重要的接口，ISourceBlock<T> 和ITargetBlock<T>接口。繼承於ISourceBlock<T>的對象時作為提供數據的數據源對象-生產者，而繼承於ITargetBlock<T>接口類主要是扮演目標對象-消費者。在這個類庫中，System.Threading.Tasks.Dataflow名稱空間下，提供了很多以Block名字結尾的類，ActionBlock，BufferBlock，TransformBlock，BroadcastBlock等9個Block，我們在開發中通常使用單個或多個Block組合的方式來實現一些功能。以下我們逐個來簡單介紹一下。
BufferBlock
BufferBlock是TDF中最基礎的Block。BufferBlock提供了一個有界限或沒有界限的Buffer，該Buffer中存儲T。該Block很像BlockingCollection<T>。可以用過Post往里面添加數據，也可以通過Receive方法阻塞或異步的的獲取數據，數據處理的順序是FIFO的。它也可以通過Link向其他Block輸出數據。
簡單的同步的生產者消費者代碼示例：
private static BufferBlock<int> m_buffer = new BufferBlock<int>();
// Producer
private static void Producer()
{
while(true)
{
int item = Produce();
m_buffer.Post(item);
}
}
// Consumer
private static void Consumer()
{
while(true)
{
int item = m_buffer.Receive();
Process(item);
}
}
// Main
public static void Main()
{
var p = Task.Factory.StartNew(Producer);
var c = Task.Factory.StartNew(Consumer);
Task.WaitAll(p,c);
}
ActionBlock
ActionBlock實現ITargetBlock，說明它是消費數據的，也就是對輸入的一些數據進行處理。它在構造函數中，允許輸入一個委托，來對每一個進來的數據進行一些操作。如果使用Action(T)委托，那說明每一個數據的處理完成需要等待這個委托方法結束，如果使用了Func<TInput, Task>)來構造的話，那么數據的結束將不是委托的返回，而是Task的結束。默認情況下，ActionBlock會FIFO的處理每一個數據，而且一次只能處理一個數據，一個處理完了再處理第二個，但也可以通過配置來並行的執行多個數據。
先看一個例子：
public ActionBlock<int> abSync = new ActionBlock<int>((i) =>
{
Thread.Sleep(1000);
Console.WriteLine(i + " ThreadId:" + Thread.CurrentThread.ManagedThreadId + " Execute Time:" + DateTime.Now);
}
);
public void TestSync()
{
for (int i = 0; i < 10; i++)
{
abSync.Post(i);
}
Console.WriteLine("Post finished");
}
可見，ActionBlock是順序處理數據的，這也是ActionBlock一大特性之一。主線程在往ActionBlock中Post數據以后馬上返回，具體數據的處理是另外一個線程來做的。數據是異步處理的，但處理本身是同步的，這樣在一定程度上保證數據處理的准確性。下面的例子是使用async和await。
public ActionBlock<int> abSync2 = new ActionBlock<int>(async (i) =>
{
await Task.Delay(1000);
Console.WriteLine(i + " ThreadId:" + Thread.CurrentThread.ManagedThreadId + " Execute Time:" + DateTime.Now);
}
![U55C4LS4`0SY0O)}[5W]{%C](/image/aHR0cHM6Ly9pbWFnZXMwLmNuYmxvZ3MuY29tL2Jsb2cvMTU3MDAvMjAxMzAzLzAxMTYxNTI1LWFiM2QxNWUzZTkxNjQxYmU5YTE1OTg4YTg4ZWZmZGRmLmpwZw==.png "U55C4LS4`0SY0O)}[5W]{%C")
雖然還是1秒鍾處理一個數據，但是處理數據的線程會有不同。
如果你想異步處理多個消息的話，ActionBlock也提供了一些接口，讓你輕松實現。在ActionBlock的構造函數中，可以提供一個ExecutionDataflowBlockOptions的類型，讓你定義ActionBlock的執行選項，在下面了例子中，我們定義了MaxDegreeOfParallelism選項，設置為3。目的的讓ActionBlock中的Item最多可以3個並行處理。
public ActionBlock<int> abAsync = new ActionBlock<int>((i) =>
{
Thread.Sleep(1000);
Console.WriteLine(i + " ThreadId:" + Thread.CurrentThread.ManagedThreadId + " Execute Time:" + DateTime.Now);
}
, new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = 3 });
public void TestAsync()
{
for (int i = 0; i < 10; i++)
{
abAsync.Post(i);
}
Console.WriteLine("Post finished");
}
![XVGW}JJK7YY7(%E}])11J7V](/image/aHR0cHM6Ly9pbWFnZXMwLmNuYmxvZ3MuY29tL2Jsb2cvMTU3MDAvMjAxMzAzLzAxMTYxNTMwLWM1MWI3YTJlODQyNTRlZGViZTJkODVhNjNlM2JmNTc2LmpwZw==.png "XVGW}JJK7YY7(%E}])11J7V")
運行程序，我們看見，每3個數據幾乎同時處理，並且他們的線程ID也是不一樣的。
ActionBlock也有自己的生命周期，所有繼承IDataflowBlock的類型都有Completion屬性和Complete方法。調用Complete方法是讓ActionBlock停止接收數據，而Completion屬性則是一個Task，是在ActionBlock處理完所有數據時候會執行的任務，我們可以使用Completion.Wait()方法來等待ActionBlock完成所有的任務，Completion屬性只有在設置了Complete方法后才會有效。
public void TestAsync()
{
for (int i = 0; i < 10; i++)
{
abAsync.Post(i);
}
abAsync.Complete();
Console.WriteLine("Post finished");
abAsync.Completion.Wait();
Console.WriteLine("Process finished");
}
TransformBlock
TransformBlock是TDF提供的另一種Block，顧名思義它常常在數據流中充當數據轉換處理的功能。在TransformBlock內部維護了2個Queue，一個InputQueue，一個OutputQueue。InputQueue存儲輸入的數據，而通過Transform處理以后的數據則放在OutputQueue，OutputQueue就好像是一個BufferBlock。最終我們可以通過Receive方法來阻塞的一個一個獲取OutputQueue中的數據。TransformBlock的Completion.Wait()方法只有在OutputQueue中的數據為0的時候才會返回。
舉個例子，我們有一組網址的URL，我們需要對每個URL下載它的HTML數據並存儲。那我們通過如下的代碼來完成：
public TransformBlock<string, string> tbUrl = new TransformBlock<string, string>((url) =>
{
WebClient webClient = new WebClient();
return webClient.DownloadString(new Uri(url));
}
public void TestDownloadHTML()
{
tbUrl.Post("www.baidu.com");
tbUrl.Post("www.sina.com.cn");
string baiduHTML = tbUrl.Receive();
string sinaHTML = tbUrl.Receive();
}
當然，Post操作和Receive操作可以在不同的線程中進行，Receive操作同樣也是阻塞操作，在OutputQueue中有可用的數據時，才會返回。
TransformManyBlock
TransformManyBlock和TransformBlock非常類似，關鍵的不同點是，TransformBlock對應於一個輸入數據只有一個輸出數據，而TransformManyBlock可以有多個，及可以從InputQueue中取一個數據出來，然后放多個數據放入到OutputQueue中。
TransformManyBlock<int, int> tmb = new TransformManyBlock<int, int>((i) => { return new int[] { i, i + 1 }; });
ActionBlock<int> ab = new ActionBlock<int>((i) => Console.WriteLine(i));
public void TestSync()
{
tmb.LinkTo(ab);
for (int i = 0; i < 4; i++)
{
tmb.Post(i);
}
Console.WriteLine("Finished post");
}
![GC(K]J4DB4UKP$S@8C9ZVMV](/image/aHR0cHM6Ly9pbWFnZXMwLmNuYmxvZ3MuY29tL2Jsb2cvMTU3MDAvMjAxMzAzLzAxMTYxNTQyLTZkNmMwZGQyMzI3YTQzZjU5ODJhZDU0MDU0ODdiNzE3LmpwZw==.png "GC(K]J4DB4UKP$S@8C9ZVMV")
BroadcastBlock
BroadcastBlock的作用不像BufferBlock，它是使命是讓所有和它相聯的目標Block都收到數據的副本，這點從它的命名上面就可以看出來了。還有一點不同的是，BroadcastBlock並不保存數據，在每一個數據被發送到所有接收者以后，這條數據就會被后面最新的一條數據所覆蓋。如沒有目標Block和BroadcastBlock相連的話，數據將被丟棄。但BroadcastBlock總會保存最后一個數據，不管這個數據是不是被發出去過，如果有一個新的目標Block連上來，那么這個Block將收到這個最后一個數據。
BroadcastBlock<int> bb = new BroadcastBlock<int>((i) => { return i; });
ActionBlock<int> displayBlock = new ActionBlock<int>((i) => Console.WriteLine("Displayed " + i));
ActionBlock<int> saveBlock = new ActionBlock<int>((i) => Console.WriteLine("Saved " + i));
ActionBlock<int> sendBlock = new ActionBlock<int>((i) => Console.WriteLine("Sent " + i));
public void TestSync()
{
bb.LinkTo(displayBlock);
bb.LinkTo(saveBlock);
bb.LinkTo(sendBlock);
for (int i = 0; i < 4; i++)
{
bb.Post(i);
}
Console.WriteLine("Post finished");
}
![A][PVWN1@4UMGZ[YTEV$[E9](/image/aHR0cHM6Ly9pbWFnZXMwLmNuYmxvZ3MuY29tL2Jsb2cvMTU3MDAvMjAxMzAzLzAxMTYxNTQ1LWU5NzI0ZjNlMDc0NjQ0ZjViNWQ3NTZlZmUzZWE0YzAyLmpwZw==.png "A][PVWN1@4UMGZ[YTEV$[E9")
如果我們在Post以后再添加連接Block的話，那些Block就只會收到最后一個數據了。
public void TestSync()
{
for (int i = 0; i < 4; i++)
{
bb.Post(i);
}
Thread.Sleep(5000);
bb.LinkTo(displayBlock);
bb.LinkTo(saveBlock);
bb.LinkTo(sendBlock);
Console.WriteLine("Post finished");
}
WriteOnceBlock
如果說BufferBlock是最基本的Block，那么WriteOnceBock則是最最簡單的Block。它最多只能存儲一個數據，一旦這個數據被發送出去以后，這個數據還是會留在Block中，但不會被刪除或被新來的數據替換，同樣所有的接收者都會收到這個數據的備份。
和BroadcastBlock同樣的代碼，但是結果不一樣：
WriteOnceBlock<int> bb = new WriteOnceBlock<int>((i) => { return i; });
ActionBlock<int> displayBlock = new ActionBlock<int>((i) => Console.WriteLine("Displayed " + i));
ActionBlock<int> saveBlock = new ActionBlock<int>((i) => Console.WriteLine("Saved " + i));
ActionBlock<int> sendBlock = new ActionBlock<int>((i) => Console.WriteLine("Sent " + i));
public void TestSync()
{
bb.LinkTo(displayBlock);
bb.LinkTo(saveBlock);
bb.LinkTo(sendBlock);
for (int i = 0; i < 4; i++)
{
bb.Post(i);
}
Console.WriteLine("Post finished");
}
WriteOnceBock只會接收一次數據。而且始終保留那個數據。
同樣使用Receive方法來獲取數據也是一樣的結果，獲取到的都是第一個數據：
public void TestReceive()
{
for (int i = 0; i < 4; i++)
{
bb.Post(i);
}
Console.WriteLine("Post finished");
Console.WriteLine("1st Receive:" + bb.Receive());
Console.WriteLine("2nd Receive:" + bb.Receive());
Console.WriteLine("3rd Receive:" + bb.Receive());
}
![7M5Q]MH5K82OVQ}N]E(J8MV](/image/aHR0cHM6Ly9pbWFnZXMwLmNuYmxvZ3MuY29tL2Jsb2cvMTU3MDAvMjAxMzAzLzAxMTYxNjA1LWU0ZmYwZmE3MDVjOTQxMDU5YjNkMGQ2ZDAwODdlYjI1LmpwZw==.png "7M5Q]MH5K82OVQ}N]E(J8MV")
BatchBlock
BatchBlock提供了能夠把多個單個的數據組合起來處理的功能，如上圖。應對有些需求需要固定多個數據才能處理的問題。在構造函數中需要制定多少個為一個Batch，一旦它收到了那個數量的數據后，會打包放在它的OutputQueue中。當BatchBlock被調用Complete告知Post數據結束的時候，會把InputQueue中余下的數據打包放入OutputQueue中等待處理，而不管InputQueue中的數據量是不是滿足構造函數的數量。
BatchBlock<int> bb = new BatchBlock<int>(3);
ActionBlock<int[]> ab = new ActionBlock<int[]>((i) =>
{
string s = string.Empty;
foreach (int m in i)
{
s += m + " ";
}
Console.WriteLine(s);
});
public void TestSync()
{
bb.LinkTo(ab);
for (int i = 0; i < 10; i++)
{
bb.Post(i);
}
bb.Complete();
Console.WriteLine("Finished post");
}
BatchBlock執行數據有兩種模式：貪婪模式和非貪婪模式。貪婪模式是默認的。貪婪模式是指任何Post到BatchBlock，BatchBlock都接收，並等待個數滿了以后處理。非貪婪模式是指BatchBlock需要等到構造函數中設置的BatchSize個數的Source都向BatchBlock發數據，Post數據的時候才會處理。不然都會留在Source的Queue中。也就是說BatchBlock可以使用在每次從N個Source那個收一個數據打包處理或從1個Source那里收N個數據打包處理。這里的Source是指其他的繼承ISourceBlock的，用LinkTo連接到這個BatchBlock的Block。
在另一個構造參數中GroupingDataflowBlockOptions，可以通過設置Greedy屬性來選擇是否貪婪模式和MaxNumberOfGroups來設置最大產生Batch的數量，如果到達了這個數量，BatchBlock將不會再接收數據。
JoinBlock
JoinBlock一看名字就知道是需要和兩個或兩個以上的Source Block相連接的。它的作用就是等待一個數據組合，這個組合需要的數據都到達了，它才會處理數據，並把這個組合作為一個Tuple傳遞給目標Block。舉個例子，如果定義了JoinBlock<int, string>類型，那么JoinBlock內部會有兩個ITargetBlock，一個接收int類型的數據，一個接收string類型的數據。那只有當兩個ITargetBlock都收到各自的數據后，才會放到JoinBlock的OutputQueue中，輸出。
JoinBlock<int, string> jb = new JoinBlock<int, string>();
ActionBlock<Tuple<int, string>> ab = new ActionBlock<Tuple<int, string>>((i) =>
{
Console.WriteLine(i.Item1 + " " + i.Item2);
});
public void TestSync()
{
jb.LinkTo(ab);
for (int i = 0; i < 5; i++)
{
jb.Target1.Post(i);
}
for (int i = 5; i > 0; i--)
{
Thread.Sleep(1000);
jb.Target2.Post(i.ToString());
}
Console.WriteLine("Finished post");
}
BatchedJoinBlock
BatchedJoinBlock一看就是BacthBlock和JoinBlick的組合。JoinBlick是組合目標隊列的一個數據，而BatchedJoinBlock是組合目標隊列的N個數據，當然這個N可以在構造函數中配置。如果我們定義的是BatchedJoinBlock<int, string>， 那么在最后的OutputQueue中存儲的是Tuple<IList<int>, IList<string>>，也就是說最后得到的數據是Tuple<IList<int>, IList<string>>。它的行為是這樣的，還是假設上文的定義，BatchedJoinBlock<int, string>， 構造BatchSize輸入為3。那么在這個BatchedJoinBlock種會有兩個ITargetBlock，會接收Post的數據。那什么時候會生成一個Tuple<IList<int>，IList<string>>到OutputQueue中呢，測試下來並不是我們想的需要有3個int數據和3個string數據，而是只要2個ITargetBlock中的數據個數加起來等於3就可以了。3和0,2和1，1和2或0和3的組合都會生成Tuple<IList<int>，IList<string>>到OutputQueue中。可以參看下面的例子：
BatchedJoinBlock<int, string> bjb = new BatchedJoinBlock<int, string>(3);
ActionBlock<Tuple<IList<int>, IList<string>>> ab = new ActionBlock<Tuple<IList<int>, IList<string>>>((i) =>
{
Console.WriteLine("-----------------------------");
foreach (int m in i.Item1)
{
Console.WriteLine(m);
};
foreach (string s in i.Item2)
{
Console.WriteLine(s);
};
});
public void TestSync()
{
bjb.LinkTo(ab);
for (int i = 0; i < 5; i++)
{
bjb.Target1.Post(i);
}
for (int i = 5; i > 0; i--)
{
bjb.Target2.Post(i.ToString());
}
Console.WriteLine("Finished post");
}
![GZ}X_[]DM}42_()PXL05A(T](/image/aHR0cHM6Ly9pbWFnZXMwLmNuYmxvZ3MuY29tL2Jsb2cvMTU3MDAvMjAxMzAzLzAxMTYxNjMxLTE1MjgxNWNhYzI1NzQ2YzlhNWEwNGQwOWJjN2NmMTcyLmpwZw==.png "GZ}X_[]DM}42_()PXL05A(T")
最后剩下的一個數據1，由於沒有滿3個，所以一直被保留在Target2中。
TDF中最有用的功能之一就是多個Block之間可以組合應用。ISourceBlock可以連接ITargetBlock，一對一，一對多，或多對多。下面的例子就是一個TransformBlock和一個ActionBlock的組合。TransformBlock用來把數據*2，並轉換成字符串，然后把數據扔到ActionBlock中，而ActionBlock則用來最后的處理數據打印結果。
public ActionBlock<string> abSync = new ActionBlock<string>((i) =>
{
Thread.Sleep(1000);
Console.WriteLine(i + " ThreadId:" + Thread.CurrentThread.ManagedThreadId + " Execute Time:" + DateTime.Now);
}
);
public TransformBlock<int, string> tbSync = new TransformBlock<int, string>((i) =>
{
i = i * 2;
return i.ToString();
}
);
public void TestSync()
{
tbSync.LinkTo(abSync);
for (int i = 0; i < 10; i++)
{
tbSync.Post(i);
}
tbSync.Complete();
Console.WriteLine("Post finished");
tbSync.Completion.Wait();
Console.WriteLine("TransformBlock process finished");
}
