返回

计算限制的异步操作

CLR线程池基础

线程池是你的应用程序能使用的线程集合。每个CLR都有一个线程池,这个线程池由CLR控制的所有AppDomain共享

CLR初始化时,线程池是没有线程的。在内部,线程池维护了一个操作请求队列。应用程序想执行一个异步操作时,就调用某个方法,将一个记录项(entry)追加到线程池的队列中。线程池的代码从这个队列中提取记录项,将这个记录项派遣(dispatch)给一个线程池线程。如果线程池中没有线程,就创建新的线程。创建线程要产生一定的性能损失。然而,当线程池完成任务后,线程不会被销毁。相反,线程会返回线程池,在那里进入空闲状态,等待响应另一个请求。由于线程不销毁自身,所以不再产生额外的性能损失。

如果你的应用程序向线程池发出许多请求,线程池会尝试只用一个线程来服务所有的请求。然而,如果你的应用程序发出请求的速度超过了线程池处理它们的速度,就会创建额外的线程。最终,你的应用程序所有请求都可能有少量的线程处理,所有线程池不必创建大量的线程。

当线程闲着没事一段时间后, 线程会自己醒来终止自己以释放资源, 会产生一定的性能损失。 这个性能损失关系不大。

执行简单的计算限制操作

将一个异步的、计算限制的操作放到一个线程池的队列中,通常可以调用ThreadPool类定义的以下方法之一:

//将方法排入队列以便执行。此方法在有线程池线程变得可用时执行。
static Boolean QueueUserWorkItem(WaitCallback callBack);
//将方法排入队列以便执行,并指定包含该方法所用数据的对象。此方法在有线程池线程变得可用时执行。
static Boolean QueueUserWorkItem(WaitCallback callBack, Object state);

这些方法向线程池的队列中添加一个"工作项"(work item)以及可选的状态数据,然后所有方法立即返回。最终线程池中的某个线程会处理工作项,造成你指定的方法被调用。你写的回调方法必须匹配System.Threading.WaitCallBack委托类型,它的定义如下:

delegate void WaitCallback(Object state);

工作项其实就是由callBack参数标识的一个方法该方法将由线程池线程调用。可通过state实参(状态数据)向方法传递一个参数。无state参数的那个版本的QueueUserWorkItem则向回调方法传递null

WaitCallback委托, TimerCallback委托, ParameterizedThreadStart委托签名完全一致, 使用ThreadPool.QueueUserWorkItem, System.Threading.Timer和System.Threading.Thread对象都可以调用该方法。

以下演示了如何让一个线程池线程以异步方式调用一个方法:

    class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("Main thread: queuing an asynchronous operation");
            ThreadPool.QueueUserWorkItem(ComputeBoundOp, 5);
            Console.WriteLine("Main thread: Doing other work here...");
            Thread.Sleep(10000);  // 模拟其它工作 (10 秒钟)
            //Console.ReadLine();
        }

        // 这是一个回调方法,必须和WaitCallBack委托签名一致
        private static void ComputeBoundOp(Object state)
        {
            // 这个方法通过线程池中线程执行
            Console.WriteLine("In ComputeBoundOp: state={0}", state);
            Thread.Sleep(1000);  // 模拟其它工作 (1 秒钟)

            // 这个方法返回后,线程回到线程池,等待其他任务
        }
    }
    // 可以输出以下结果
    // Main thread: queuing an asynchronous operation
    // Main thread: Doing other work here...
    // In ComputeBoundOp: state=5
    // 但有时也会得到一下输出:
    // Main thread: queuing an asynchronous operation
    // In ComputeBoundOp: state=5
    // Main thread: Doing other work here...

执行上下文

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

执行上下文(execution context)包括的东西有:

  • 安全设置(压缩栈、Thread的Principal属性[指示线程的调度优先级]和Windows身份)
  • 宿主设置(参见System.Threading.HostExecutionContextManager[提供使公共语言运行时宿主可以参与执行上下文的流动(或移植)的功能])
  • 逻辑调用上下文数据 (参见System.Runtime.Remoting.Messaging.CallContext[提供与执行代码路径一起传送的属性集]LogicalSetData[将一个给定对象存储在逻辑调用上下文中并将该对象与指定名称相关联]LogicalGetData[从逻辑调用上下文中检索具有指定名称的对象]方法).

线程执行代码时,有的操作会受到线程执行上下文设置(尤其是安全设置)的影响。

默认情况下,CLR自动造成初始线程的执行上下文会”流向”(复制到)任何辅助线程。因为执行上下文中包含大量信息,而收集这些信息,再将这些信息复制到辅助线程,要耗费不少时间。

System.Threading命名空间中有一个ExecutionContext类[管理当前线程的执行上下文],它允许你控制线程的执行上下文如何从一个线程”流向”(复制到)另一个线程。下面展示了这个类的样子:

public sealed class ExecutionContext : IDisposable, ISerializable
{
    [SecurityCritical]
    //取消执行上下文在异步线程之间的流动
    public static AsyncFlowControl SuppressFlow();
    //恢复执行上下文在异步线程之间的流动
    public static void RestoreFlow();
    //指示当前是否取消了执行上下文的流动。
    public static bool IsFlowSuppressed();

    //不常用方法没有列出
}

下例展示了向CLR的线程池队列添加一个工作项的时候,如何通过阻止执行上下文的流动来影响线程逻辑调用上下文中的数据:

class Program
{
        static void Main(string[] args)
        {
            // 将一些数据放到Main线程的逻辑调用上下文中
            CallContext.LogicalSetData("Name", "Jeffrey");

            // 线程池能访问到逻辑调用上下文数据,加入到程序池队列中
            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();

            //再次访问逻辑调用上下文的数据
            ThreadPool.QueueUserWorkItem(
               state => Console.WriteLine("Name={0}", CallContext.LogicalGetData("Name")));
            Console.Read();
        }
}

// Name=Jeffrey
// Name=
// Name=Jeffrey

协作式取消和超时

Microsoft .NET Framework提供了一个标准的取消操作模式。这个模式是协作式的,意味着你想取消的操作必须显式的支持取消

首先必须创建一个System.Thread.CancellationTokenSource[通知 CancellationToken,告知其应被取消]对象。这个类如下所示:

public class CancellationTokenSource : IDisposable
{
         //构造函数
         public CancellationTokenSource();
         //获取是否已请求取消此 System.Threading.CancellationTokenSource
         public bool IsCancellationRequested { get; }
         //获取与此 System.Threading.CancellationTokenSource 关联的 System.Threading.CancellationToken
         public CancellationToken Token;
         //传达取消请求。
         public void Cancel();
         //传达对取消的请求,并指定是否应处理其余回调和可取消操作。
         public void Cancel(bool throwOnFirstException);
        ...
 }

这个对象包含了管理取消有关的所有状态。构造好一个CancellationTokenSource(引用类型)之后,可以从它的Token属性获得一个或多个CancellationToken(值类型)实例,并传给你的操作,使那些操作可以取消。以下是CancellationToken值类型最有用的一些成员:

public struct CancellationToken  //一个值类型
{
       // 获取此标记是否能处于已取消状态,
       // 由非通过Task来调用(invoke)的一个操作调用(call)
       public bool IsCancellationRequested { get; }
       // 如果已请求取消此标记,则引发 System.OperationCanceledException,
       // 由通过Task来调用的操作调用
       public void ThrowIfCancellationRequested();
       // 获取在取消标记时处于有信号状态的
       // CancellationTokenSource 取消时,WaitHandle会收到信号
       public WaitHandle WaitHandle { get; }
       //返回空 CancellationToken 值。
       public static CancellationToken None;
       // 判断此CancellationToken是否是特殊的None实例, 特殊的是不允许取消操作
       public bool CanBeCanceled{ get; }
       //注册一个将在取消此 System.Threading.CancellationToken 时调用的委托。省略了简单重载版本
       public CancellationTokenRegistration Register(Action<object> callback, object state, bool useSynchronizationContext);

       //省略了GetHashCode、Equals成员
}

CancellationToken实例是一个轻量级的值类型,它包含单个私有字段:对它的CancellationTokenSource对象的一个引用。在一个计算限制操作的循环中,可以定时调用CancellationTokenIsCancellationRequested属性了解循环是否应该提前终止,进而终止计算限制的操作

用一些示例代码演示一下:

class Program
{
        static void Main(string[] args)
        {
            CancellationTokenSource cts = new CancellationTokenSource();

            // 将CancellationToken和"要循环到的目标数"传入操作中
            ThreadPool.QueueUserWorkItem(o => Count(cts.Token, 1000));

            Console.WriteLine("Press <Enter> to cancel the operation.");
            Console.ReadLine();
            cts.Cancel();  // 如果Count方法已返回,Cancel没有任何效果
            // Cancel立即返回,方法从这里继续运行

            Console.ReadLine();
        }

        private static void Count(CancellationToken token, Int32 countTo)
        {
            for (Int32 count = 0; count < countTo; count++)
            {
                //判断是否接收到了取消任务的信号
                if (token.IsCancellationRequested)
                {
                    Console.WriteLine("Count is cancelled");
                    break; // 退出循环以停止操作
                }

                Console.WriteLine(count);
                Thread.Sleep(200);   // 出于演示浪费一点时间
            }
            Console.WriteLine("Count is done");
        }
}

