我想比较两个理论场景。为了问题的目的,我简化了案例。但基本上它是您典型的生产者消费者场景。 (我关注的是消费者)。
我有一个很大的Queue<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);
}
}
}
现在这个设置工作得很好。它坐在那里并轮询 Queue
当有数据要发送时,它会将其发送到所有目的地。
问题:
- 如果其中一个目的地不可用或速度慢,则会影响所有其他目的地。
- 在并行执行的情况下不使用多线程。
- 每次传输到每个目的地的 block 。
这是我的第二次尝试:
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 个核心 cpu)。但它必须在任务 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 个任务应该立即开始执行,而不是阻塞或按顺序执行。 (线程池将在它们变为事件状态或 awaiting
时在它们之间交换)
现在我的问题。
选项 3 是否比选项 2 有任何性能优势。
特别是在更高性能的服务器端场景中。在我现在正在开发的特定软件中。上面会有我的简单用例的多个实例。即多个消费者。
我对这两种解决方案的理论差异和优缺点很感兴趣,如果有的话,可能还有更好的第四种选择。
最佳答案
Parallel.ForEach
将使用线程池。异步代码不会,因为it doesn't need any threads at all (链接是我的博客)。
正如 Mrinal 指出的那样,如果您的代码受 CPU 限制,则并行是合适的;如果您有 I/O 绑定(bind)代码,异步是合适的。在这种情况下,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 与异步 For 循环,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/39075845/