using System;using System.Collections.Concurrent;using System.Collections.Generic;using System.Linq;using System.Text;using System.Threading;using System.Threading.Tasks;namespace ConsoleApplication6{ public class PCQueue : IDisposable { class WorkItem { public readonly TaskCompletionSource TaskSource; public readonly Action Action; public readonly CancellationToken? CancelToken; public WorkItem( TaskCompletionSource taskSource, Action action, CancellationToken? cancelToken) { TaskSource = taskSource; Action = action; CancelToken = cancelToken; } } BlockingCollection
_taskQ = new BlockingCollection
(); ///
/// 为每个消费者创建并启动单独的任务： 这里我启动了2个任务，用于打印easy和easy2这两个的顺序不一定 /// ///
/// 默认任务取消标识为null的任务进队方法 /// ///
/// 任务进队方法含标识 /// ///
(); _taskQ.Add(new WorkItem(tcs, action, cancelToken)); //通过TaskCompletionSource返回任务本身，可查看任务的响应信息如result，exception,status等等 return tcs.Task; } void Consume() { foreach (WorkItem workItem in _taskQ.GetConsumingEnumerable()) if (workItem.CancelToken.HasValue && workItem.CancelToken.Value.IsCancellationRequested) { workItem.TaskSource.SetCanceled(); } else try { workItem.Action(); workItem.TaskSource.SetResult(1234); // 表示完成 } catch (OperationCanceledException ex) { if (ex.CancellationToken == workItem.CancelToken) workItem.TaskSource.SetCanceled(); else workItem.TaskSource.SetException(ex); } catch (Exception ex) { workItem.TaskSource.SetException(ex); } } } class Program { static void Main(string[] args) { CancellationToken token1 = new CancellationToken(true); // var pcQ = new PCQueue(2); Task task = pcQ.EnqueueTask(() => Console.WriteLine("Easy!"),token1);//输出easy的任务不会执行 task = pcQ.EnqueueTask(() => Console.WriteLine("Easy2!"));//会执行 // ... Console.Read(); } }}