如果要执行一个不允许被取消的操作,可以向该操作传递一个通过调用CancellationToken的静态None属性而返回的CancellationToken(特殊的实例, 它不与任何CancellationTokenSource关联,实例的私有字段为null)。 由于没有CancellationTokenSource, 所以没有代码能调用Cancel

如果愿意,可以调用Register方法登记一个或多个在取消一个CancellationTokenSource时调用的方法。每个回调方法都用CancellactionTokenRegister方法来登记的。要向这个方法传递一个Action<Object>委托;一个要通过委托传给回调的状态;以及一个Boolean值(名为useSynchronizationContext),该值指定了是否要使用调用线程的SynchronizationContext(同步上下文)来调用委托。

  • 如果为useSynchronizationContext参数传递的是false,那么调用Cancel的线程会顺序调用已登记的所有方法(post)。
  • 如果为useSynchronizationContext参数传递的是true,那么回调(方法)会被send(而不是post) 给已捕捉的SynchronizationContext对象,由同步上下文对象决定由哪个线程调用回调方法。

如果执行send操作,要等到目标线程那里处理完毕之后才会返回。再次期间,调用线程会被阻塞。这相当于同步调用。而如果执行post操作,是指将东西post到一个队列中便完事,调用线程可以立即返回。相当于异步调用。

向被取消的CancellationTokenSource登记一个回调方法, 将由调用Register的线程调用回调方法. 如果为useSynchronizationContext参数传递了true值, 就可能要通过调用线程的SynchronizationContext(同步上下文进行).

如果多次调用Regiser,那么多个回调方法都会调用。这些回调方法可能抛出未处理的异常。

如果调用CancellationTokenSourceCancel方法,

  • 向它传递true,那么抛出了未处理异常的第一个回调方法会阻止其他回调方法的执行,抛出的异常也会从Cancel中抛出。
  • 如果调用Cancel并向它传递false,那么登记的所有回调方法都会调用。所有未处理的异常都会添加到一个集合中。所有回调方法都执行后,如果其中任何一个抛出一个未处理的异常,Cancel就会抛出一个AggregateException,该异常实例的InnerException属性会被设为已抛出的所有异常对象的一个集合。如果以登记的所有回调方法都没有抛出异常,那么Cancel直接返回,不抛出任何异常。

没有办法将AggregateExceptionInnerException集合中的一个异常对象和特定操作对应起来; 你只知道某个操作出错, 并通过异常类型知道出了什么错. 要跟踪错误的具体位置, 需要检查异常对象的StackTrace属性, 并手动扫描你的源代码.

CancellactionTokenRegister方法返回一个CancellationTokenRegistration,如下所示:

public struct CancellationTokenRegistration : IEquatable<CancellationTokenRegistration>, IDisposable
{
        public void Dispose();
        .......
}

可调用Dispose从关联的CancellationTokenSource中删除一个已登记的回调方法;这样一来,在调用Cancel时,便不会再调用这个回调。以下代码演示了如何向一个CancellationTokenSource登记两个回调:

private static void Register()
{
       var cts = new CancellationTokenSource();
       // 向CancellationTokenSource注册 取消时的回调方法
       cts.Token.Register(() => Console.WriteLine("Canceled 1"));
       cts.Token.Register(() => Console.WriteLine("Canceled 2"));

       // 出于测试目的,让我们取消它,以便执行两个回调
       cts.Cancel();
}

//输出:
// Canceled 2
// Canceled 1

最后,可通过链接(CreateLinkedTokenSource)另一组的CancellationTokenSource来新建一个CancellationTokenSource对象。任何一个链接的CancellationTokenSource被取消,这个CancellationTokenSource对象就会被取消。以下代码对此进行的演示:

 class Program
{
        static void Main(string[] args)
        {
            // 创建一个 CancellationTokenSource
            var cts1 = new CancellationTokenSource();
            cts1.Token.Register(() => Console.WriteLine("cts1 canceled"));

            // 创建另一个 CancellationTokenSource
            var cts2 = new CancellationTokenSource();
            cts2.Token.Register(() => Console.WriteLine("cts2 canceled"));

            // 创建新的CancellationTokenSource,它在 cts1 o或 ct2 is 取消时取消
            var ctsLinked = CancellationTokenSource.CreateLinkedTokenSource(cts1.Token, cts2.Token);
            ctsLinked.Token.Register(() => Console.WriteLine("linkedCts canceled"));

            // 取消其中一个 CancellationTokenSource objects (这里选择了 cts2)
            cts2.Cancel();

            // 显示哪个 CancellationTokenSource objects 被取消 了
            Console.WriteLine("cts1 canceled={0}, cts2 canceled={1}, ctsLinked canceled ={2}",
               cts1.IsCancellationRequested, cts2.IsCancellationRequested, ctsLinked.IsCancellationRequested);

            Console.ReadLine();
        }

}

// 输出结果:
// linkedCts canceled
// cts2 canceled
// cts1 canceled=False, cts2 canceled=True, ctsLinked canceled =True

CancellationTokenSource 提供了在指定时间后自动取消的机制, 为了利用这个机制,

  • 要么用接收延时参数的构造器构造一个CancellationTokenSource对象
  • 要么调用CancellationTokenSourceCancelAfter方法.
public sealed class CancellationTokenSource : IDisposable  // 一个引用类型
{
   public CancellationToken(Int32 millisecondsDelay);
   public CancellationToken(TimeSpan delay);
   public void CancelAfter(Int32 millisecondsDelay);
   public void CancelAfter(TimeSpan delay);
   ...
}

任务

调用ThreadPoolQueueUserWorkItem方法来发起一次异步的受计算限制的操作是非常简单的。然而。这个技术存在许多限制。最大的问题是没有一个内建的机制让你知道操作在什么时候完成,也没有一个机制在操作完成时获得一个返回值。为了克服这些限制并解决一些其它问题,Microsoft引入了任务(Task) 的概念。我们通过System.Treading.Tasks命名空间中的类型来使用它们。

所以, 不是调用ThreadPool.QueueUserWorkItem方法, 而是用任务来做相同的事情.

// 旧的方式
ThreadPool.QueueUserWorkItem(ComputeBoundOp, 5);  // 调用QueueUserWorkItem
// 任务概念, 新的方式
new Task(ComputeBoundOp, 5).Start();  // 用Task来做相同的事情

// 等价写法
Task.Run(()=> ComputeBoundOp(5));

在上述代码中,创建了Task对象,并立即调用Start方法调度该任务方法。当然,也可以先创建好Task对象,以后在调用Start方法

创建一个Task的方式总是调用构造器,传递一个Action或者Action<Object>委托这个委托就是你想执行的操作。如果传递期待一个Object的方法,还必须向Task的构造器传递最终想传给操作的实参。还可以选择向Task的构造器传递一个CancellationToken ,这便允许Task在调度之前取消。

还可以选择向构造器传递一些TaskCreationOptions标志来控制Task的执行方式。TaskCreationOptions是一个枚举类型,定义了一组可按位OR到一起的标志。它的定义如下:

[FlagsAttribute, SerializableAttribute]
public enum TaskCreationOptions
{
        // 指定应使用默认行为
        None             = 0x0,

        // [提议] TaskScheduler(任务调度器) 以一种尽可能公平的方式安排任务,这意味着较早安排的任务将更可能较早运行,而较晚安排运行的任务将更可能较晚运行。
        // 造成默认的TaskScheduler(任务调度器) 将线程池中的任务放到全局队列中,而不是放到一个工作者线程的本地队列中
        PreferFairness   = 0x1,

        // [提议] TaskScheduler(任务调度器)应尽可能地创建线程池线程
        // 指定某个任务将是长时间运行、粗粒度的操作(最好是创建线程池线程)。
        LongRunning      = 0x2,

        // [提议总是被采纳]: 将一个任务和它的父Task关联。
        AttachedToParent = 0x4,

        // [提议总是被采纳]: 如果一个任务试图和这个父任务连接, 它就是一个普通任务,而不是子任务
        DenyChildAttach  = 0x8,

        // [提议总是被采纳]: 强迫子任务使用默认调度器而不是父任务的调度器
        HideScheduler    = 0x10
}

大多是标志只是一些提议而已,TaskScheduler在调度一个Task时,可能会也可能不会采纳这些提议。不过,AttacedToParent标志总是得到采纳,因为它和TaskScheduler本身无关。

等待任务完成并获取结果

对于任务,可以等待它完成,然后获取它们的结果。假定有一个Sum方法,在n值很大的时候,它要执行较长的时间:

private static Int32 Sum(Int32 n)
{
    Int32 sum = 0;
    for (; n > 0; n--)
      checked { sum += n; }    //如果n太大,这一行代码会抛出异常
    return sum;
}

现在可以构造一个Task<TResult>对象(派生自Task),并为泛型TResult参数传递计算限制操作的返回类型。在开始任务后,可以等待它完成并获取结果,如以下代码所示:

