不用任何与创建线程、资源互斥有关系的API写多线程程序
这次的例子，是一个很简单的控制台，她将面对瞬间提交的百万的数据，而面不改色（CPU、内存非常平稳），队列中始终只保存最新的数据，每次只处理cpu 个数据（我的机器是双核的，所以，在我这里，就是每个CPU一个线程，真正的并行运行哦....），OK不废话，进入正题：
呃，既然是实例，那么就直接看代码好了：
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using Microsoft.Ccr.Core;
namespace CCRDemo1
{
class Program
{
static void Main(string[] args)
{
int maxiQueueDepth = 10;
// step1: 创建一个Dispatcher对象
Dispatcher dispatcher = new Dispatcher(0, "调度器名称");
// step2: 创建一个与step1创建对象关联的DispatcherQueue对象
DispatcherQueue depthThrottledQueue = new DispatcherQueue(
"任务队列的名称",
// 关联到该队列的调度器
dispatcher,
// 队列保存数据的策略：保存最近消息策略
TaskExecutionPolicy.ConstrainQueueDepthDiscardTasks,
// 队列的深度
maxiQueueDepth
);
// step3: 创建一个能够接收整型数据的Port
Port<int> intPort = new Port<int>();
// step4: 把Port与处理函数关联，然后再与DispatcherQueue关联
Arbiter.Activate(depthThrottledQueue,
Arbiter.Receive(true,
intPort,
delegate(int i) // 这里用了一个匿名方法，作为处理函数
{
Thread.Sleep(2000);
Console.WriteLine("[{0}] {1}", DateTime.Now.ToString("o"), i);
}
)
);
// step5: 快速的提交大量的任务
Console.WriteLine("[{0}] 开始提交大量的任务", DateTime.Now.ToString("o"));
for (int i = 0; i < maxiQueueDepth * 100000; i++)
{
// 把数据Post到intPort内
intPort.Post(i);
}
Console.WriteLine("[{0}] 大量任务提交完毕。", DateTime.Now.ToString("o"));
Console.WriteLine("Press any key to exit");
Console.ReadKey();
dispatcher.Dispose();
}
}
}
二、原理讲解
其实在代码里面的注释，我想应该写的很明白了，不过毕竟是第一个例子，我还是稍微讲些CCR的大致原理（至于详细的实现原理，我后面会专门起文想说，这次主要的目的还是，先领略下CCR的神奇之处）：
首先，多线程是以操作系统线程池的形式被Dispatcher管理的，因此创建线程的工作实际上由Dispatcher代劳了；
其次，在CCR内，他规定Dispatcher只能处理任务，而这个任务只能从任务队列DispatcherQueue内获取，因此，我们要创建任务队列，并关联上Dispatcher；
然后，我们把自己要提交给处理函数处理的数据，封装在Port<T>内，Port其实是一个FIFO队列，专门用来接收用户提交的数据的；
最后，我们要把数据、处理函数、任务队列 组合起来，这就是上面代码中的step4，这步其实做了2个工作：
1、把port和处理函数，封装为Receive关联起来；
2、把Receive和DispatcherQueue关联起来；
这样，我们就完成了，所有的工作。
总之，CCR提供了一个模式，让我们只需要把需要并发、异步处理的工作，分解为：
1、输入数据--->post到Port内；
2、处理过程--->做成委托关联到任务队列中
这种方法，是的编写多线程程序的工作大大简化了，而且也能够让后台的代码能够被编译器统一优化。
二、CCR的任务调度
查阅MSDN：http://msdn.microsoft.com/en-us/library/bb648756.aspx
可知：（下面引用了Ncindy翻译的部分内容，感谢ncindy的辛苦劳动）
【一】：当一个元素被投递到附加了接收器的port，port的实现中将会发生如下操作：
step1. 为投递进来的元素创建一个容器。容器的类型（IPortElement）允许CCR在不知道元素类型的情况下将元素排队并将元素赋值给Task实例。
step2. 容器被放入队列。
step3. 如 果接收器列表不是null，并且其中有一个以上的接收器，port对象将会调用ReceiverTask.Evaluate方法来让接收器和它里面的仲裁 器层次检测元素是否可以被使用，在这个例子中，Evaluate方法将会返回true，并使用收到的元素和用户的delegate作为参数创建一个 Task<int>实例。
step4. port使用调用Evaluate方法返回的Task对象作为参数调用taskQueue.Enqueue，注意，当一个接收器是第一次被激活，它会被关联到由Arbiter.Activate方法提供的DispatcherQueue实例。
当上面的4步完成之后，生成的Task对象现在已经被调度逻辑分发(dealt)给了对应的DispatcherQueue。
【二】：一旦一个元素被放入DispatcherQueue，接下来将会做如下操作：
step1. DispatcherQueue向它所属的Dispatcher发信号，告诉Dispatcher一个新的任务可以被执行了。
step2. Dispatcher通知一个或者多个TaskExecutionWorker类型对象。每个TaskExecutionWorker对象管理一个操作系统线程。它将线程设置到一种高效的休眠状态，直到Dispatcher发出信号通知有元素可以被调度时。
step3. TaskExecutionWorker对象调用DispatcherQueue.Test方法从队列中获取一个任务。如果是可用的任务，TaskExecutionWorker对象则调用ITask.Execute。
step4. Task.Execute方法调用关联在task对象上的delegate，并将一个或者多个关联在task上的参数传递进去。
总之：在CCR中，线程池处理的任务，是由DispatcherQueue产生的；而DispathcerQueue有是根据用户线程通过Port或PortSet提交给的数据 和 初始化时指定的委托来产生任务的。因此可知影响任务调度的地方有3处：
1、客户端提交数据的地方：Port/PortSet的Post方法；
2、DispatcherQueue产生任务的地方：ReceiverTask的Evaluate方法；
3、Dispacher内线程池处理任务的地方：Task执行关联delegate的Execute方法；
而CCR就是通过给上面三处加入调度机制来达到任务调度的负载均衡目的的。
三、CCR的四种任务调度策略
using System;
namespace Microsoft.Ccr.Core
{
public enum TaskExecutionPolicy
{
Unconstrained = 0,
ConstrainQueueDepthDiscardTasks = 1,
ConstrainQueueDepthThrottleExecution = 2,
ConstrainSchedulingRateDiscardTasks = 3,
ConstrainSchedulingRateThrottleExecution = 4,
}
}
这4中策略分别应用在一下场景：
1、ConstrainQueueDepthDiscardTasks 按队列深度丢弃最旧任务
适用于：要处理的消息可以丢弃但是必须保存最近N条的情况。这对于CPU处理速度低于消息产生速度的情况很有好处，该策略能够保证丢弃的最旧任务的同时最新的N个任务能都得到调度。特别是在阻塞深度为1的时候，队列中保存的始终都是最新的任务。
2、ConstrainQueueDepthThrottleExecution 按照队列深度阻塞任务产生
适用于：消息不是规律产生，而是随机、爆炸性到达的情况。这对于来自网络获知其他机器的消息很相似，该策略保证任务不会被丢失，通过阻塞消息Post到Port/PortSet的方法来降低任务产生的速度。
3、ConstrainSchedulingRateDiscardTasks 按照固定速度处理消息且丢失未处理的最旧消息
适用于：处理产生速度有规律的消息，比如播放视频。在这种情况下一般所有的消息已经不是最重要的了，但保存最新的消息却很有意义，该策略能够保证代码会以固定的速度执行，即使消息以爆炸式的速度产生也没关系。
4、ConstrainSchedulingRateThrottleExecution 按照固定速度处理消息且阻塞任务缠上
适用于：消息产生源是同一处理器中的另一线程时。该策略会让消息的产生源慢下来，适应消息的处理速度，保证不会有任务丢失。
四、问题&解答