Как заставить работать ограниченное количество потоков в C#?

Question

Возникла некоторая задача(скорее интерес распирает меня самого, ибо я только начал разбираться в потоках) написать следующую программу: для некоторого порта проверить имеются ли подключения в адресах в отрезке от a до b, при этом ограничить количество потоков(что у меня не получилось) где a начальный IP b - конечный IP Написал следующий код, хотелось бы услышать заметки, советы, а так же узнать что почитать(желательно на понятном языке) как это реализовывается. Заранее спасибо

async void rangeChecker(byte[] startIP, byte[] endIP)
        {

            await (Task.Run(() =>
            {
                int count = 0;
                while (startIP[0] != endIP[0] || startIP[1] != endIP[1] || startIP[2] != endIP[2] || startIP[3] != endIP[3])
                {
                    if (startIP[3] != 255) startIP[3]++;
                    else
                    {
                        if (startIP[2] != 255) startIP[2]++;
                        else
                        {
                            if (startIP[1] != 255) startIP[1]++;
                            else
                            {
                                if (startIP[0] != 255) startIP[2]++; else break;
                                startIP[1] = 0;
                            }
                            startIP[2] = 0;
                        }
                        startIP[3] = 0;
                    }
                    string current = (startIP[0] + "." + startIP[1] + "." + startIP[2] + "." + startIP[3]);
                    Console.WriteLine(startIP[0] + "." + startIP[1] + "." + startIP[2] + "." + startIP[3]);
                    Thread myThread = new Thread(
                        ()=>
                        {
                            CheckPortConection(current, this.port);
                        }
                        );
                   
                    myThread.Start();
                    count++;
                }
                Console.WriteLine("Checked IPs : " + count);

            }));

        }
        public static void CheckPortConection(string currentIP, int port)
        {
            using (TcpClient tcpClient = new TcpClient())
            {
                try
                {
                    tcpClient.Connect(currentIP, port);
                    Console.WriteLine(currentIP+":"+port + " Port open");
                    
                }
                catch (Exception)
                {
                    Console.WriteLine(currentIP + ":" + port + "Port closed");
                }
            }
        }

Answer 1

Такое делается через очередь сообщений MSMQ, RabbitMQ. Устанавливается количество потоков обрабатывающих очередь сообщений. Как только поток освободится берёт следующее сообщение на обработку. Ну и соответственно те кто скидывают сообщения в очередь

Answer 2

Сейчас для этого модно использовать SemaphoreSlim:

private static SemaphoreSlim semaphore;

async void rangeChecker(byte[] startIP, byte[] endIP)
{
    ...
    semaphore = new SemaphoreSlim(5); // или сколько максимум нужно потоков
    while(...)
    {
        ...
        semaphore.Wait(); // либо await semaphore.WaitAsync();
        Thread myThread = new Thread(
        ...
    }
}

public static void CheckPortConection(string currentIP, int port)
{
    try
    {
        ...
    }
    finally
    {
        semaphore.Release();
    }
}

Ну и вообще хорошо бы уже без потоков писать, а на Task-ах с async/await.

Answer 3

Переделаем получение IP адресов

IEnumerable<string> getIPRange(byte[] startIP, byte[] endIP)
{
    while (startIP[0] != endIP[0] || startIP[1] != endIP[1] || startIP[2] != endIP[2] || startIP[3] != endIP[3])
    {
        if (startIP[3] != 255) startIP[3]++;
        else
        {
            if (startIP[2] != 255) startIP[2]++;
            else
            {
                if (startIP[1] != 255) startIP[1]++;
                else
                {
                    if (startIP[0] != 255) startIP[2]++; else break;
                    startIP[1] = 0;
                }
                startIP[2] = 0;
            }
            startIP[3] = 0;
        }
        string current = (startIP[0] + "." + startIP[1] + "." + startIP[2] + "." + startIP[3]);
        yield return current;
    }
}

Вторая функция остается той же

public static void CheckPortConection(string currentIP, int port)
{
    using (TcpClient tcpClient = new TcpClient())
    {
        try
        {
            tcpClient.Connect(currentIP, port);
            Console.WriteLine(currentIP + ":" + port + " Port open");

        }
        catch (Exception)
        {
            Console.WriteLine(currentIP + ":" + port + "Port closed");
        }
    }
}

Ну и как вызвать с ограничением количества потоков

int port = 1;
Parallel.ForEach(getIPRange(null, null), // ваши параметры
        new ParallelOptions() {MaxDegreeOfParallelism = 10},
        ip=>CheckPortConection(ip, port));

Answer 4

Переделаем получение IP адресов

IEnumerable<string> getIPRange(byte[] startIP, byte[] endIP)
{
    while (startIP[0] != endIP[0] || startIP[1] != endIP[1] || startIP[2] != endIP[2] || startIP[3] != endIP[3])
    {
        if (startIP[3] != 255) startIP[3]++;
        else
        {
            if (startIP[2] != 255) startIP[2]++;
            else
            {
                if (startIP[1] != 255) startIP[1]++;
                else
                {
                    if (startIP[0] != 255) startIP[2]++; else break;
                    startIP[1] = 0;
                }
                startIP[2] = 0;
            }
            startIP[3] = 0;
        }
        string current = (startIP[0] + "." + startIP[1] + "." + startIP[2] + "." + startIP[3]);
        yield return current;
    }
}

Переделаем проверку тоже

public static async Task CheckPortConection(string currentIP, int port)
{
    using (TcpClient tcpClient = new TcpClient())
    {
        try
        {
            await tcpClient.ConnectAsync(currentIP, port);
            Console.WriteLine(currentIP + ":" + port + " Port open");

        }
        catch (Exception)
        {
            Console.WriteLine(currentIP + ":" + port + "Port closed");
        }
    }
}

Как это вызвать

int port = 1;
var tasks = getIPRange(null, null)
    .Select(x=>CheckPortConection(x, port))
    .ToList();
    
await Task.WhenAll(tasks);

И никаких ограничений не надо, этот код не будет блокировать потоки.

БЛОГ НА HUSL

Как заставить работать ограниченное количество потоков в C#?

Ответы (4 шт):