class Program
{
        static void Main(string[] args)
        {
           // 创建 Task, 现在还没开始运行
            Task<Int32> t = new Task<Int32>(n => Sum((Int32)n), 10000);

            // 启动任务
            t.Start();

            // 可以选择显式的等待任务完成
            t.Wait();
            // 可获得结果(Result属性内部会调用Wait)
            Console.WriteLine("The sum is: " + t.Result);   //一个Int32的值
            Console.ReadLine();
        }

        private static Int32 Sum(Int32 n)
        {
            Int32 sum = 0;
            for (; n > 0; n--) checked { sum += n; }    //如果n太大,这一行代码会抛出异常
            return sum;
        }
}

Result属性内部会调用Wait. 如果计算限制的任务抛出一个未处理的异常,异常会被”吞噬”并存储到一个集合中,而线程池线程允许返回到线程池中。调用Wait方法或者Result属性时,这些成员会抛出一个System.AggregateException对象。

一个线程调用Wait方法时,系统会检查系统要等待的Task是否已开始执行

  • 如果是,调用Wait的线程会阻塞,直到Task运行结束为止。

  • 如果Task 还没有开始执行,系统可能(取决于TaskSecheduler )使用调用Wait的线程来执行Task 。如果发生这种情况,

    那么调用Wait的线程不会阻塞;它会执行Task并立即返回。

    这样做的好处在于:

    • 没有线程会被阻塞,所以减少了资源的使用(因为不需要创建一个线程来替代被阻塞的线程),
    • 并提升了性能(因为不需要花时间创建一个线程,也没有上下文切换)。

    但是不好的地方在于:

    • 假如线程在调用Wait前已经获得了一个线程同步锁,而Task试图获取同一个锁,就会造成一个死锁的线程。

AggregateException类型封装了异常对象的一个集合(如果父任务生成了多个子任务,而多个子任务都抛出异常,这个集合便有可能包含多个异常)。该类型有一个InnerExceptions属性,它返回一个ReadOnlyCollection<Excepyion>对象。不要混淆InnerExceptions(有s)和InnerException(没有s)属性,后者是AggregateException类System.Exception基类继承来的。对于上例来说,AggregateExceptionInnerExceptions属性的元素0将引用由计算限制方法(Sum)抛出的实际System.OverflowException对象。

为方便编码,AggregateException重写了ExceptionGetBaseException方法。AggregateException的这个实现会返回作为问题根源的最内层的AggregateException

AggregateException还提供了一个Flatten方法,它创建一个新的AggregateException,其InnerExceptions属性包含一个异常列表,其中的异常是通过遍历原始AggregateException的内层异常层次结构而生成的。

最后,AggregateException还提供了一个Handle方法,它为AggregateException中包含的每个异常都调用一个回调方法,然后,回调方法可以为每个在调用Handle之后,如果至少有一个异常没有处理,就创建一个新的AggregateException对象,其中只包含未处理的异常,并抛出这个新的AggregateException对象。

除了等待单个任务, Task类还提供了两个静态方法, 允许线程等待一个Task对象数组.

其中Task的静态WaitAny方法会阻塞调用线程, 直到数组中的任何一个Task对象完成. 方法返回Int32数组索引值, 指明完成的是哪个Task对象. 方法返回后, 线程被唤醒并继续运行, 如果发生超时, 方法返回-1. 如果WaitAny通过CancellationToken取消, 会抛出一个OpreationCanceledException异常.

类似的,Task类还提供了静态WaitAll方法,它阻塞调用线程,直到数组中所有的Task对象都完成。如果Task对象都完成,WaitAll方法返回true。如果发生超时,就返回false。如果WaitAll通过一个CancellationToken而取消,会抛出一个OpreationCanceledException

取消任务

可以用一个CancellationTokenSource取消一个Task。首先,我们必须修订前面的Sum方法,让它接受一个CancellationToken参数:

private static Int32 Sum(CancellationToken ct, Int32 n)
{
    Int32 sum = 0;
    for (; n > 0; n--)
    {
        // 定时检查操作是否已取消
        // 在取消标志引用的CancellationTokenSource上如果调用Cancel,
        // 下面这一行就抛出OpreationCanceledException
        ct.ThrowIfCancellationRequested();

        checked { sum += n; }  //如果n太大,这一行代码会抛出异常
    }
    return sum;
}

在上述代码中,在计算限制的循环中,我们通过调用CancellationTokenThrowIfCancellationRequested方法来定时检查操作是否已取消。这个方法和CancellationTokenIsCancellationRequested属性相似。如果CancellationTokenSource已经取消,ThrowIfCancellationRequested会抛出一个OpreationCanceledException

现在,我们向下面这样创建CancellationTokenSourceTask对象

static void Main(string[] args)
{
    CancellationTokenSource cts = new CancellationTokenSource();
    Task<Int32> t   = new Task<Int32>(() => Sum(cts.Token, 10000), cts.Token);

    t.Start();

    // 在之后的某个时间,取消CancellationTokenSource以取消Task
    cts.Cancel();

    try
    {
        // 如果任务已经取消,Result会抛出一个AggregateException
        Console.WriteLine("The sum is: " + t.Result); // An Int32 value
    }
    catch (AggregateException ae)
    {
        // 将任何OperationCanceledException对象都视为已处理
        // 其他任何异常都造成抛出一个新的AggregateException,其中
        // 只包含未处理的异常      
        ae.Handle(e => e is OperationCanceledException);

        // 所有的异常都处理好之后,执行下面这一行
        Console.WriteLine("Sum was canceled");
    }

    Console.ReadLine();
}

private static Int32 Sum(CancellationToken ct, Int32 n)
{
    Int32 sum = 0;
    for (; n > 0; n--)
    {
        // 定时检查操作是否已取消
        // 在取消标志引用的CancellationTokenSource上如果调用Cancel,
        // 下面这一行就抛出OpreationCanceledException
        ct.ThrowIfCancellationRequested();

        checked
        {
            sum += n;
        } //如果n太大,这一行代码会抛出异常
    }
    return sum;
}

创建一个Task时,可以将一个CancellationToken传给Task的构造器,从而将这个CancellationToken和该Task关联起来

  • 如果CancellationTokenTask调度前取消,Task会被取消,永远不会执行。

    • 一个任务还没开始就试图取消它,会抛出InvalidOperationException
  • 如果Task已经调度(通过调用Start方法),那么Task为了允许它的操作在执行期间取消,Task的代码必须显式支持取消 。

    • 调用静态的Run方法会自动创建Task对象并立即调用Start

遗憾的是,虽然Task对象关联了一个CancellationToken,但没有办法访问它。因此,必须通过某种方式,在Task的代码本身中获得用于创建Task对象的同一个CancellationToken。为了写这样的代码,最简单的方法就是使用一个lambda表达式,并将CancellationToken作为一个闭包变量”传递”(就像上例所示)。

任务完成时自动启动新任务

有更好的方式知道一个任务在什么时候结束运行。一个任务完成时,它可以启动另一个任务。下面重写了前面的代码,它不会阻塞线程:

// 创建 Task, 推迟启动它, 继续另一个任务
Task<Int32> t = new Task<Int32>(n => Sum((Int32) n), 10000);

// ContinueWith 返回一个 Task 但一般不再关心这个对象
Task cwt = t.ContinueWith(task => Console.WriteLine("The sum is: " + task.Result));

现在,当执行Sum的任务完成后,这个任务会启动另一个任务(也在某个线程池线程上)以显示结果。执行上述代码的线程不会进入阻塞状态并等待这个两个任务中的任何一个完成

注意ContinueWith会返回新的Task对象的一个引用(在上述代码中, 放入了cwt变量中)。当然,可以用这个Task对象调用各种成员(比如WaitResult,甚至ContinueWith),但你一般都是忽略这个Task对象,不把它的引用保存到一个变量中

Task对象内部包含了ContinueWith任务的一个集合。所以,实际上可以用一个Task对象来多次调用ContinueWith。任务完成时,所有ContinueWith任务都会进入线程池的队列中。

此外,调用ContinueWith时,可以传递对一组TaskContinuationOptions枚举值进行按位OR运行的结果。下面是TaskContinuationOptions类型的定义:

[System.FlagsAttribute, System.SerializableAttribute]
public enum TaskContinuationOptions
{
        // [默认]
        None                  = 0x00000,
        // [提议] 希望任务尽快运行
        PreferFairness        = 0x00001,
        // [提议] 应尽可能地创建线程池线程
        LongRunning           = 0x00002,
        // [提议总是被采纳] 将一个任务和它的父Task关联
        AttachedToParent      = 0x00004,
        // 任务和这个父任务链接将抛出异常
        DenyChildAttach       = 0x00008,
        // 强迫子任务使用默认调度器而不是父任务的调度器
        HideScheduler         = 0x00010,
        // 除非前置任务完成,否则禁止延续任务完成(取消)
        LazyCancellation      = 0x00020,


        // 这些标志指出在什么情况下运行ContinueWith任务
        // 指定不应在延续任务前面的任务已完成运行的情况下安排延续任务。 此选项对多任务延续无效。
        NotOnRanToCompletion  = 0x10000,
        //指定不应在延续任务前面的任务引发了未处理异常的情况下安排延续任务。 此选项对多任务延续无效。
        NotOnFaulted          = 0x20000,
        //指定不应在延续任务前面的任务已取消的情况下安排延续任务。 此选项对多任务延续无效。
        NotOnCanceled         = 0x40000,

