.NET 實現並行的幾種方式（二）

本隨筆續接：.NET 實現並行的幾種方式（一） 

四、Task  

3）Task.NET 4.5 中的簡易方式

在上篇隨筆中，兩個Demo使用的是 .NET 4.0 中的方式，代碼寫起來略顯麻煩，這不 .NET 4.5提供了更加簡潔的方式，讓我們來看一下吧。

        /// <summary>
        /// Task.NET 4.5 中的簡易方式
        /// </summary>
        public void Demo3()
        {
            Task.Run(() =>
            {
                SetTip("簡潔的代碼");
            });


            Task.Run(() =>
            {
                SetTip("驗證 CreationOptions 屬性");
            }).ContinueWith((t)=> {
                SetTip("CreationOptions:" + t.CreationOptions.ToString());                
            });
        }

 

五、TPL (Task Parallel Library)

TPL （任務並行庫）是 .NET 4.0 中的另一個重量級模塊，可以極其優雅、便捷地完成並行邏輯的編碼工作。

1）Parallel.Invoke並行多個獨立的Action

        /// <summary>
        /// Parallel.Invoke並行多個獨立的Action
        /// </summary>
        public void Demo1()
        {
            Task.Run(() =>
            {
                List<Action> actions = new List<Action>();

                // 生成並行任務
                for (int i = 0; i < 5; i++)
                {
                    // 注意、這里很關鍵，不可直接使用i變量。 
                    // 原因在稍后的隨筆中進行說明
                    int index = i;
                    actions.Add(new Action(() =>
                    {
                        SetTip(string.Format("Task{0} 開始", index));

                        SetTip(string.Format("Task{0} 休眠1秒", index));
                        Thread.Sleep(1000);


                        SetTip(string.Format("Task{0} 休眠5秒", index));
                        Thread.Sleep(5000);
                        
                        SetTip(string.Format("Task{0} 結束", index));
                    }));
                }

                // 執行並行任務
                Parallel.Invoke(actions.ToArray());

                // 當上述的5個任務全部執行完畢后，才會執行該代碼
                SetTip("並行任務執行完畢");
            });
        }

 

2）Parallel簡單的For並行

如果 Parallel.Invoke 看做是任務並行， 則 Parallel.For 則是數據並行，可方便的完成For循環並行遍歷。

        /// <summary>
        /// Parallel簡單的For並行
        /// </summary>
        public void Demo2()
        {
            // 為了實時更新UI、將代碼異步執行
            Task.Run(() =>
            {
                Parallel.For(1, 100, (index) =>
                {
                    SetTip(string.Format("Index:{0}, 開始執行Task", index));

                    Thread.Sleep(1000);
                    SetTip(string.Format("Index:{0}, 開始休眠Action 1秒", index));

                    SetTip(string.Format("Index:{0}, Task執行完畢", index));
                });

                SetTip("並行任務執行完畢");
            });
        }

 

3）Parallel.For並行 並行中的 break、 return、 continue

break : 在 Parallel.For 中使用 ParallelLoopState.Break() 方法代替。

return: 在 Parallel.For 中使用 ParallelLoopState.Break() 方法代替。

continue : 在 Parallel.For 中直接使用 return 即可。

        /// <summary>
        /// 中斷Parallel.For並行
        /// </summary>
        public void Demo3()
        {
            // 為了實時更新UI、將代碼異步執行
            Task.Run(() =>
            {
                int breakIndex = new Random().Next(10, 50);
                SetTip(" BreakIndex : -------------------------" + breakIndex);

                Parallel.For(1, 100, (index, state) =>
                {
                    SetTip(string.Format("Index:{0}, 開始執行Task", index));

                    if (breakIndex == index)
                    {
                        SetTip(string.Format("Index:{0}, ------------------ Break Task", index));
                        state.Break();
                        // Break方法執行后、
                        // 大於 當前索引的並且未被安排執行的迭代將被放棄
                        // 小於 當前索引的的迭代將繼續正常執行直至迭代執行完畢
                        return;
                    }

                    Thread.Sleep(1000);
                    SetTip(string.Format("Index:{0}, 休眠Action 1秒", index));

                    SetTip(string.Format("Index:{0}, Task執行完畢", index));
                });

                SetTip("並行任務執行完畢");
            });
        }


        /// <summary>
        /// 終止Parallel.For並行
        /// </summary>
        public void Demo4()
        {
            // 為了實時更新UI、將代碼異步執行
            Task.Run(() =>
            {
                int stopIndex = new Random().Next(10, 50);
                SetTip(" StopIndex : -------------------------" + stopIndex);

                Parallel.For(1, 100, (index, state) =>
                {
                    SetTip(string.Format("Index:{0}, 開始執行Task", index));

                    if (stopIndex == index)
                    {
                        SetTip(string.Format("Index:{0}, ------------------ Stop Task", index));
                        state.Stop();
                        // Stop方法執行后
                        // 整個迭代將被放棄
                        return;
                    }

                    Thread.Sleep(1000);
                    SetTip(string.Format("Index:{0}, 休眠Action 1秒", index));

                    SetTip(string.Format("Index:{0}, Task執行完畢", index));
                });

                SetTip("並行任務執行完畢");
            });
        }

 

