按顺序安排任务

Question

我的任务如下：

1. 任务taskInput将数据发送到inputQueue
2. 任务taskOutput从inputQueue到outputQueue获取数据
3. 使用来自outputQueue的数据的并行任务

我想以以下条件运行这些任务：

先运行taskInput，然后运行taskOutput，最后运行消费者。

相应的代码：

            // 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方法bot不确定如何将其应用于Consumer。
2. 我不确定它是否线程安全，因为在运行inputQueue时，可能会将新项添加到Consumer中。

Answer 1

您可以尝试启动第一个任务(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.Run 这里的更多信息。

Answer 2

使用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封装Task以使它们是异步的，所以您只是增加了开销。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));
}