        // 这些标志是以上三种标志的便利组合
        //指定只应在延续任务前面的任务已完成运行的情况下才安排延续任务。 此选项对多任务延续无效。
        OnlyOnRanToCompletion = NotOnRanToCompletion|NotOnFaulted,
        //指定只有在延续任务前面的任务引发了未处理异常的情况下才应安排延续任务。 此选项对多任务延续无效。
        OnlyOnFaulted         = NotOnRanToCompletion|NotOnCanceled,
        //指定只应在延续任务前面的任务已顺利完成(中途没有取消,也没有抛出未处理异常)的情况下安排延续任务。此选项对多任务延续无效。
        OnlyOnCanceled        = NotOnFaulted|NotOnCanceled,

        //指定应同步执行延续任务。
        // 指定此选项后,延续任务将在导致前面的任务转换为其最终状态的相同线程上运行。 如果在创建延续任务时已经完成前面的任务,则延续任务将在创建此延续任务的线程上运行。 只应同步执行运行时间非常短的延续任务。
        //-------------------------------------------
        // 这个标志指出你希望由执行第一个任务的线程执行ContinueWith任务,
        // 第一个任务完成后调用
        // ContinueWith的线程接着执行ContinueWith任务
        // 这里是指同步执行, 两个任务在使用同一个线程一前一后的执行
        ExecuteSynchronously  = 0x80000,
}

默认情况下,如果没有指定上述任何标志,新任务无论如何都会执行下去,不管第一个任务是如何完成的。一个Task完成时,它的所有未运行(不满足前面说的各种条件)的延续任务都会自动取消。下面用一个例子演示所有这些概念。

CancellationTokenSource cts = new CancellationTokenSource();
Task<Int32>  t   = new Task<Int32>(() => Sum(cts.Token, 10000), cts.Token);
t.Start();

// 每个 ContinueWith 都返回一个 Task,但你不必关心这些Task对象
t.ContinueWith(task => Console.WriteLine("The sum is: " + task.Result),TaskContinuationOptions.OnlyOnRanToCompletion);
t.ContinueWith(task => Console.WriteLine("Sum threw: " + task.Exception),TaskContinuationOptions.OnlyOnFaulted);
t.ContinueWith(task => Console.WriteLine("Sum was canceled"),TaskContinuationOptions.OnlyOnCanceled);

// 如果注释这行代码, 则只输出 The sum is: 50005000
// 否则只输出: Sum was canceled
cts.Cancel();

任务可以启动子任务

任务支持父/子关系,如下代码所示:

static void Main(string[] args)
{
    Task<Int32[]> parent = new Task<Int32[]>(() =>
    {
        var results = new Int32[3]; // 创建数组来存储结果

        // 这个任务创建并启用了3个子任务
        new Task(() => results[0] = Sum(10000), TaskCreationOptions.AttachedToParent).Start();
        new Task(() => results[1] = Sum(20000), TaskCreationOptions.AttachedToParent).Start();
        new Task(() => results[2] = Sum(30000), TaskCreationOptions.AttachedToParent).Start();

        // 返回对数组的一个引用(即使数组元素可能还没有初始化)
        return results;
    });

    // 父任务及其子任务运行完成后, 用一个延续任务显示结果
    var cwt = parent.ContinueWith(parentTask => Array.ForEach(parentTask.Result, Console.WriteLine));
    // 启动父任务, 便于启动它的子任务
    parent.Start();
    Console.ReadLine();
}

在本例子中,父任务创建并启用3个Task对象。默认情况下,一个任务创建的Task对象是顶级任务,这些任务与创建它们的那个任务无关。然而,TaskContinuationOptions.AttachedToParent 标志将一个Task和创建它的那个Task关联起来结果是除非所有子任务结束运行,否则父任务不会认为已经结束。调用ContinueWith方法创建一个Task时,可以指定TaskContinuationOptions. AttachedToParent 标志将延续任务指定的一个子任务, 用于控制台循环输出结果。

任务内部揭秘

每个Task对象都有一组构成任务状态的字段。

  • 有一个Int32 ID(Task的只读Id属性)

    创建一个Task对象时,字段会被初始化为零。第一次查询Task的只读ID属性,属性将一个唯一Int32值分配给该字段,并从属性中返回它。TaskID从1开始,每分配一个ID都会递增1.

  • 代表Task执行状态的一个Int32

  • 对父任务的一个引用、

  • Task创建时指定的TaskScheduler的一个引用、

  • 回调方法的一个引用、

  • 对要传给回调方法的对象的一个引用(可通过Task的只读AsynState属性查询)、

  • 对一个ExecutionContext(执行上下文)的引用

  • 以及对一个ManualResetEventSlim对象的引用。

  • 除此之外,每个Task对象都有对根据需要创建的补充状态的一个引用。

    • 包含一个CancellationToken
    • 一个ContinueWithTask对象集合
    • 为抛出未处理异常的子任务而准备的一个Task对象集合等.

如果不需要任务提供的附加功能,那么使用ThreadPool.QueueUserWorkItem,资源的使用效率上会更高一些。

TaskTask<TResult>类实现了IDisposable接口,允许你在用完Task对象后调用Dispose。如今,所有Dispose方法所做的都是关闭ManuaResetEventSlim对象

然而,可以定义从TaskTask<Result>派生的类,在这些类中分配它们自己的资源,并在它们重写的Dispose方法中释放这些资源。当然,大多数开发人员都不会在自己的代码中显式的为一个Task对象调用Dispose;他们只让垃圾回收器回收任何不再需要的资源。

运行一个任务的代码时,可以查询Task静态CurrenId属性,它返回一个可空的Int32(Int32?)。还可以在调式期间,在Vasul Studio的”监视”或”即时”窗口中调用它,以便获得当前正在调试的代码的ID。然后,可以在”并行任务”和”并行堆栈”窗口中找到自己的任务。如果当前没有任务正在执行,查询CurrenId属性会返回null

一个Task对象存在期间,可查询Task的只读Status属性了解它在其生存期的什么位置。这个属性返回一个TaskStatus值,定义如下:

public enum TaskStatus
{
    //这些标志指出了一个Task在其生命周期内的状态
    // 任务已显式创建,可以手动Start()这个任务
    Created,

    // 任务已隐式创建,会自动开始
    WaitingForActivation,

    // 任务已调度,但尚未运行
    WaitingToRun,

    // 任务正在运行
    Running,

    // 任务正在等待它的子任务完成,子任务完成后它才完成
    WaitingForChildrenToComplete,

    // 一个任务的最终状态是以下三种之一
    // 已成功完成执行的任务
    RanToCompletion,
    // 该任务已通过对其自身的 CancellationToken引发 OperationCanceledException
    // 对取消进行了确认,此时该标记处于已发送信号状态;
    // 或者在该任务开始执行之前,已向该任务的 CancellationToken 发出了信号
    Canceled,
    // 由于未处理异常的原因而完成的任务
    Faulted
}
  • 首先构造一个Task对象时,它的状态是Created
  • 以后,任务启动时,它的状态变为WaitngToRun
  • Task在一个线程上运行时,它的状态就变成了Running
  • 任务停止运行,并等待它的任何子任务时,状态变成WaitingForChildrenToComplete
  • 任务完全结束时,它会进入以下三种状态的一种:RanToCompletion(运行完成)、Canceled(取消)或Faulted(出错)。
  • 一个Task运行完成时,可通过Task<TResult>Result属性来查询任务的结果。
  • 一个Task或者Task出错时,可以查询TaskException属性来获得任务抛出的未处理的异常:该属性总是返回一个AggregateException对象,对象的InnerExceptions集合包含了所有未处理的异常。

为简化编码,Task提供了几个只读的Boolean属性:IsCanceledIsFaultedIsCompleted注意,当Task处于RanToCompletedCanceled或者Faulted状态时,IsCompleted返回true。为了判断一个Task是否成功完成,最简单的办法就是使用如下所示的代码:

if (task.Status == TaskStatus.RanToCompleted )

如果Task是通过调用以下某个函数来创建的,这个Task对象就处于WaitingForActivation状态:

  • ContinueWith
  • ContinueWithAll
  • ContinueWithAny或者FromAsync等方法来创建。

如果通过构造一个TaskCompletionSource<TResult>(任务基础结构)对象[表示未绑定到委托的 Task<TResult> 的制造者方,并通过 Task 属性提供对使用者方的访问]创建一个Task,该Task在创建时也处于WaitingForActivation状态。这个状态意味着该Task的调度由任务基础结构控制

例如,不能显式启动一个通过ContinueWith创建的对象。这个Task会在它的前置任务(antecedent task) 执行完毕后自动开始。

任务工厂

有的时候需要创建一组共享相同配置的Task对象。为了避免机械地将相同的参数传给每一个Task的构造器,可以创建一个任务工厂来封装通用的配置,System.Threding.Tasks命名空间定义了一个TaskFactory类型和一个TaskFactory<TResult>类型。两个类型都派生自System.Object;也就是说,它们是平级的。

