我有一个大的队列< string> dataQueue,我必须传输到多个客户端.
所以让我们从更简单的情况开始:
class SequentialBlockingCase { public static Queue<string> DataQueue = new Queue<string>(); private static List<string> _destinations = new List<string>(); /// <summary> /// Is the main function that is run in its own thread /// </summary> private static voID Run() { while (true) { if (DataQueue.Count > 0) { string data = DataQueue.Dequeue(); foreach (var destination in _destinations) { SendDataToDestination(destination,data); } } else { Thread.Sleep(1); } } } private static voID SendDataToDestination(string destination,string data) { //Todo: Send data using http post,instead simulate the send Thread.Sleep(200); }}}
现在这个设置工作正常.它位于那里并轮询队列,当有数据要发送时,它将它发送到所有目的地.
问题:
>如果其中一个目的地不可用或速度较慢,则会影响所有其他目的地.
>在并行执行的情况下,它不使用多线程.
>每个目的地的每次传输阻止.
所以这是我的第二次尝试:
class ParalleBlockingCase{ public static Queue<string> DataQueue = new Queue<string>(); private static List<string> _destinations = new List<string>(); /// <summary> /// Is the main function that is run in its own thread /// </summary> private static voID Run() { while (true) { if (DataQueue.Count > 0) { string data = DataQueue.Dequeue(); Parallel.ForEach(_destinations,destination => { SendDataToDestination(destination,data); }); } else { Thread.Sleep(1); } } } private static voID SendDataToDestination(string destination,string data) { //Todo: Send data using http post Thread.Sleep(200); }}
如果1个目的地缓慢或不可用,则此修订至少不会影响其他目的地.
但是这个方法仍然是阻塞的,我不确定Parallel.ForEach是否使用了线程池.我的理解是它将创建X个线程/任务并一次执行4个(4个核心@R_301_6947@).但它必须在任务5开始之前完全完成芬兰任务1.
因此我的第三个选择:
class ParalleAsyncCase{ public static Queue<string> DataQueue = new Queue<string>(); private static List<string> _destinations = new List<string> { }; /// <summary> /// Is the main function that is run in its own thread /// </summary> private static voID Run() { while (true) { if (DataQueue.Count > 0) { string data = DataQueue.Dequeue(); List<Task> tasks = new List<Task>(); foreach (var destination in _destinations) { var task = SendDataToDestination(destination,data); task.Start(); tasks.Add(task); } //Wait for all tasks to complete Task.WaitAll(tasks.ToArray()); } else { Thread.Sleep(1); } } } private static async Task SendDataToDestination(string destination,string data) { //Todo: Send data using http post await Task.Delay(200); }}
现在从我理解这个选项,仍然会在Task.WaitAll(tasks.ToArray())的主线程上阻塞;这很好,因为我不希望它以比执行任务更快的速度创建任务.
但是并行执行的任务应该使用ThreadPool,并且所有X个任务应该立即开始执行,而不是阻塞或按顺序执行. (线程池将在它们变为活动或正在等待时在它们之间交换)
现在我的问题.
选项3是否比选项2具有任何性能优势.
特别是在更高性能的服务器端方案中.在我正在处理的特定软件中.上面我的简单用例会有多个实例.即几个消费者.
我对两种解决方案的理论差异和专业与缺点感兴趣,如果有的话,甚至可能是更好的第四种选择.
解决方法 Parallel.ForEach将使用线程池.自 it doesn’t need any threads at all以来,异步代码不会(链接到我的博客).正如Mrinal所指出的,如果你有@R_301_6947@绑定代码,并行性是合适的;如果您有I / O绑定代码,则异步是合适的.在这种情况下,http POST显然是I / O,因此理想的消费代码将是异步的.
maybe even a better 4th option if there is one.
我建议让你的消费者完全异步.为此,您需要使用与异步兼容的生产者/消费者队列.有一个相当先进的(BufferBlock<T>
) in the TPL Dataflow library,一个相当简单的(AsyncProducerConsumerQueue<T>
) in my AsyncEx library.
使用其中任何一个,您可以创建完全异步的使用者:
List<Task> tasks = new List<Task>();foreach (var destination in _destinations){ var task = SendDataToDestination(destination,data); tasks.Add(task);}await Task.WhenAll(tasks);
或者,更简化:
var tasks = _destinations .Select(destination => SendDataToDestination(destination,data));await Task.WhenAll(tasks);总结
以上是内存溢出为你收集整理的c# – 在重I / O *** 作中的Parallel.ForEach与Async Forloop全部内容,希望文章能够帮你解决c# – 在重I / O *** 作中的Parallel.ForEach与Async Forloop所遇到的程序开发问题。
如果觉得内存溢出网站内容还不错,欢迎将内存溢出网站推荐给程序员好友。
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)