c# - 最大化任务并发性 - 使用 TCP/IP 和特定端口过滤

Question

这是 this 的后续问题问题。在下一个级别，我现在想使用最大任务并发连接到大量 IP 地址上的预期主机，在特定端口上使用 TCP/IP。

我自己的研究以及社区引用使我找到了关键文章，例如:

• How to check TCP/IP port availability using C# (Socket Communication)
• Checking if ip with port is available?
• How to set the timeout for a TcpClient?
• 一个非常令人印象深刻的大规模 ping 解决方案:Multithreading C# GUI ping example
• 当然还有这个问题的前身:C#, Maximize Thread Concurrency

这让我可以设置自己的代码，它工作正常，但目前需要整整 30 秒才能完成扫描 255 个 IP，仅使用一个特定端口。鉴于测试，机器有 8 个逻辑核心，这一观察结果表明我的构造实际上最多产生 8 个并发任务 (255/8=31.85)。

我编写的函数返回一个响应 IP 列表 {IPs}，它是要检查的所有 IP 列表 {IP_Ports} 的子集。这是我当前的代码，工作正常但尚不适合在更大的网络上使用，因为我怀疑缺乏有效的任务并发:

// 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 <TCP_check>
    public static List<string> TCP(Dictionary<string, int> IP_Ports, int TimeoutInMS = 100)
    {
        // Locals
        IPs = new List<string>();

        // Perform remote host check
        AllDone = false;
        TCP_check(IP_Ports, TimeoutInMS);
        while (!AllDone) { Thread.Sleep(50); }

        // Finish
        return IPs;
    }
    private static async void TCP_check(Dictionary<string, int> IP_Ports, int timeout)
    {// async worker method:  check remote host via TCP-IP


        // Build task-set for parallel IP queries
        var tasks = IP_Ports.Select(host => TCP_IPAndUpdateAsync(host.Key, host.Value, timeout));

        // Start execution queue
        await Task.WhenAll(tasks).ContinueWith(t =>
        {
            AllDone = true;
        });
    }
    private static async Task TCP_IPAndUpdateAsync(string ip, int port, int timeout)
    {// method to call IP-check

        // Run method asynchronously
        await Task.Run(() =>
        {
            // Locals
            TcpClient client;
            IAsyncResult result;
            bool success;

            try
            {
                client  = new TcpClient();
                result  = client.BeginConnect(ip, port, null, null);
                success = result.AsyncWaitHandle.WaitOne(TimeSpan.FromMilliseconds(timeout));

                if (success)
                {
                    lock (lockObj)
                    {
                        IPs.Add(ip);
                    }
                }
            }
            catch (Exception e)
            {
                // do nothing
            }
        });
    }


}// end     public static class CheckRemoteHost

所以我的问题是:我如何才能最大化在端口 X 使用 TCP/IP 请求响应的任务并发性，以便我可以在大型内部网络上获得非常快速的 IP 端口网络扫描？

Answer 1

详情

默认的任务调度器通常是 ThreadPool 调度器。这意味着并发任务的数量将受到池中可用线程的限制。

Remarks

The thread pool provides new worker threads or I/O completion threads on demand until it reaches the minimum for each category. By default, the minimum number of threads is set to the number of processors on a system. When the minimum is reached, the thread pool can create additional threads in that category or wait until some tasks complete. Beginning with the .NET Framework 4, the thread pool creates and destroys threads in order to optimize throughput, which is defined as the number of tasks that complete per unit of time. Too few threads might not make optimal use of available resources, whereas too many threads could increase resource contention.

(来源:https://msdn.microsoft.com/en-us/library/system.threading.threadpool.getminthreads(v=vs.110).aspx)

您可能刚好低于线程池在任务完成后启动新线程的阈值。因此，为什么您一次只能运行 8 个并发任务。

解决方案

1.使用带超时的 ConnectAsync。

而不是创建一个单独的任务来阻塞等待连接。您可以调用 ConnectAsync 并延迟加入它以创建超时。 ConnectAsync 似乎不会阻塞线程池线程。

public static async Task<bool> ConnectAsyncWithTimeout(this Socket socket, string host, int port, int timeout = 0)
{
    if (timeout < 0)
        throw new ArgumentOutOfRangeException("timeout");
    try
    {
        var connectTask = socket.ConnectAsync(host, port);
        var res = await Task.WhenAny(connectTask, Task.Delay(timeout));
        await res;
        return connectTask == res && connectTask.IsCompleted && !connectTask.IsFaulted;
    } 
    catch(SocketException se)
    {
        return false;
    }
}

示例用法

private static async Task TCP_IPAndUpdateAsync(string ip, int port, int timeout)
{// method to call IP-check
    client  = new TcpClient();
    var success = await client.Client.ConnectAsyncWithTimeout(ip, port, timeout);
    if (success)
    {
        lock (lockObj)
        {
            IPs.Add(ip);
        }
    }
}

2.使用长时间运行的任务。

使用 Task.Factor.StartNew您可以指定任务为 LongRunning .线程池任务调度程序专门为任务创建一个新线程，而不是使用线程池。这将绕过您达到的 8 个线程限制。但是，应该注意，如果您打算天真地创建数千个任务，这不是一个好的解决方案。从那时起，瓶颈将是线程上下文切换。但是，您可以将所有工作拆分为 100 个任务。

3.使用非阻塞连接

此方法不需要创建多个任务。相反，您可以在单个线程上调用多个连接并同时检查多个套接字的状态。不过，此方法涉及更多。如果您更愿意采用这种方法并想要一个更完整的示例，请发表评论让我知道。以下是有关如何使用 API 的简短片段。

var socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
socket.Blocking = false;
try
{
    socket.Connect("127.0.0.1", 12345);
} 
catch(SocketException se)
{
    //Ignore the "A non-blocking socket operation could not be completed immediately" error
    if (se.ErrorCode != 10035)
        throw;
}

//Check the connection status of the socket.
var writeCheck = new List<Socket>() { socket };
var errorCheck = new List<Socket>() { socket };
Socket.Select(null, writeCheck, errorCheck, 0);

if (writeCheck.Contains(socket)) 
{
    //Connection opened successfully.
}
else if (errorCheck.Contains(socket))
{
    //Connection refused
}
else
{
    //Connect still pending
}