更好地理解任务和调度

Question

我是任务、并行和调度等方面的新手。在这里我创建了两个阻塞集合。一个用于收集输入，另一个用于输出。我添加了 100 个项目。所以我认为它们的总数应该是100或0。或者它们的总和等于100。

但是我发现它们都是0。请帮助我用简单的语言理解这些概念。

    static void Main(string[] args)
    {
        new Program().run();
    }
    void run()
    {
        int threadCount = 4;
        Task[] workers = new Task[threadCount];

        Task.Factory.StartNew(consumer);
        // We can do other work in parallel
        for (int i = 0; i < threadCount; ++i)
        {
            int workerId = i;
            Task task = new Task(() => worker(workerId));
            workers[i] = task;
            task.Start();
        }

        for (int i = 0; i < 100; ++i)
        {
            Console.WriteLine("Queueing work item {0}", i);
            inputQueue.Add(i);
            Thread.Sleep(50);
        }

        Console.WriteLine("Stopping adding.");
        inputQueue.CompleteAdding();
        Console.WriteLine("The count in InputQueue= {0}", inputQueue.Count);// 0
        Task.WaitAll(workers);
        outputQueue.CompleteAdding();
        Console.WriteLine("The count in OutputQueue= {0}", outputQueue.Count); // 0
        Console.WriteLine("Done.");

        Console.ReadLine();
    }

    void worker(int workerId)
    {
        Console.WriteLine("Worker {0} is starting.", workerId);

        foreach (var workItem in inputQueue.GetConsumingEnumerable())
        {
            Console.WriteLine("Worker {0} is processing item {1}", workerId, workItem);
            Thread.Sleep(100);          // Simulate work.
            outputQueue.Add(workItem);  // Output completed item.
        }

        Console.WriteLine("Worker {0} is stopping.", workerId);
    }

    void consumer()
    {
        Console.WriteLine("Consumer is starting.");

        foreach (var workItem in outputQueue.GetConsumingEnumerable())
        {
            Console.WriteLine("Consumer is using item {0}", workItem);
            Thread.Sleep(25);
        }

        Console.WriteLine("Consumer is finished.");
    }

    BlockingCollection<int> inputQueue = new BlockingCollection<int>();
    BlockingCollection<int> outputQueue = new BlockingCollection<int>();
}

Answer 1

你有一个 3 阶段的管道

运行 -- (inputQueue) -- worker -- (outputQueue) -- 消费者

所有 3 个阶段同时运行，因此只要有 1 个项目在 inputQueue 中，就可以立即取出并移动到 outputQueue，只要将一个项目放入 outputQueue 即可立即取出并处理。

所以要获得 100 的计数，您需要这样做

(100 - "run" 放入 "inputQueue" 的项目数) + "inputQueue.Count" + 介于 0 和 "threadCount" 之间的某个数字表示已从 "inputQueue" 中取出但尚未放入 "outputQueue" 中的项目 + "outputQueue.Count" + “消费者”从“输出队列”中取出的项目数

因为您的线程数和等待数与负载完全平衡，所以上述公式很可能是 0 + 0 + 4 + 0 + 96，最后 4 个元素在 worker 中等待处理，其他所有元素都已由 consumer 处理.如果您执行了一半的工作线程并使消费者花费 4 倍的时间来处理您将得到像 0 + 48 + 2 + 25 + 25 这样的数字，其中 48 个等待工作人员处理，2 个正在由工作人员处理，25 个等待消费者处理，并且25 个已由消费者处理。