Parallel——执行并行任务

作者：追风剑情发布于：2017-11-8 21:57 分类：C#
示例
using System;
using System.IO;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
 
//参考 https://www.cnblogs.com/ricky-wang/p/7003162.html
namespace Test3
{
    class Program
    {
        static void Main(string[] args)
        {
            //Parallel调用线程池中的线程并性处理工作,Parallel内部会创建Task
            //Parallel.For的执行效率高于Parallel.ForEach
            //Parallel内部是并行执行,但For或ForEach方法会使调用线程挂起(即,同步执行),直到所有工作完成。
            //如果任何操作抛出未处理的异常,For或ForEach方法会抛出AggregateException
            ParallelLoopResult result = Parallel.For(0, 10, (i, state) => DoWork(i, state));
            Console.WriteLine("是否完成:{0}", result.IsCompleted);
            Console.WriteLine("最低迭代:{0}", result.LowestBreakIteration);
            //如果IsCompleted=false,LowestBreakIteration!=null表示某工作项调用了Break
            //如果IsCompleted=true,LowestBreakIteration=null表示某工作项调用了Stop
 
            //Parallel.ForEach用来遍历集合
            string[] data = { "str1", "str2", "str3", "str4", "str5" };
            result = Parallel.ForEach<string>(data, (str, state, i) =>
            {
                Console.WriteLine("迭代次数：{0},{1}", i, str);
                if (i > 3)
                    state.Break();
                //state.Break() 不再处理当前项之后的项
                //state.Stop() 停止处理任何更多的工作
                //state.IsExceptional true:其他工作项抛出了未处理异常
            });
            Console.WriteLine("是否完成:{0}", result.IsCompleted);
            Console.WriteLine("最低迭代:{0}", result.LowestBreakIteration);
 
            //Parallel.Invoke用于执行并行任务
            Parallel.Invoke(() =>
            {
                Thread.Sleep(100);
                Console.WriteLine("method1");
            }, () =>
            {
                Thread.Sleep(10);
                Console.WriteLine("method2");
            });
 
            //利用Parallel.ForEach统计某个目录下的文件大小
            long masterTotal = DirectoryBytes(@"E:\", "*.jpg", SearchOption.TopDirectoryOnly);
            Console.WriteLine("File TotalSize={0}KB", masterTotal);
            Console.ReadLine();
        }
 
        private static void DoWork(int i, ParallelLoopState state)
        {
            Console.WriteLine("DoWork({0}): 任务ID={1}, 线程ID={1}", i,
                Task.CurrentId,
                Thread.CurrentThread.ManagedThreadId);
            Thread.Sleep(10);
            if (i > 5)
                state.Break();//result.LowestBreakIteration的值为此时的i
        }
 
        private static Int64 DirectoryBytes(String path, String searchPattern, SearchOption searchOption)
        {
            var files = Directory.EnumerateFiles(path, searchPattern, searchOption);
            Int64 masterTotal = 0;
 
            ParallelLoopResult result = Parallel.ForEach<String, Int64>(
                files,
                () =>//localInit: 每个任务开始之前调用一次
                {
                    //每个任务开始之前，总计值都初始化为0
                    return 0;
                },
                (file, loopState, index, taskLocalTotal) =>//body: 每个工作项调用一次
                {
                    //获得这个文件的大小，把它添加到这个任务的累加值上
                    Int64 fileLength = 0;
                    FileStream fs = null;
                    try
                    {
                        fs = File.OpenRead(file);
                        fileLength = fs.Length;
                    }
                    catch (IOException) { /*忽略拒绝访问的任何文件*/ }
                    finally { if (fs != null) fs.Dispose();  }
                    return taskLocalTotal + fileLength;
                },
                taskLocalTotal =>//localFinally: 每个任务完成时调用一次
                {
                    //将这个任务的总计算(taskLocalTotal)加到总的总计值(masterTotal)上
                    //保证线程安全
                    Interlocked.Add(ref masterTotal, taskLocalTotal);
                });
 
            return masterTotal;
        }
    }
}