以下实例代码演示了如何使用一个TaskFatory:

    class Program
    {
        static void Main(string[] args)
        {
            // 创建一个父Task
            Task parent = new Task(() =>
            {
                var cts = new CancellationTokenSource();
                var tf = new TaskFactory<Int32> // 创建一个任务工厂, 通用设置如下
                    (
                    cts.Token,  // 传入CancellationTokenSource任务基础结构控制的CancellationToken对象
                    TaskCreationOptions.AttachedToParent,         // 由TaskFactory创建的任务都视为其 子任务
                    TaskContinuationOptions.ExecuteSynchronously, // 指定由TaskFactory创建的任务都以同步方式执行
                    TaskScheduler.Default // 都使用默认调度器
                    );

                // 创建并启动3个子任务
                // 每个Task对象都共享相同的CancellationTokenSource标记
                // 任务都被视为parent的子任务
                var childTasks = new[]
                {
                    // 通过任务工厂的StartNew方法创建任务并启动
                    tf.StartNew(() => Sum(cts.Token, 10000)),
                    tf.StartNew(() => Sum(cts.Token, 20000)),
                    tf.StartNew(() => Sum(cts.Token, Int32.MaxValue)) // 将最大值传入,计算之后会超出,抛异常
                };

                // 通过循环告诉每个子任务, 如果抛出未处理异常,
                // 就取消其他仍在运行的所有子任务
                for (Int32 task = 0; task < childTasks.Length; task++)
                    childTasks[task].ContinueWith // 完成指定任务并达到OnlyOnFaulted条件之后, 进行后续的特定任务
                        (
                        t => cts.Cancel(), // 特定任务: 取消其他仍在运行的所有子任务
                        TaskContinuationOptions.OnlyOnFaulted   // 条件: 指定任务完成失败
                        );

                // ContinueWhenAll当所有子任务完成后,启动一个特定任务,
                // 由于是通过任务工厂创建的, 需要覆盖CancellationToken.None来覆盖它的父子任务设定
                // 使该任务在取消父任务时, 不作为子任务一起取消.
                // 当处理所有结果的任务完成后, 创建另一个后续任务来显示从所有子任务中返回的最大值
                tf.ContinueWhenAll(  
                        childTasks,  // 传入子任务数组
                        completedTasks =>
                            completedTasks.Where(t => t.Status == TaskStatus.RanToCompletion).Max(t => t.Result),
                        CancellationToken.None) // 子任务都完成(没有异常和取消)后, 获取返回最大值
                    .ContinueWith(
                        t => Console.WriteLine("最大的是: " + t.Result), // 将最大值输出的后续任务
                        TaskContinuationOptions.ExecuteSynchronously  // 设定为同步执行延续任务
                        ).Wait(); // 等待只是为了测试
            });

            // 子任务完成后, 也显示任何未处理的异常
            parent.ContinueWith(p =>
            {
                // 我将所有文本放到一个StringBuilder中, 并只调用Console.WriteLine一次
                // 因为这个任务可能和上面的任务并行执行, 而我不希望任务的输出变的不连续
                StringBuilder sb = new StringBuilder("发生了以下异常:" + Environment.NewLine);
                foreach (var e in p.Exception.Flatten().InnerExceptions)
                    sb.AppendLine("   " + e.GetType().ToString());
                Console.WriteLine(sb.ToString());
            }, TaskContinuationOptions.OnlyOnFaulted);

            // 启动父任务, 使它能启动子任务
            parent.Start();

            try
            {
                parent.Wait(); // 出于测试目的
            }
            catch (AggregateException)
            {
            }

    //            最大的是: 200010000
    //            发生了以下异常:
    //            System.OverflowException
        }
    }

调用TaskFactoryTaskFactory<TResult>的静态ContinueWhenAllContinueWhenAny方法时, 以下TaskContinuationOption标志是不允许的:

  • NotOnRanToCompletion
  • NotOnFaulted
  • NotOnCanceled
  • OnlyOnFaulted
  • OnlyOnCanceled
  • OnlyOnRanToCompletion

也就是说, 无论前置任务是如何完成的, ContinueWhenAllContinueWhenAny方法都会执行延续任务.

任务调度器

任务基础结构是很灵活的,其中TaskScheduler对象功不可没。TaskScheduler对象负责执行调度的任务,同时向Visual Studio 调试器公开任务信息。FCL提供了两个派生自TaskScheduler的类型:

  • 线程池任务调度器(thread pool task scheduler)
  • 同步上下文任务调度器(synchronization context task scheduler)

默认情况下,所有应用程序使用的都是线程池任务调度器。这个任务调度器将任务调度给线程池的工作者线程,将在后面进行更详细的讨论。可以查询TaskScheduler的静态Default属性来获得对默认任务调度器的一个引用

同步上下文任务调度器通常用于Windows窗体、WPF和Silverlight应用程序。这个任务调度器将所有任务都调度给应用程序的GUI线程,使所有任务代码都能成功更新UI,比如按钮。菜单项等。同步上下文任务调度器不使用线程池。可以查询TaskSchedulerFromCurrentSynchronizationContext方法来获取对一个同步上下文任务调度器的引用

private sealed class MyForm : System.Windows.Forms.Form
    {
        private readonly TaskScheduler m_syncContextTaskScheduler;

        public MyForm()
        {
            // 获得一个堆上下文任务调度器的引用
            m_syncContextTaskScheduler = TaskScheduler.FromCurrentSynchronizationContext();

            Text    = "Synchronization Context Task Scheduler Demo";
            Visible = true;
            Width   = 400;
            Height  = 100;
        }

        private CancellationTokenSource m_cts;

        protected override void OnMouseClick(System.Windows.Forms.MouseEventArgs e)
        {
            if (m_cts != null) // 如果有操作正在进行, 取消它
            {
                m_cts.Cancel();
                m_cts = null;
            }
            else // 操作还没开始,启动它
            {

                Text  = "Operation running";
                m_cts = new CancellationTokenSource();

                // 这个任务使用默认任务调度器, 在一个线程池上执行
                Task<Int32> t = Task.Run(() => Sum(m_cts.Token, 20000), m_cts.Token);

                // 这些任务使用同步上下文任务调度器, 在GUI线程上执行
                t.ContinueWith(task => Text = "Result: " + task.Result,
                    CancellationToken.None, TaskContinuationOptions.OnlyOnRanToCompletion,
                    m_syncContextTaskScheduler);

                t.ContinueWith(task => Text = "Operation canceled",
                    CancellationToken.None, TaskContinuationOptions.OnlyOnCanceled,
                    m_syncContextTaskScheduler);

                t.ContinueWith(task => Text = "Operation faulted",
                    CancellationToken.None, TaskContinuationOptions.OnlyOnFaulted,
                    m_syncContextTaskScheduler);
            }

            base.OnMouseClick(e);
        }
    }

获得一个对同步上下文任务调度器的引用, 然后在调用ContinueWith延续任务时, 使用此调度器.

m_syncContextTaskScheduler = TaskScheduler.FromCurrentSynchronizationContext();

**使用线程池很好, 因为GUI线程在此期间不会被阻塞, 能响应其他UI操作, 但线程池线程执行的代码不应尝试更新UI组件, 否则会抛出InvalidOperationException。**计算限制的任务完成后,执行3个延续任务, 它们由GUI线程对应的同步上下文任务器来调度, 任务调度器将任务放到GUI线程的队列中, 使它们的代码能成功更新UI组件, 所有任务都通过继承的Text属性来更新窗体的标题。

由于计算限制的工作(Sum)在线程池上运行, 所以用户可以和UI交互来取消操作

当然, 如果有特殊的任务调度需求, 完全可以定义自己的TaskScheduler派生类. Microsoft在Parallel Extensions Extras包中提供了大量和任务有关的示例代码.

Parallel的静态For,ForEach和Invoke方法

在一些常见的编程情形中,使用任务也许会提升性能。为了简化编程,静态类System.Threading.Tasks.Paraller封装了这些常见的情形,它内部使用Task对象

例如,不要像下面一样处理一个集合中的所有项:

// 不建议写法
// 一个线程顺序执行这个工作(每次迭代调用一次DoWork)
for (Int32 i = 0; i< 1000; i++ ) DoWork(i);

相反,可以使用Parallel类型的For方法,让多个线程池线程辅助完成这个工作:

// 建议写法
// 线程池的线程并行处理工作
Parallel.For(0, 1000, i => DoWork(i));

类似的,如果有一个集合,那么不要像下面这样写:

// 不建议写法
// 一个线程顺序执行这个工作(每次迭代调用一次DoWork)
foreach ( var item in conllection) DoWork(item);

// 建议写法
// 线程池的线程并行处理工作
Parallel.ForEach(conllection,item=>DoWork(item));

如果代码中既可以用For,也可以用ForEach,那么建议使用For,因为它执行的快一点。最后,如果要执行几个方法,那么可以顺序执行它们,如下所示:

//  一个线程顺序执行所有方法    
Method1();
Method2();
Method3();

// 线程池的线程并行执行
parallel.Invoke(
            () => Method1(),
            () => Method2(),
            () => Method3());

Parallel的所有方法都让调用线程参与处理。从资源利用的角度说,这是一件好事,因为我们不希望调用线程停下来(阻塞),等待线程池做完所有工作后才继续。然而,如果调用线程在线程池完成自己的那一部分工作之前完成工作,调用程序就会将自己挂起,直到所有工作完成。如果任何操作抛出一个未处理的异常,你调用的paraller方法最后会抛出一个AggregateException

