C#，最大化线程并发

Question

在 Google 和社区的帮助下，我能够构建一组很好的方法，让我能够异步调用函数。此函数正在测试远程主机属性，因此大部分时间都处于空闲状态。出于这个原因，我想最大化启动的并发线程数，以便可以在最短的时间内处理所有调用。

这是我目前的代码：

// Check remote host connectivity
public static class CheckRemoteHost
{

    // Private Class members
    private static bool AllDone     = false;
    private static object lockObj   = new object();
    private static List<string> IPs;


    // Wrapper: manage async method <Ping>
    public static List<string> Ping(HashSet<string> IP_Ports, int TimeoutInMS = 100)
    {// async worker method:  check remote host via <Ping>

        // Locals
        IPs = new List<string>();

        // Perform remote host check
        AllDone = false;
        Ping_check(IP_Ports, TimeoutInMS);
        while (!AllDone) { CommonLib.Utils.ApplicationWait(10, 10); }

        // Finish
        return IPs;
    }
    private static async void Ping_check(HashSet<string> IP_Ports, int timeout)
    {
        // Locals
        var tasks = new List<Task>();

        // Build task-set for parallel Ping checks
        foreach (string host in IP_Ports)
        {
            var task = PingAndUpdateAsync(host, timeout);
            tasks.Add(task);
        }

        // Start execution queue
        await Task.WhenAll(tasks).ContinueWith(t =>
        {
            AllDone = true;
        });

    }
    private static async Task PingAndUpdateAsync(string ip, int timeout)
    {

        // Locals
        System.Net.NetworkInformation.Ping ping;
        System.Net.NetworkInformation.PingReply reply;

        try
        {
            ping    = new System.Net.NetworkInformation.Ping();
            reply   = await ping.SendPingAsync(ip, timeout);

            if(reply.Status == System.Net.NetworkInformation.IPStatus.Success)
            {
                lock (lockObj)
                {
                    IPs.Add(ip);
                }
            }
        }
        catch
        {
            // do nothing
        }
    }

}// end     public static class CheckRemoteHost

此代码经过了相当广泛的测试，代码看起来稳定且可靠地报告了实时主机。话虽如此，我知道它一次只产生 8 个线程（= 我的测试机器上的逻辑核心数）。

代码的关键部分是这样的：

// Start execution queue
await Task.WhenAll(tasks).ContinueWith(t =>
{
    AllDone = true;
});

这就是我想将并发启动的线程数增加/最大化到每个内核 25 个的地方（请记住线程作业是 99% 空闲）。

到目前为止，我的线程并发研究已经提出了显式线程和 Parallel.For 方法。但是，这些似乎具有相同的缺点，即产生不超过 8 个线程。

任何帮助将不胜感激，因此非常感谢大家的关注！

Answer 1

您的代码让您的生活变得艰难。它有很多不需要的管道，如果您在第一次运行时第二次调用 Ping，您正在共享静态字段，这将导致您的代码失败。

你需要摆脱所有这些东西。

我建议使用 Microsoft 的 Reactive Framework - 只需 NuGet "System.Reactive" 并将 using System.Reactive.Linq; 添加到您的代码中。然后你可以这样做：

public static class CheckRemoteHost
{
    public static IList<string> Ping(HashSet<string> IP_Ports, int TimeoutInMS = 100)
    {
        var query =
            from host in IP_Ports.ToObservable()
            from status in Observable.FromAsync(() => PingAsync(host, TimeoutInMS))
            where status
            select host;

        return query.ToList().Wait();
    }

    private static async Task<bool> PingAsync(string ip, int timeout)
    {
        try
        {
            var ping = new System.Net.NetworkInformation.Ping();
            var reply = await ping.SendPingAsync(ip, timeout);

            return reply.Status == System.Net.NetworkInformation.IPStatus.Success;
        }
        catch
        {
            return false;
        }
    }
}

就是这样。这就是您需要的所有代码。它会自动最大化使用线程来完成工作。