按顺序安排任务

Question

我的任务如下：

1. 任务 taskInput 以获取数据到 inputQueue
2. 任务 taskOutput 将数据从 inputQueue 抓取到 outputQueue
3. 使用 outputQueue 中的数据的并行任务

我想运行符合条件的任务：

先运行 taskInput，再运行 taskOutput，最后运行 Consumer。

对应代码：

            // Get data to inputQueue
            Task taskInput = new Task(()=>AddingItemToInputQueue());
            taskInput.Start();  
            // Grab data from inputQueue to outputQueue.
            Task taskOutput = new Task(() => AddItemToOutputQueue());
            taskOutput.Start();

            // Parallel tasks for consume data from outputQueue
            int threadCount = n;
            Task[] workers = new Task[threadCount];
            for (int i = 0; i < threadCount; ++i)
            {
                Task task=Task.Run(()=>Consumer(i));
                workers[i] = task;
            }
            Task.WaitAll(workers);

关于inputQueue和outputQueue：

    BlockingCollection<Messages> InputQueue = new BlockingCollection<Messages>();
    BlockingCollection<Messages> OutputQueue = new BlockingCollection<Messages>();

我的问题：

1. 安排任务。我以为我们可以使用 Task.ContinueWith Method bot 不确定如何将其应用于 Consumer。
2. 我不确定它是否是线程安全的，因为在运行 Consumer 时可能会将新项目添加到 inputQueue。

Answer 1

async-await 可以这样写：

async Task DoStuffAsync()
{
    // Get data to inputQueue
    await Task.Run(()=>AddingItemToInputQueue());

    // Grab data from inputQueue to outputQueue.
    await Task.Run(() => AddItemToOutputQueue());

    // Parallel tasks for consume data from outputQueue
    int threadCount = n;
    Task[] workers = new Task[threadCount];
    for (int i = 0; i < threadCount; ++i)
    {
        Task task=Task.Run(()=>Consumer(i));
        workers[i] = task;
    }

    await Task.WhenAll(workers);
}

但是由于您需要将 AddingItemToInputQueue 和 AddItemToOutputQueue 与 Tasks 包装起来以使它们异步，因此您只是在增加开销。 Consumer 也是如此。

你不妨这样做：

void DoStuff()
{
    // Get data to inputQueue
    AddingItemToInputQueue();

    // Grab data from inputQueue to outputQueue.
    AddItemToOutputQueue();

    // Parallel tasks for consume data from outputQueue
    int threadCount = n;
    Parallel.For(0, n, i => Consumer(i));
}

Answer 2

您可以尝试启动第一个任务 (taskInput)，完成后，继续执行第二个任务 (taskOutput)，除非需要并行工作。在这种情况下，您必须分别启动两者，就像您已经在做的那样。

Task.Run(() => AddingItemToInputQueue())
            .ContinueWith(task => AddItemToOutputQueue());

同时，启动任务以消耗来自outputQueue的数据

// Parallel tasks for consume data from outputQueue
int threadCount = n;
Task[] workers = new Task[threadCount];
for (int i = 0; i < threadCount; ++i)
{
    Task task = Task.Run(() => Consumer(i));
    workers[i] = task;
}
Task.WaitAll(workers);

或者你可以试试这样的：

Task.Run(() => AddingItemToInputQueue())
    .ContinueWith(x => AddItemToOutputQueue())
    .ContinueWith(t =>
    {
        int threadCount = n;
        Task[] workers = new Task[threadCount];
        for (int i = 0; i < threadCount; ++i)
        {
            Task task = Task.Run(() => Consumer(i));
            workers[i] = task;
        }
        Task.WaitAll(workers);
    });

在这种情况下，它将按照您的意愿运行：首先是taskInput，然后是taskOutput，最后是Consumer。

BlockingCollection 是线程安全的，因此您可以在多任务中添加和删除数据，它会在必要时自行管理阻塞。

您可以查看更多关于Task.Factory.StartNew和Task.Runhere的信息