调用Parallel的方法时,有一个前提条件务必记住:工作项要能并行执行。因此,如果工作项必须顺序执行,就不要调用Parallel的方法。另外,要避免会修改任何共享数据的工作项,因为多个线程同时处理的数据可能损坏。为了解决这个问题,一般的方法就是围绕数据访问添加线程同步锁。但是这样一来,一次就只能有一个线程访问数据,无法享受并行处理多个想带来的好处。

Parallel的方法本身也有开销:委托对象必须分配,而针对每一个工作项,都要调用一次这些委托。如果有大量可由多个线程处理的工作项,那么也许会获得性能的提升。

rallelForForEachInvoke方法都能接受一个ParallelOptions对象的重载版本。这个对象的定义如下:

     // 存储用于配置 Parallel 类的方法的操作的选项。
     public class ParallelOptions
     {
        // 初始化 ParallelOptions 类的新实例
        public ParallelOptions();

        // 允许取消操作, 默认None
        public CancellationToken CancellationToken { get; set; }

        // 允许指定可以并发操作的最大工作项数目,默认为-1(可用CPU数)
        public int MaxDegreeOfParallelism { get; set; }

        // 指定调度器, 默认为TaskScheduler.Default
        public TaskScheduler TaskScheduler { get; set; }
     }

除此之外,ForForEach方法有一些重载版本允许传递3个委托:

  • 任务局部初始化委托(localInit),为参与工作的每一个任务都调用一次委托。这个委托是在任务被要求处理一个工作项之前调用。
  • 主体委托(body),为参与工作的各个线程所处理的每一项都调用一次委托。
  • 任务局部终结委托(localFinally),为参与工作的每一个任务都调用一次委托。这个委托是在任务处理好派遣给它的所有工作之后调用。即使主体委托引发一个未处理的异常,也会调用它。

以下示例代码演示了如何利用3个委托,计算一个目录中的所有文件的字节长度总计值:

        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, parallelLoopState, 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;
        }

每个任务都通过taskLocalTotal变量为分配给它的文件维护自己的总计值。每个任务完成工作之后,都调用Interlocked.Add方法[对两个 32 位整数进行求和并用和替换第一个整数],以一种线程安全的方式更新总的总计值

只有在每个任务返回之后,masterTotal才需要以一种线程安全的方式更新materTotal变量。所以,因为调用Interlocked.Add方法而造成的性能损失每个任务只发生一次,而不会每个工作项都发生。

注意,我们向主体委托传递一个ParallelLoopState对象,它的定义如下:

    // 可用来使 Parallel 循环的迭代与其他迭代交互
    public class ParallelLoopState
    {
        // 获取循环的任何迭代是否已引发相应迭代未处理的异常
        public bool IsExceptional { get; }
        // 获取循环的任何迭代是否已调用 Stop
        public bool IsStopped { get; }
        // 获取从中调用 Break 的最低循环迭代。
        public long? LowestBreakIteration { get; }
        // 获取循环的当前迭代是否应基于此迭代或其他迭代发出的请求退出。
        public bool ShouldExitCurrentIteration { get; }

        // 告知 Parallel 循环应在系统方便的时候尽早停止执行当前迭代之外的迭代。
        public void Break();
        // 告知 Parallel 循环应在系统方便的时候尽早停止执行。
        public void Stop();
     }

参与工作的每一个任务都会获得它自己的ParallelState对象,并可通过这个对象和参与工作的其他任务进行交互。

  • Stop方法告诉循环停止处理任何更多的工作,未来对IsStopped属性的查询会返回true。
  • Break方法告诉循环不再继续处理当前项之后的项。
  • LowestBreakIteration属性返回在处理过程中调用过Break方法的最低的项。从来没有调用过Break,LowestBreakIteration会返回null.

处理任何一项时,如果造成一个未处理的异常,IsExceptional属性会返回true。如果处理一项时会花费大量的时间,代码可查询ShouldExitCurrentIteration属性看它是否应该提前退出。如果调用过Stop,调用过Break,取消过CancellationTokenSource,或者处理一项时造成了未处理的异常,这个属性就会返回true。

Parallel的For和ForEach方法都返回一个ParallelLoopResult实例,他看起来像下面这样:

    // 可以根据以下两属性判断某个线程调用Break方法还是Stop方法
    public struct ParallelLoopResult
    {
        // 如果操作提前终止, 以下方法返回false
        public bool IsCompleted { get; }

        // 获取从中调用 System.Threading.Tasks.ParallelLoopState.Break() 的最低迭代的索引。
        public long? LowestBreakIteration { get; }
    }

