c# – 在重I / O操作中的Parallel.ForEach与Async Forloop

我有一个大的队列< 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个核心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个任务应该立即开始执行,而不是阻塞或按顺序执行. (线程池将在它们变为活动或正在等待时在它们之间交换)

现在我的问题.

选项3是否比选项2具有任何性能优势.

特别是在更高性能的服务器端方案中.在我正在处理的特定软件中.上面我的简单用例会有多个实例.即几个消费者.

我对两种解决方案的理论差异和专业与缺点感兴趣,如果有的话,甚至可能是更好的第四种选择.