04-计算限制的异步操作（下）

2018-07-05  本文已影响0人  _秦同学_

[TOC]

计算限制的异步操作（下）

三、Parallel并行循环

3.1 Parallel说明

静态System.Threading.Tasks.Parallel对任务进行了封装，其内部使用Task对象：

// 并行For循环
Parallel.For(0, 1000, i => DoWork(i));
// 并行ForEach循环
Parallel.ForEach(collection, item => DoWork(item));
// 并行执行方法
Parallel.Invoke(
    () => Method1(),
    () => Method2(),
    () => Method3());

Parallel的所有方法都会让调用线程参与处理，若调用线程在其它工作线程之前完成工作，它会将自己挂起，直到所有工作完成。

并行的任何操作抛出未处理的异常，Parallel方法最后都会抛出一个AggregateException异常对象。
Parallel的方法本身也有开销，委托对象必须分配，而针对每个工作项都要调用一次这些委托。为简单的工作项使用Parallel可能会得不偿失。

3.1、Parallel方法的使用

Parallel的For、ForEach和Invoke方法都提供了接受一个ParallelOptions对象的重载,ParallelOptions定义如下：

public class ParallelOptions {
    public CancellationToken CancellationToken{get;set;}    // 默认为Token.None
    public Int32 MaxDegreeOfParallelism {get;set;}  // 可用CPU数，默认-1
    public TaskScheduler TaskScheduler{get;set;}    // 默认为TaskSchduler.Default
}

For、ForEach方法有一些重载版本允许传递3个委托：

以下为演示代码：

