基于网络的任务的多线程/并发策略

Question

我不擅长充分利用资源，因此我正在为需要并行且高效完成的任务寻找最佳方式。

我们有一个场景，我们必须 ping 数百万个系统并接收响应。响应本身不需要计算时间，但任务是基于网络的。

我当前的实现是这样的 -

Parallel.ForEach(list, ip =>
{
    try
    {
        // var record = client.QueryAsync(ip);
        var record = client.Query(ip);
        results.Add(record);
    }
    catch (Exception)
    {
        failed.Add(ip);
    }
});

我测试了这段代码

• 100 件商品大约需要 4 秒
• 1k 个项目大约需要 10 秒
• 10k 个项目大约需要 80 秒
• 100k 个项目大约需要 710 秒

我需要处理接近 20M 的查询，我应该使用什么策略来进一步加快速度

Answer 1

问题来了

Parallel.ForEach 使用线程池。此外，IO 绑定 操作会阻塞那些等待设备响应的线程并占用资源。

• 如果您有CPU 绑定 代码，并行度 是合适的；
• 虽然如果您有IO 绑定 代码，Asynchrony 是合适的。

在这种情况下，client.Query 显然是 I/O，因此理想的消费代码应该是异步的。

既然你说有一个异步版本，你最好使用async/await 模式和/或对并发任务的某种类型的限制，另一个巧妙的解决方案是在TPL dataflow 中使用ActionBlock Class图书馆。

---

数据流示例

public static async Task DoWorkLoads(List<IPAddress> addresses)
{
   var options = new ExecutionDataflowBlockOptions
                     {
                        MaxDegreeOfParallelism = 50
                     };

   var block = new ActionBlock<IPAddress>(MyMethodAsync, options);

   foreach (var ip in addresses)
      block.Post(ip);

   block.Complete();
   await block.Completion;

}

...

public async Task MyMethodAsync(IpAddress ip)
{

    try
    {
        var record = await client.Query(ip);
        // note this is not thread safe best to lock it
        results.Add(record);
    }
    catch (Exception)
    {
        // note this is not thread safe best to lock it
        failed.Add(ip);
    }
}

这种方式给你异步，也给你MaxDegreeOfParallelism，不浪费资源，让IO成为IO无需消耗不必要的资源

*免责声明，DataFlow 可能不是您想要的，但我只是认为 id 可以为您提供更多信息

---

Demo here

更新

我刚刚用 Parallel.Foreaceh 和 DataFlow 做了一些基准测试

运行多次 10000 ping

Parallel.Foreach = 30 秒

数据流 = 10 秒