任务工厂，当一个任务结束时启动新的任务。

Question

我发现了很多使用 TaskFactory 的方法，但是我找不到任何关于启动更多任务并观察何时结束一个任务并开始另一个任务的信息。

我总是希望有10个任务在工作。

我想要像这样的东西

int nTotalTasks=10;
int nCurrentTask=0;

Task<bool>[] tasks=new Task<bool>[nThreadsNum];

for (int i=0; i<1000; i++)
{
  string param1="test";
  string param2="test";

  if (nCurrentTask<10) // if there are less than 10 tasks then start another one
    tasks[nCurrentThread++] = Task.Factory.StartNew<bool>(() =>
    {
       MyClass cls = new MyClass();
       bool bRet = cls.Method1(param1, param2, i); // takes up to 2 minutes to finish
       return bRet;
    });

  // How can I stop the for loop until a new task is finished and start a new one?
}

- Mario

---

你有看过 WaitAny() 方法或者尝试等待一个 task.Result 吗？ - Broots Waymb

这些任务是I/O限制还是CPU限制？ - alexm

Answer 1

请查看Task.WaitAny方法：

等待提供的任一任务对象完成执行。

以下是文档中的示例：

var t1 = Task.Factory.StartNew(() => DoOperation1());
var t2 = Task.Factory.StartNew(() => DoOperation2());

Task.WaitAny(t1, t2)

- Cory Charlton

---

而我如何知道哪个任务完成了呢？因为我需要知道已完成任务的返回值。 - Mario

@MarioM - 没有 WaitAny 调用时，您可以像平常一样使用 Task.Result。 - Cory Charlton

@MarioM WaitAny() 返回任务输入数组中的索引。 - VMAtm

Answer 2

我会使用微软的响应式框架（NuGet“Rx-Main”）和TPL的组合来完成这个任务。这将变得非常简单。

以下是代码：

int nTotalTasks=10;
string param1="test";
string param2="test";

IDisposable subscription =
    Observable
        .Range(0, 1000)
        .Select(i => Observable.FromAsync(() => Task.Factory.StartNew<bool>(() =>
        {
            MyClass cls = new MyClass();
            bool bRet = cls.Method1(param1, param2, i); // takes up to 2 minutes to finish
            return bRet;
        })))
        .Merge(nTotalTasks)
        .ToArray()
        .Subscribe((bool[] results) =>
        {
            /* Do something with the results. */
        });

重点在于.Merge(nTotalTasks)，它限制了并发任务的数量。

如果需要在处理过程中停止，请调用subscription.Dispose()，一切都会被清理干净。

如果想要在产生结果时逐个处理每个结果，可以将代码从.Merge(...)更改为以下内容：

        .Merge(nTotalTasks)
        .Subscribe((bool result) =>
        {
            /* Do something with each result. */
        });

- Enigmativity

Answer 3

你见过BlockingCollection类吗？它允许您在多个线程并行运行，并且您可以等待一个任务的结果以执行另一个任务。更多信息请参见这里。

- Hernan Guzman

---

不，非常有趣。 - Mario

Answer 4

这就是你所需要的全部内容，虽不完整，但你只需等待第一个完成，然后运行第二个即可。

Task.WaitAny(task to wait on);

Task.Factory.StartNew()

- thewisegod

Answer 5

答案取决于要安排的任务是CPU密集型还是I/O密集型。

对于CPU密集型工作，我会使用Parallel.For() API，并通过ParallelOptions的MaxDegreeOfParallelism属性设置线程/任务数。

对于I/O密集型工作，同时执行的任务数量可以显著大于可用的CPU数量，因此策略是尽可能依赖异步方法，从而减少等待完成的线程总数。

如何停止for循环直到新任务完成并启动一个新任务？

可以使用await来限制循环。

    static void Main(string[] args)
    {
        var task = DoWorkAsync();
        task.Wait();

        // handle  results
        // task.Result;

        Console.WriteLine("Done.");
    }

    async static Task<bool> DoWorkAsync()
    {
        const int NUMBER_OF_SLOTS = 10;

        string param1="test";
        string param2="test";

        var results = new bool[NUMBER_OF_SLOTS]; 

        AsyncWorkScheduler ws = new AsyncWorkScheduler(NUMBER_OF_SLOTS);
        for (int i = 0; i < 1000; ++i)
        {
            await ws.ScheduleAsync((slotNumber) => DoWorkAsync(i, slotNumber, param1, param2, results));
        }

        ws.Complete();
        await ws.Completion;
    }

    async static Task DoWorkAsync(int index, int slotNumber, string param1, string param2, bool[] results)
    {
      results[slotNumber] = results[slotNumber} && await Task.Factory.StartNew<bool>(() =>
      {
          MyClass cls = new MyClass();
          bool bRet = cls.Method1(param1, param2, i); // takes up to 2 minutes to finish
          return bRet;
      }));

    }

一个帮助类AsyncWorkScheduler使用TPL.DataFlow组件以及Task.WhenAll():

class AsyncWorkScheduler
{
    public AsyncWorkScheduler(int numberOfSlots)
    {
        m_slots = new Task[numberOfSlots];
        m_availableSlots = new BufferBlock<int>();
        m_errors = new List<Exception>();
        m_tcs = new TaskCompletionSource<bool>();
        m_completionPending = 0;

        // Initial state: all slots are available
        for(int i = 0; i < m_slots.Length; ++i)
        {
            m_slots[i] = Task.FromResult(false);
            m_availableSlots.Post(i);
        }
    }

    public async Task ScheduleAsync(Func<int, Task> action)
    {
        if (Volatile.Read(ref m_completionPending) != 0)
        {
            throw new InvalidOperationException("Unable to schedule new items.");
        }

        // Acquire a slot 
        int slotNumber = await m_availableSlots.ReceiveAsync().ConfigureAwait(false);

        // Schedule a new task for a given slot
        var task = action(slotNumber);

        // Store a continuation on the task to handle completion events
        m_slots[slotNumber] = task.ContinueWith(t => HandleCompletedTask(t, slotNumber), TaskContinuationOptions.ExecuteSynchronously);
    }


    public async void Complete()
    {
        if (Interlocked.CompareExchange(ref m_completionPending, 1, 0) != 0)
        {
            return;
        }

        // Signal the queue's completion
        m_availableSlots.Complete();

        await Task.WhenAll(m_slots).ConfigureAwait(false);

        // Set completion
        if (m_errors.Count != 0)
        {
            m_tcs.TrySetException(m_errors);
        }
        else
        {
            m_tcs.TrySetResult(true);
        }

    }

    public Task Completion
    {
        get
        {
            return m_tcs.Task;
        }
    }



    void SetFailed(Exception error)
    {
        lock(m_errors)
        {
            m_errors.Add(error);
        }

    }

    void HandleCompletedTask(Task task, int slotNumber)
    {
       if (task.IsFaulted || task.IsCanceled)
       {
           SetFailed(task.Exception);
           return;
       }

       if (Volatile.Read(ref m_completionPending) == 1)
       {
           return;
       }


        // Release a slot
        m_availableSlots.Post(slotNumber);
    }

    int m_completionPending;
    List<Exception> m_errors;
    BufferBlock<int> m_availableSlots;
    TaskCompletionSource<bool> m_tcs;
    Task[] m_slots;

}

- alexm