static long DirectoryBytes(string path, string searchPattern, SearchOption searchOption)
{
    var files = Directory.EnumerateFiles(path, searchPattern, searchOption);
    long masterTotal = 0;

    ParallelLoopResult result = Parallel.ForEach<String, long>(
        files,

        () => {// localInit ：每个任务开始之前调用一次
            // 每个任务开始之前，总计值都初始化为0
            return 0;   // 将 taskLocalTotal 初始值设置为0
        },

        (file, loopState, index, taskLocalTotal) => {   // body:每个工作项调用一次
            // 获得这个文件的大小，把它添加到这个任务的累加值上
            long 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;
}

每个任务都通过 taskLocalTotal 变量为分配给它的文件维护它自己的总计值；
每个任务在完成之后都通过调用 Interlocked.Add 方法以一种线程安全的方式更新总的总计值(masterTotal)。

3.2、ParallelLoopState 和 ParallelLoopResult

3.2.1 ParallelLoopState

上例中的 body 委托主体中存在一个 ParallelLoopState 参数对象，每个工作任务都将获得它自己的ParallelLoopState对象，并可通过该对象与其它任务进行交互。该参数的主体定义如下：

public class ParallelLoopState {
    public void Stop();
    public Boolean IsStopped { get; }

    public void Break();
    public Int64? LowestBreakIteration { get; }

    public Boolean IsExceptional { get; }
    public Boolean ShouldExitCurrentIteration { get; }
}

成员解释如下：

Break说明：加入ForEach要处理100项，在第五项时调用了Break，那么循环会确保前5项处理好之后ForEach才返回。由于并行循环是无序的，第5项之后的项可能在以前就处理完毕了。

3.2.2 ParallelLoopResult

Parallel的For、ForEach方法都返回一个ParallelLoopResult实例，可检查该实例的相关属性来了解循环结果，该结构的定义如下：

public struct ParallelLoopResult {
    public Boolean IsCompleted { get; }     // 如果操作提前终止则返回 false
    public Int64? LowestBreakIteration { get; }
}

其IsCompleted返回true，则表明循环运行完成，所有项都得到了正确处理。若返回false：

四 定时器

4.1 说明

命名空间：System.Threading.Timer
定义：

public sealed class Timer : MarshalByRefObject, IDisposable {
    public Timer(TimerCallback callback, 
                 Object state, 
                 Int32/Uint32/TimeSpan dueTime, 
                 Int32/Uint32/TimeSpan period);

    public Boolean Change(Int32/UInt32/Int64/TimeSpan dueTime, 
                          Int32/UInt32/Int64/TimeSpan period);

    public Boolean Dispose();
    public Boolean Dispose(WaitHandle notifyObject);
}

其中，TimerCallBack的定义如下

delegate void TimerCallBack(Object state);

Timer构造器的参数说明：

在内部，线程池为所有的Timer对象只使用了一个线程。该线程知道下一个Timer对象在什么时候到期（计时器还有多久触发）。下一个Timer对象到期时，线程就会唤醒，在内部调用ThreadPool.QueueUserWorkItem，将一个工作项添加到线程池的队列中来调用回调方法。
【注意】若回调方法执行时间过长，可能上个回调还没完成而计时器再次触发，会造成多个线程池线程同时执行你的回调方法。
解决方案：

Change方法的签名如下：

public Boolean Change(TimeSpan duetime, TimeSpan period);

完整的Demo如下

private static Timer s_timer;
public static void Main(string[] args) {
    s_timer = new Timer(Status, null, Timeout.Infinite, Timeout.Infinite);
    // 启动定时器
    s_timer.Change(0, Timeout.Infinite);

    Console.ReadLine();
}
private static void Status(Object state) {
    Console.WriteLine("In Status at {0}", DateTime.Now);
    // Do Some Works in here
    s_timer.Change(2000, Timeout.Infinite);
}

也可以使用以下方式完成定时器等效功能：

while(true){
    // Do some workds in here
    await Task.Delay(2000);
}

4.2 FCL提供的多种计时器

五、线程池如何管理线程

5.1 线程池基础

创建和销毁线程是一个昂贵的操作，要耗费大量时间，太多的线程也会浪费内存资源，调度线程时会执行上下文切换影响性能。每个CLR管理一个线程池，这个线程池供CLR控制的所有AppDomain共享。若一个进程中加载了多个CLR，那么每个CLR都有它自己的线程池。
CLR初始化时，线程池中没有线程。线程池维护了一个请求队列，应用每执行一个异步操作时，就调用某个方法，将一个记录项（entry）追加到线程池的队列中。线程池的代码从这个队列中提取记录项并派发（dispatch）给一个线程池线程。
线程池线程处理完毕后会回到线程池中进入空闲状态，等待新的任务。一段时间之后，线程池线程会醒来自己终止自己以释放资源。
随着CLR版本的发布，线程池管理工作线程的内部实现发生了很多变化。开发者使用线程池，最好将它看做一个黑盒。

线程的执行上下文

每个线程都关联了一个执行上下文的数据结构：

线程执行它的代码时，一些操作会受到线程执行上下文设置（尤其是安全设置）的影响。理想情况下，每当一个线程使用另一个线程执行任务时，前者的执行上下文应该流向（复制）辅助线程。这样确保了辅助线程执行的任何操作使用的是相同的安全设置和宿主设置，以及初始线程的逻辑调用上下文中存储的任何数据都适用于辅助线程。

由于执行上下文中包含大量信息，而收集这些信息，再把它们赋值到辅助线程要耗费不少时间。若辅助线程中又启用了更多的辅助线程，则必须创建和初始化更多的执行上下文。

System.Threading.ExecutionContext类，允许控制线程的执行上下文如何从一个线程“流”向另一个。可使用该类组织执行上下文的流动以提升程序性能。其常用类型定义如下：

public sealed class ExecutionContext : IDisposable, ISerializable {
    [SecurityCritical]
    public static AsyncFlowControl SuppressFlow();
    public static void RestoreFlow();
    public static Boolean IsFlowSuppressed();
    ...
}

ExecutionContext 可以阻止上下文流动以提升性能，对于客户端来说性能提升不了多少，但对服务器应用程序来说，性能的提升会非常显著。另外，由于SuppressFlow()方法使用了[SecurityCritical]特性，某些客户端应用程序，如Silverlight，是无法调用的。

static void Main(string[] args)
{
    // 将一些数据放到Main线程的逻辑调用上下文中
    CallContext.LogicalSetData("Name", "Jimmie");
    // 初始化要由一个线程池线程做的一些工作，线程池线程能访问逻辑调用上下文数据
    ThreadPool.QueueUserWorkItem(state => Console.WriteLine("Name={0}", CallContext.LogicalGetData("Name")));
    // 阻止Main线程的执行上下文流动
    ExecutionContext.SuppressFlow();
    // 初始化要由线程池线程做的工作，线程池线程不能访问逻辑调用上下文数据
    ThreadPool.QueueUserWorkItem(state => Console.WriteLine("Name={0}", CallContext.LogicalGetData("Name")));
    // 恢复Main线程的执行上下文流动
    ExecutionContext.RestoreFlow();
    Console.ReadLine();
}
/*
输出如下：
Name=Jimmie
Name=
*/

只有在辅助线程不需要访问上下文信息时，才应该阻止上下文的流动。若初始线程的执行上下文不流向辅助线程，辅助线程就会使用上一次和它关联的任意执行上下文，此时，辅助线程的代码不应该依赖任何执行上下文状态的代码（如用户的Windows身份）。

5.2 设置线程池限制

System.Threading.ThreadPool类提供了几个静态方法用于查询和设置线程池的线程数：

建议不要限制线程的数量，尤其是设置线程上限，这样可能发生饥饿或死锁。

如设置线程池线程最大为1000个，假定队列中有1000个工作项都因为一个事件而堵塞，只有等待第1001个工作项才会解锁。而1001永远不会被创建，此时1000个线程全部堵塞，进程只能被迫中止。

每个线程的创建都要为其用户模式栈和线程环境块（TEB）准备超过1MB的内存。32位的进程中，最多能有1360个线程，视图创建更多的线程会抛出OutOfMemoryException。64位进程提供了8TB的地址空间，理论上可以创建千百万个线程。
而CLR对线程线程的数量存在一个默认的最大值，现在大约1000左右。

5.3 如何管理工作者线程

ThreadPool.QueueUserWorkItem 方法和 Timer 类总是将工作项放到全局队列中。工作者线程采用一个先入先出(First-in First-out,FIFO)算法将工作项从这个队列中取出，并处理它们。由于多个工作者线程可能同时从全局队列中拿走工作项，所以所有工作者线程都竞争一个线程同步锁，以保证两个或多个线程不会获取同一个工作项。这个线程同步锁在某些应用程序中可能成为瓶颈，对伸缩性和性能造成某种程度的限制。

以默认的 TaskScheduler 调度 Task 对象的方式（其他 TaskScheduler 派生对象的行为可能和这里描述的不同）。

暂且将非工作者线程理解成UI线程。

工作者线程准备好处理工作项时，它总是先检查本地队列来查找一个 Task。存在一个 Task，工作者线程就从本地队列中移除 Task 并处理工作项。要注意的是，工作者线程采用后入先出(LIFO)算法将任务从本地队列取出。由于工作者线程是唯一允许访问它自己的本地队列头的线程，所以无需同步锁，而且在队列中添加和删除 Task 的速度非常快。这个行为的副作用是 Task 按照进入队列时相反的顺序执行。

线程池从来不保证排队中的工作项的处理顺序。

如果工作者线程发现它的本地队列变空了，会尝试从另一个工作者线程的本地队列“偷”一个 Task。这个 Task 是从本地队列的尾部“偷”走的，并要求获取一个线程同步锁，这对性能有少许影响。当然，希望这种“偷盗”行为很少发生，从而很少需要线程同步锁。如果所有本地队列都变空，那么工作者线程就会使用FIFO算法，从全局队列提取一个工作项（取得他的锁）。如果全局队列也为空，工作者线程就会进入睡眠状态，等待事情的发生。如果睡眠时间过长，它会自己醒来并销毁自身，允许系统回收线程使用的资源（内核对象、栈、TEB等）。

线程池会快速创建工作者线程，使工作者线程的数量等于传给 ThreadPool 的 SetMinThreads 方法的值。如果从不调用这个方法（也建议永远不要调用这个方法），那么默认的值等于你的进程允许使用的CPU数量，这是由进程的 affinity mask（关联掩码）决定的。通常，你的进程允许使用机器上所有的CPU，所以线程池创建的工作者线程数量很快就会达到机器的CPU数。创建了这么多的CPU数量的线程后，线程池会监视工作项的完成速度。如果工作项完成的时间太长（具体多长时间没有正式公布），线程池会创建更多的工作者线程，如果工作项完成的速度开始变快，工作者线程就会被销毁。

上一篇下一篇

猜你喜欢

热点阅读