利用LINQ实现简单的异步并发

在.NET应用程序的开发过程中，处于性能考虑，我们可能需要多个互不相关的异步作业同时执行。此时我们可能会想到利用Parallel.ForEach函数，但倘若这些作业是有返回值的，而且我们需要将这些返回值收集起来，那么我们就会需要利用await/async操作，然后目前基本的Parallel.ForEach函数并不支持await/async操作。因此，我们需要另辟蹊径，比如Task.WhenAll。本文的示例代码的实现思路分为两步，一是将完整的作业集分割成若干个子集，之后利用LINQ的Select函数未子集中的每个作业创建一个task thread并运行。

1. 切割数组

参考：https://*.com/questions/419019/split-list-into-sublists-with-linq

public static class ListExtensions
{
    public static IEnumerable<IEnumerable<T>> Split<T>(this List<T> source, int chunkSize)
    {
        var pos = 0;
        while (source.Skip(pos).Any())
        {
            yield return source.Skip(pos).Take(chunkSize);
            pos += chunkSize;
        }
    }
}

2. 并发运行作业

// Simulate the job
public class MyTask
{
    private readonly string _name;
    private static readonly Random Rnd = new Random();
    private static readonly ILog Log = LogManager.GetLogger(typeof(MyTask));

    public MyTask(string name) => _name = name;

    public async Task<int> LongRunAsync()
    {
        var time = Rnd.Next(1, 10);
        Log.Info($"{_name} will sleep for {time} secs.");
        await Task.Delay(time * 1000);
        Log.Info($"{_name} is awake.");
        return time;
    }
}

// Main
class Program
{
    static async Task Main(string[] args)
    {
        var myTasks = new List<MyTask>();
        for (var i = 0; i < 321; ++ i)
        {
            myTasks.Add(new MyTask($"Task_{i}"));
        }

        var timeBag = new List<int>();
        var chunkSize = 50;
        foreach (var tasks in myTasks.Split(chunkSize))
        {
            // Key logic for concurrency
            var taskList = tasks.Select(async x => timeBag.Add(await x.LongRunAsync()));
            await Task.WhenAll(taskList);
            // End
        }
    }
}

值得注意的是，Select函数会为作业集中的每个作业创建一个thread，因此一定要根据实际情况来调整chunkSize的值，以防系统资源被耗尽。