可通过检查属性来了解循环的结果:

  • 如果IsCompleted返回true。表明循环运行完成,所有项都得到了处理。
  • 如果IsCompleted为false,而且LowestBreakIteration为null,表明参与工作的某个线程调用了Stop方法。
  • 如果LowestBreakIteration返回false,而且LowestBreakIteration不为null,表名参与工作的某个线程调用的Break方法,
  • LowestBreakIteration`返回的Int64值指明了保证已得到处理的最低一项的索引。
  • 如果抛出异常, 应捕捉AggregateException来得体地恢复.

并行语言集成查询 PLINQ

使用LINQ to Object时,只有一个线程顺序处理数据集合中的所有项;我们称为顺序查询。为了提高处理性能,可以使用并行LINQ(Parallel LINQ)它将顺序查询转换成为一个并行查询,在内部使用任务(这些任务的排列由默认TaskScheduler来调度),将集合中的数据项的处理工作分散到多个CPU上,一边并发处理多个数据。

静态System.Linq.ParallelEnumerable类(在System.Core.dll中定义)实现了PLINQ的所有功能,所以必须通过C#的using指令将System.Linq命名空间导入到你的源码中。尤其是,这个类公开了所有标准LINQ操作符的并行版本,比如WhereSelectSelectManyGroupByJoinSkipTask等。所有这些方法都是扩展了System.Linq.ParallelQuery<T>类型的扩展方法。

为了让自己的LINQ to Object查询调用这些方法的并行版本,必须将自己的顺序查询(基于IEnumberable或者IEnumerable<T>)转换成并行查询(基于ParallelQuery或者ParallelQuery<T>),这是用ParallelEnumerableAsParallel扩展方法来实现的,如下所示:

public static ParallelQuery<TSource> AsParallel<TSource>(this IEnumerable<TSource> source)

下面是将一个顺序查询转换成并行查询的例子。查询返回的是一个程序集中定义的所有过时(obsolete)方法。

public class MyClass
 {
     [Obsolete("过时函数不应使用", false)]
     public void testobsolete(bool a)
     {

     }
 }

 class Program
 {
     static void Main(string[] args)
     {
         ObsoleteMethods(typeof(MyClass).Assembly);
     }

     private static void ObsoleteMethods(Assembly assembly)
     {
         var query =
             from type in assembly.GetExportedTypes().AsParallel() // 将顺序查询转换成并行查询
             from method in type.GetMethods(BindingFlags.Public | BindingFlags.Instance | BindingFlags.Static)
             let obsoleteAttrType = typeof(ObsoleteAttribute)
             where Attribute.IsDefined(method, obsoleteAttrType)
             orderby type.FullName
             let obsoleteAttrObj = (ObsoleteAttribute) Attribute.GetCustomAttribute(method, obsoleteAttrType)
             select String.Format("Type={0}\nMethod={1}\nMessage={2}\n", type.FullName, method.ToString(), obsoleteAttrObj.Message);

         // 显示结果
         foreach (var result in query)
             Console.WriteLine(result);
// 控制台输出:         
// Type    = ConsoleApplication.MyClass
// Method  = Void testobsolete(Boolean)
// Message = 过时函数不应使用

     }
 }

在一个查询中,可以从执行并行操作切换回执行顺序操作,这是通过调用ParallelEnumerableAsSequential方法做到的。这个方法将一个ParallQuery<T>转换回一个IEnumerable<T>。这样一来,在调用了AsDequential之后执行的操作将只由一个线程执行。

public static IEnumerable <TSource> AsSequential<TSource>(this ParallelQuery<TSource> source)

通常,一个LINQ查询的结果数是让某个线程执行一个foreach语句计算获得的。这意味着只有一个线程遍历查询的所有结果。如果希望以并行的方式处理查询的结果,就应该使用ParallelEnumerableForAll方法处理查询:

static void ForAll<TSource>(this ParallelQuery<TSource> source,Action<TSource> action)

这个方法允许多个线程同时处理结果,可以修改前面的代码来使用该方法:

// (单线程对结果进行操作)显示结果
foreach (var result in query)
    Console.WriteLine(result);

//(多线程同时处理结果)显示结果
query.ForAll(Console.WriteLine);

然而,让多个线程同时调用Console.WriteLine反而会损害性能,因为Console类内部会对线程进行同步,确保每次只有一个线程 能访问控制台程序窗口,避免来自多个线程的文本最后显示成一团乱麻。希望为每个结果都执行计算时,才使用ForAll方法

由于PLINQ可用多个线程处理数据项,所以数据项被并发处理,结果被无序返回。如果需要让PLINQ保存数据项的顺序,可调用ParallelEnumerableAsOrderd方法。调用这个方法时,线程会成组处理数据项。然后,这些数据项被合并回去,以保持顺序。这样会损害到性能。

以下操作符生成不排序的操作:DistinctExceptIntersectUnionJoinGroupByGroupJoinToLookup。在这些操作符之后,如果想再次强制排序,只需调用AsOrdered方法。

以下操作符生成排序的操作:OrderByOrderByDescendingThenbyThenByDescending。在这些出操作符之后,如果想再次恢复不排序的处理,只需调用AsUnordered方法。

PLINQ提供了一些额外的ParallelEnumerable方法,可调用它们来控制查询的处理方式:

// 设置要与查询关联的 CancellationToken
public static ParallelQuery<TSource> WithCancellation<TSource>(this ParallelQuery<TSource> source, CancellationToken cancellationToken);

// 设置要在查询中使用的并行度。 并行度是将用于处理查询的同时执行的任务的最大数目。
public static ParallelQuery<TSource> WithDegreeOfParallelism<TSource>(this ParallelQuery<TSource> source, int degreeOfParallelism);

// 设置查询的执行模式。
public static ParallelQuery<TSource> WithExecutionMode<TSource>(this ParallelQuery<TSource> source, ParallelExecutionMode executionMode);

//设置此查询的合并选项,它指定查询对输出进行缓冲处理的方式。
public static ParallelQuery<TSource> WithMergeOptions<TSource>(this ParallelQuery<TSource> source, ParallelMergeOptions mergeOptions);

WithCancellation方法允许传递一个CancellationToken,使查询处理能提前停止。

WithDegreeOfParallelism方法指定最多允许多少个线程处理查询;他不会强迫创建满全部线程,如果并不是全部都需要的话。默认情况下,会为每个内核用一个线程来执行查询。但如果想空出一些内核做其他工作, 可调用WithDegreeOfParallelism并传递小于可用内核数的一个数字。

PLINQ分析一个查询,然后决定如何最好地处理它。有的时候,顺序处理一个查询可以获得更好的性能,尤其在使用以下任何操作时:ConcatElementAt(OrDefault)First(OrDefault)Last(OrDefault)Skip(While)Task(While)Zip。使用Select(Many)Where的重载版本,并向你的selectorpredicate委托传递一个位置索引时也是如此。然而,可以调用WithExecutionMode,向它传递某个ParallelExecuteMode标志,从而强迫查询以并行方式处理:

public enum ParallelExecutionMode
{
    Default = 0,           // 让并行LINQ决定处理查询的最佳方式
    ForceParallelism = 1   // 强迫查询以其并行方式处理
}

如前所述,并行LINQ让多个线程处理数据项,结果必须再合并回去。可调用WithMergeOptions向它传递以下某个ParallelMargeOptions标志,从而控制这些结果的缓冲和合并方式:

    // 指定查询中要使用的输出合并的首选类型。 换言之,它指示 PLINQ 如何将多个分区的结果合并回单个结果序列。 这仅是一个提示,系统在并行处理所有查询时可能不会考虑这一点。
    public enum ParallelMergeOptions
    {
        //   使用默认合并类型,即 AutoBuffered。
        Default = 0,

        //  不利用输出缓冲区进行合并。 一旦计算出结果元素,就向查询使用者提供这些元素。
        NotBuffered = 1,

        //  利用系统选定大小的输出缓冲区进行合并。 在向查询使用者提供结果之前,会先将结果累计到输出缓冲区中。
        AutoBuffered = 2,

        //   利用整个输出缓冲区进行合并。 在向查询使用者提供任何结果之前,系统会先累计所有结果。
        FullyBuffered = 3,
    }

这些选项使你能在某种程度上控制速度和内存消耗的对应关系。最好亲自试验所有选项,并对比其性能,来选择那种方式。

  • NotBuffered 最省内存,但处理速度慢一些。
  • FullyBuffered 消耗较多内存,但运行得最快。
  • AutoBuffered 介于NotBufferedFullyBuffered 之间。

执行定时计算限制操作

System.Threading命名空间定义了一个Timer类,可用它让一个线程池线程定时调用一个方法。构造Timer类的一个实例相当于告诉线程池:在将来的某个时间会回调你的一个方法。Timer类提供了几个构造函数,相互都非常相似:

public sealed class Timer : MarshalByRefObject, IDisposable
{
        public Timer(TimerCallback callback, object state, int dueTime, int period);
        public Timer(TimerCallback callback, object state, long dueTime, long period);
        public Timer(TimerCallback callback, object state, TimeSpan dueTime, TimeSpan period);
        public Timer(TimerCallback callback, object state, uint dueTime, uint period);
}
  • callback参数标识希望由一个线程池线程回调的方法。当然,你写的对调方法必须和System.Threading.TimerCallback委托类型匹配,如下所示:

    delegate void TimerCallback(Object state);

  • state参数允许在每次调用回调方法时都像它传递状态数据;如果没有需要传递的状态数据,可以传递null。

  • dueTime参数告诉CLR在首次调用回调方法之前要等待多少毫秒。可以使用一个有符号或无符号的32位值,一个有符号的64位值或者一个TimeSpan值指定毫秒数。如果希望回调方法立即调用,为dueTime参数指定0即可。

  • period周期参数指定了以后每次调用回调方法需要等待的时间(毫秒)。如果为这个参数传递Timeout.Infinite(-1),线程池线程值调用回调方法一次

在内部,线程池为所有Timer对象只使用了一个线程。这个线程知道下一个Timer对象在什么时候到期。下一个Timer对象到期时,线程就会唤醒,在内部调用TreadPoolQueueUserWorkItem,将一个工作项添加到线程池队列中,使你的回调方法得到调用。

如果回调方法的执行时间很长,计时器可能(在上个回调还没有完成时)再次触发。这个能造成多个线程同时执行你的回调方法。为解决这个问题,我的建议是:构造Timer时,为period参数指定Timeout.Infinite。这样,计时器就只触发一次。然后,在你的回调方法中,调用Change方法指定一个新的dueTime,并再次为period参数指定Timeout.Infinite

以下是Change方法的各个重载版本.

public sealed class Timer : MarshalByRefObject, IDisposable
{
        public bool Change(int dueTime, int period);
        public bool Change(long dueTime, long period);
        public bool Change(TimeSpan dueTime, TimeSpan period);
        public bool Change(uint dueTime, uint period);
}

Timer类还提供了Dispose方法,允许完全取消计时器,并可在当时处于pending状态的所有回调完成之后,向notifyObject参数标识的内核对象发送信号。以下是Dispose方法的各个重载版本:

public sealed class Timer : MarshalByRefObject, IDisposable
{
       public void Dispose();
       public bool Dispose(WaitHandle notifyObject);
}

一个Timer对象被垃圾回收时,它的终结代码告诉线程池取消计时器,使它不再触发。所以,使用一个Timer对象时,要确定有一个变量在保持Timer对象的存活,否则对你的回调方法调用就会停止。

以下代码演示了如何让一个线程池线程立即开始调用一个回调方法,以后每2秒钟调用一次:

class Program
{
    private static Timer s_timer;

    static void Main(string[] args)
    {

        Console.WriteLine("每2秒检查一次状态");

        // 创建但不启动计时器
        // 确保s_timer在线程池线程调用 Status 之前引用该计时器
        s_timer = new Timer(Status, null, Timeout.Infinite, Timeout.Infinite);

        // 现在s_timer已被赋值, 可以启动计时器
        // 现在在Status中调用Change, 保证不会抛出NullReferenceException
        // (防止被垃圾回收, 确定有一个变量在保持Timer对象的存活)
        s_timer.Change(0, Timeout.Infinite);

        Console.ReadLine(); // 防止进程终止
    }

    // 这个方法签名必须匹配 TimerCallback 委托匹配
    private static void Status(Object state)
    {
        // 这个方法由一个线程池线程执行
        Console.WriteLine("In Status at {0}", DateTime.Now);
        Thread.Sleep(1000); // 模拟其他工作1秒钟

        // 返回前放Timer在2秒后再次触发
        s_timer.Change(2000, Timeout.Infinite);

        // 这个方法返回后, 线程回归池中, 等待下一个工作项
    }
}

// In Status at 2019/9/26 12:58:50
// In Status at 2019/9/26 12:58:53
// In Status at 2019/9/26 12:58:56
// In Status at 2019/9/26 12:58:59

如果需要定时执行的操作, 可以利用Task静态Delay方法和C#的asyncawait关键字.

class Program
{
     private static Timer s_timer;

     static void Main(string[] args)
     {
         Console.WriteLine("每2秒检查一次状态");
         Status();

         Console.ReadLine(); // 防止进程终止
     }

     // 这个方法可获取你想要的任何参数
     private static async void Status()
     {
         while (true)
         {
             Console.WriteLine("检查Status在 {0}", DateTime.Now);
             // 要检查的代码放到这里.

             // 在循环末尾, 在不阻塞线程的前提下延迟2秒
             await Task.Delay(2000); // await 允许线程返回
             // 2秒之后, 某个线程会在await之后介入并继续循环
         }
     }
}
// 检查Status在 2019/9/26 13:09:14
// 检查Status在 2019/9/26 13:09:16
// 检查Status在 2019/9/26 13:09:18
// 检查Status在 2019/9/26 13:09:20

在不阻塞线程的前提下延迟2秒await Task.Delay(2000); await 允许线程返回,2秒之后, 某个线程会在await之后介入并继续循环

计时器类的说明

FCL事实上提供了几个计时器,大多是开发人员都不清楚每个计时器到底有什么独到之处,在这里试着解释一下:

  • System.ThreadingTimer类

    这是刚刚讨论过的计时器。要在一个线程池线程上执行定时的(周期性发生的)后台任务,它是最好的计时器

  • System.Windows.FormsTimer类

    构造这个类的一个实例,相当于告诉Windows将一个计时器和调用线程关联。当这个计时器触发时,WIndows将一条计时器消息(WM_TIMER)注入线程的消息队列。线程必须执行一个消息泵来提取这些消息,并把它们派遣给想要的回调方法。注意,所有这些工作都只有一个线程完成——设置计时器的线程保证就是执行回调方法的线程。这还意味着你的计时器方法不会由多个线程并发执行

  • System.Windows.ThreadingDispatcherTimer类

    这个类是 System.Windows.FormsTimer类在Siverlight和WPF应用程序中的等价物。

  • Windows.UI.XamlDispatcherTimer

    这个类是System.Windows.FormsTimer类在Windows Store应用中的等价物。

  • System.TimersTimer类

    这个计时器基本是System.ThreadingTimer类的一个包装类当计时器到触发时,会导致CLR将事件放到线程池的队列中。尽量不要使用这个类而是使用System.Threading的Timer类。

线程池如何管理线程

随着CLR的每个版本的发布,其内部的实现已经发生了显著变化。未来版本还会继续变化。最好是将线程池看做一个黑盒。

设置线程池限制

CLR允许开发人员设置线程池要创建最大线程数。但实践证明,线程池永远都不该为池中的线程数设置上限,因为可能发生饥饿或死锁。假如队列中有1000个工作项,但这些工作项全都因为一个事件而阻塞,等1001个工作项发出信号才能解除阻塞。如果设置最大1000个线程,第1001个工作项就不会执行,所有1000个线程都会一直阻塞,最终用户被迫终止应用程序,并丢失他们都做的为保存的工作。

由于存在饥饿和死锁问题,CLR团队一直都在稳步地增加线程默认能够拥有的最大线程数。目前默认值是最大1000个线程。

System.Threading.ThreadPool类提供了几个静态方法,可调用它们设置和查询线程池的线程数:GetMaxThreadsSetMaxThreadsGetMinThreadsSetMinThreadsGetAvailableThreads[获得可用的线程数量]。强烈建议你不要调用上述任何方法。限制线程池的线程数,一般只会造成应用程序性能变得更差。

如何管理工作者线程

下图展示了构成线程池的一部分的工作者线程的各种数据结构。ThreadPool.QueueUserWorkItem方法和Timer类总是将工作项放到全局队列中。工作者线程采用一个先入先出算法将工作项从这个队列中取出来,并处理它们。

由于多个工作者线程可能同时从全局队列中拿走工作项,所以所有工作者线程都竞争一个线程同步锁以保证两个或多个线程不会获取同一个工作项。这个线程同步锁在某些应用程序中可能成为瓶颈,对伸缩性和性能造成某种程序的限制

现在,让我们谈谈使用默认TaskScheduler(通过查询TaskScheduler的静态Default属性来获得)来调度的Task对象。当一个非工作者线程调度一个Task时,Task会添加到全局队列中。

但是,每个工作者线程都有它自己的本地队列。当一个工作者线程调度一个Task时,Task会添加到调用线程的本地队列中。

一个工作者线程准备好处理一个工作项时,它总是先检查它的本地队列来查找一个Task。存在一个Task,工作者线程就从它的本地队列中移出Task,并对工作项进行处理。

要注意的是,工作者线程采用后入先出的算法将任务从它的本地队列中取出。由于工作者线程是唯一允许访问它自己的本地队列列头的线程,所以无需同步锁,而且在队列中添加和删除Task的速度非常快。这个行为的副作用在于,Task是按照和进入队列时相反的顺序执行的

线程池从来不保证排队中的工作项的处理顺序,这是合理的,尤其考虑到多线程可能同时处理工作项。然而,上述副作用使得这个问题变得更加恶化。你必须保证自己的应用程序对工作项或Task的执行顺序不做任何预设

如果一个工作者线程发现它的本地队列变空了,工作者线程就会尝试从另一个工作者线程的本例队列中”偷”一个Task。这个Task是从一个本地队列的尾部“偷”走的,并要求获得一个线程同步锁,这对性能可能有少许影响。当然,这种”偷窃”行为很少发生,所以很少需要获取这个锁。

如果所有本地队列都变空,那么工作者线程会使用FIFO(先进先出)算法,从全局队列中提取一个工作项。

如果全局队列也为空,那么线程就会进入睡眠状态,等待事情的发生。如果睡眠的时间太长,它会自己醒来,并销毁自己,允许系统回收线程使用的资源(包括内核对象、栈、TEB等)。

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

缓存线和伪共享(第四版没这节)

为了提升反复访问内存的性能,如今的CPU在芯片上都集成了高速缓存。线程首次从RAM去取一些值时,CPU从RAM获取所需的值,并把它存储到CPU的高速缓存中。事实上,为了进一步提升性能,CPU会在逻辑上将所有内存都划分为所谓的缓冲行(cache line)。一个缓冲行有64个字节构成,所以CPU从RAM中获取并存储64字节的块。如果应用程序需要读取一个Int32值,那么会获取包含了那个Intt32的64个字节,这样会获取到比需要的更多字节,这样通常会造成性能增强,因为大多数应用程序在访问了一些数据之后,通常会继续访问存储在那些数据周围的数据。由于相邻的数据已经提取到CPU的缓存中,就避免了慢速的RAM访问。

然而,如果两个或多个内核访问同一个缓冲行中的字节,内核必须相互通信,并在内核之间传递缓冲行,造成多个内核不能同时处理相邻的字节,这对性能会造成严重影响。

internal static class FalseSharing
{
#if true
    private class Data
    {
        // 这两个字段是相邻的,并(极有可能)在相同的缓冲行中
        public Int32 field1;
        public Int32 field2;
    }
#else
// 现在,让我们修改Data类,使它看起来项下面这样
// 这两个字段分开了,不再相同的缓冲行中
// 比上述写法运行速度上快一些
[StructLayout(LayoutKind.Explicit)]
private class Data {
  [FieldOffset(0)]
  public Int32 field1;
  [FieldOffset(64)]
  public Int32 field2;
}
#endif

    private const  Int32     iterations   = 100000000;
    private static Int32     s_operations = 2;
    private static Stopwatch s_stopwatch;

    public static void Go()
    {
        // 分配一个对象,并记录开始时间
        Data data = new Data();
        s_stopwatch = Stopwatch.StartNew();
        // // 让零个线程访问在对象中它们自己的字段
        ThreadPool.QueueUserWorkItem(o => AccessData(data, 0));
        ThreadPool.QueueUserWorkItem(o => AccessData(data, 1));
        Console.ReadLine();
    }

    private static void AccessData(Data data, Int32 field)
    {
        // 这里的线程各自访问它们在Data对象中自己的字段
        for (Int32 x = 0; x < iterations; x++)
            if (field == 0) data.field1++;
            else data.field2++;
        // 不管哪个线程最后结束,都显示它花的时间
        if (Interlocked.Decrement(ref s_operations) == 0)
            Console.WriteLine("Access time: {0}", s_stopwatch.Elapsed);// Elapsed经过的时间
    }
}

上述代码中,Data对象在构造时包含了两个字段。这两个字段极有可能在同一个缓冲行中。然后,两个线程池线程启动并执行AccessData方法。一个将1加到Data的filed1上的100 000 000 次,另一个线程对filed2字段做同样的事情。每个线程完成后,都递减s_operations字段中的值;最后一个将字段递减为0的线程就是最后一个结束的线程,它显示两个线程完成它们的工作总共发了多少时间。

在上述代码中,现在用一个缓存线(64字节)分隔两个字段。再次运行,比第一个版本快了一些。从程序角度看,两个线程处理的是不同的数据。但从CPU缓存线来看,CPU处理的是相同的数据。这称为伪共享(false sharing) 。在第二个版本中,字段在不同的缓存线上,所以CPU可以真正做到独立,不必共享什么。

通过上述讨论,应该知道在多个线程同时访问相邻的数据时,缓存线伪共享可能对应用程序产生严重影响。在性能非常紧要的情形下,这是你应该注意的一点。如果检查到这个问题,通常都可以设计出一种方式来避免它(这里用的就是FiledOffset attribute)。

要注意的是,数组在数组内存起始处维护着它的长度,具体位置是在前几个数据元素之后,访问一个数组元素时,CLR验证你使用的索引在数组的长度之内。这意味着访问一个数组总是牵涉到访问数组的长度。因此,为了避免产生额外的伪共享,应该避免让一个线程向数组的前几个元素写入,同时让其他线程访问数组中的其他元素。

Licensed under CC BY-NC-SA 4.0
0