4）Parallel.For並行中的數據聚合

在並行中，絕大多數委托都是在不同的線程中運行的，如果需要在並行中進行的數據共享、則需要考慮線程同步問題，然而線程同步會影響並行性能。

為了解決特定情況下的數據共享，而又不會因為線程同步而影響性能，Parallel.For 提供了解決方案：

        /// <summary>
        /// Parallel.For並行中的數據聚合
        /// </summary>
        public void Demo5()
        {
            Task.Run(() =>
            {
                // 求 1 到 10 的階乘的 和
                long total = 0;
                Parallel.For<long>(1, 10,
                    () =>
                    {
                        SetTip("LocalInit");
                        return 0;
                    },
                    (index, state, local) =>
                    {
                        SetTip("Body");
                        int result = 1;
                        for (int i = 2; i < index; i++)
                        {
                            result *= i;
                        }
                        local += result;
                        return local;
                    },
                    (x) =>
                    {
                        SetTip("LocalFinally");
                        Interlocked.Add(ref total, x);
                    });

                SetTip("Total : " + total);
                SetTip("並行任務執行完畢");
            });

        }

MSDN備注：
對於參與循環執行的每個線程調用 LocalInit 委托一次，並返回每個線程的初始本地狀態。
這些初始狀態傳遞到每個線程上的第一個 body 調用。 然后，每個后續正文調用返回可能修改過的狀態值，傳遞到下一個正文調用。
最后，每個線程上的最后正文調用返回傳遞給 LocalFinally 委托的狀態值。
每個線程調用 localFinally 委托一次，以對每個線程的本地狀態執行最終操作。
此委托可以被多個線程同步調用；因此您必須同步對任何共享變量的訪問。

也就是說：
1) 並行中開辟的線程數 決定了 LocalInit、LocalFinally 的調用次數
2) 多個 迭代委托、Body 可能被同一個線程調用。
3) 迭代委托、Body 中的 local值，並不一定是 LocalInit 的初始值，也有可能是被修改的返回值。
4) LocalFinally 可能是被同時調用的，需要注意線程同步問題。

 

5）Parallel.ForEach並行

Parallel.ForEach算是另一種數據並行方式， 它與大家熟知的 IEnumerable<TSource> 接口結合十分緊密，是 foreach的並行版本。

        /// <summary>
        /// Parallel.ForEach並行
        /// </summary>
        public void Demo6()
        {
            Task.Run(() =>
            {
                Parallel.ForEach<int>(Enumerable.Range(1, 10), (num) =>
                {
                    SetTip("Task 開始");


                    SetTip("Task 休眠" + num + "秒");
                    Thread.Sleep(TimeSpan.FromSeconds(num));


                    SetTip("Task 結束");
                });

                SetTip("並行任務執行完畢");
            });
        }

 

6）Parallel.ForEach中的索引，中斷、終止操作

在 Parallel.ForEach 中也可以輕易的獲得其遍歷的索引

        /// <summary>
        /// Parallel.ForEach中的索引，中斷、終止操作
        /// </summary>
        public void Demo7()
        {
            Task.Run(() =>
            {
                Parallel.ForEach<int>(Enumerable.Range(0, 10), (num, state, index) =>
                {
                    // num, 並行數據源中的數據項
                    // state, 
                    SetTip(" Index : " + index + "         Num: " + num);
                });
                SetTip("並行任務執行完畢");
            });
        }

 

 

本隨筆到此、暫告一段落。

附，Demo : http://files.cnblogs.com/files/08shiyan/ParallelDemo.zip

 

參見更多：隨筆導讀：同步與異步

(未完待續...)