1

Смысл задачи такой, есть два канала обмена данными, по первому на сервер поступает массив байт с данными для обработки, в голове массива 4 байта идентифицируют поток который, собственно, эту обработку запросил. Сервер получая данные начинает их асинхронную обработку, и возвращает ответ по второму каналу, в голове всё те же 4 байта, чтобы понять какому потоку ответные данные принадлежат, данные могут быть возвращены не в том порядке в котором поступили. В зависимости от данных их обработка может занимать продолжительное время. Нужно приостановить поток на клиенте, чтобы дождаться получения данных. Как только данные будут получены, управляющим потоком клиента, нужно дать понять спящему(потоку которому принадлежат данные), что ему пора активизироваться и эти данные забрать. Я нашел решение, но выглядит как то кастыльно, хотел бы спросить как эту задачу можно решить по-человечески. Набросал для консоли упрощенный пример.

class Program
    {
        static Dictionary<int, Thread> dict = new Dictionary<int, Thread>();
        static void Main(string[] args)
        {

            Console.WriteLine("Начало");
            for (int i = 1; i <= 10; i++)
            {
                var thread = new Thread(new ParameterizedThreadStart(DoWork));
                thread.Start(i);
                dict.Add(i, thread);
                
            }
            while (true)
            {
                var rnd = new Random();
                int id = (rnd.Next() % 10) + 1; //Получили данные для какого либо потока
                Thread.Sleep(1000);
                Console.WriteLine("Будим поток: " + id);
                dict[id].Interrupt();
            }
        }

        static void DoWork(object id)
        {
            id = (int) id;
            while (true)
            {
                try
                {
                    Thread.Sleep(Timeout.Infinite);
                }
                catch (Exception e)
                {
                    //Чтобы работало нужна обработка ошибки.
                }
                Console.WriteLine("Поток № " + id + " Проснулся");
            }
        }

    }
2
  • 1
    Auto/ManualResetEvent 5 мая 2021 в 11:52
  • 1
    а ещё Monitor.Wait/Pulse 5 мая 2021 в 11:53

1 ответ 1

4

получая данные начинает их асинхронную обработку - в показанном коде нет ничего асинхронного. Не путайте многопоточность и асинхронность.

Задача на шаблон проектирования Producer/Consumer. От изобретения велосипеда спасет BlockingCollection. Она блокирует поток, пока не будут получены данные. То есть при получении данных поток автоматически просыпается, и как только данные закончились - сам засыпает.

Кстати, совсем не понятно, зачем вы потоки храните в словаре, вы же все равно с ними ничего сделать не сможете.

Например пусть данными для обработки будет string.

private BlockingCollection<string> jobs = new BlockingCollection<string>();
static void RunLoop(BlockingCollection<string> bc)
{
    foreach (string item in bc.GetConsumingEnumerable())
    {
        DoSomething(item); // метод обработки данных
    }
}

Запуск

for (int i = 0; i < 10; i++)
{
    new Thread(() => RunLoop(jobs)).Start();
}

Остановка всех воркеров сразу

jobs.CompleteAdding();

Отправка данных в обработку

jobs.Add("Hello World!");

Работающие потоки будут разгребать очередь задач "кто быстрее", то есть данные будет получать случайный поток из ожидающих в данный момент.

Эта схема 1 продюсер, 10 консьюмеров. На самом деле jobs.Add можно вызывать из любого потока в любое время. Если вам нужно потоки данных разделить явно, заведите несколько таких коллекций, запускайте по одному консьюмеру на коллекцию, здесь уж на что фантазии хватит.

Документация.

Для асинхронной реализации шаблона Producer/Consumer начиная с .NET Core 3.1 доступен еще один набор инструментов - System.Threading.Channels.

Тестовый пример

static void Main(string[] args)
{
    BlockingCollection<string> jobs = new BlockingCollection<string>();
    const int workersCount = 5;
    Console.WriteLine($"[Main] Запускаю {workersCount} обработчиков");
    for (int i = 0; i < workersCount; i++)
    {
        int id = i + 1;
        new Thread(() => RunLoop(jobs, id)).Start();
    }
    const int jobsCount = 20;
    Console.WriteLine($"[Main] Добавляю {jobsCount} задач");
    for (int i = 0; i < jobsCount; i++)
        jobs.Add($"Задача №{i + 1}");
    Console.WriteLine("[Main] Задачи добавлены");
    Console.WriteLine("[Main] Сплю пару секунд");
    Thread.Sleep(2000);
    Console.WriteLine($"[Main] Добавляю еще {jobsCount} задач");
    for (int i = jobsCount; i < jobsCount + jobsCount; i++)
        jobs.Add($"Задача №{i + 1}");
    Console.WriteLine("[Main] Задачи добавлены");
    Console.WriteLine("[Main] Останавливаю обработчики");
    jobs.CompleteAdding();
    Console.WriteLine("[Main] Готово.");
}

static void RunLoop(BlockingCollection<string> bc, int id)
{
    Console.WriteLine($"Консьюмер №{id} запустился");
    foreach (string item in bc.GetConsumingEnumerable())
    {
        Console.WriteLine($"Консьюмер №{id} начал делать работу: {item}");
        Thread.Sleep(500);
        Console.WriteLine($"Консьюмер №{id} закончил делать работу: {item}");
    }
    Console.WriteLine($"Консьюмер №{id} завершился");
}

Вывод в консоль

[Main] Запускаю 5 обработчиков
Консьюмер №1 запустился
Консьюмер №2 запустился
Консьюмер №3 запустился
[Main] Добавляю 20 задач
Консьюмер №5 запустился
Консьюмер №4 запустился
[Main] Задачи добавлены
[Main] Сплю пару секунд
Консьюмер №1 начал делать работу: Задача №2
Консьюмер №4 начал делать работу: Задача №1
Консьюмер №3 начал делать работу: Задача №5
Консьюмер №5 начал делать работу: Задача №3
Консьюмер №2 начал делать работу: Задача №4
Консьюмер №5 закончил делать работу: Задача №3
Консьюмер №4 закончил делать работу: Задача №1
Консьюмер №2 закончил делать работу: Задача №4
Консьюмер №3 закончил делать работу: Задача №5
Консьюмер №1 закончил делать работу: Задача №2
Консьюмер №5 начал делать работу: Задача №6
Консьюмер №1 начал делать работу: Задача №10
Консьюмер №4 начал делать работу: Задача №7
Консьюмер №2 начал делать работу: Задача №8
Консьюмер №3 начал делать работу: Задача №9
Консьюмер №4 закончил делать работу: Задача №7
Консьюмер №2 закончил делать работу: Задача №8
Консьюмер №2 начал делать работу: Задача №12
Консьюмер №1 закончил делать работу: Задача №10
Консьюмер №1 начал делать работу: Задача №13
Консьюмер №4 начал делать работу: Задача №11
Консьюмер №5 закончил делать работу: Задача №6
Консьюмер №3 закончил делать работу: Задача №9
Консьюмер №5 начал делать работу: Задача №14
Консьюмер №3 начал делать работу: Задача №15
Консьюмер №4 закончил делать работу: Задача №11
Консьюмер №2 закончил делать работу: Задача №12
Консьюмер №2 начал делать работу: Задача №17
Консьюмер №1 закончил делать работу: Задача №13
Консьюмер №1 начал делать работу: Задача №18
Консьюмер №4 начал делать работу: Задача №16
Консьюмер №5 закончил делать работу: Задача №14
Консьюмер №5 начал делать работу: Задача №19
Консьюмер №3 закончил делать работу: Задача №15
Консьюмер №3 начал делать работу: Задача №20
[Main] Добавляю еще 20 задач
[Main] Задачи добавлены
[Main] Останавливаю обработчики
[Main] Готово.
Консьюмер №1 закончил делать работу: Задача №18
Консьюмер №4 закончил делать работу: Задача №16
Консьюмер №4 начал делать работу: Задача №22
Консьюмер №5 закончил делать работу: Задача №19
Консьюмер №5 начал делать работу: Задача №23
Консьюмер №1 начал делать работу: Задача №21
Консьюмер №3 закончил делать работу: Задача №20
Консьюмер №3 начал делать работу: Задача №24
Консьюмер №2 закончил делать работу: Задача №17
Консьюмер №2 начал делать работу: Задача №25
Консьюмер №3 закончил делать работу: Задача №24
Консьюмер №2 закончил делать работу: Задача №25
Консьюмер №2 начал делать работу: Задача №27
Консьюмер №1 закончил делать работу: Задача №21
Консьюмер №1 начал делать работу: Задача №28
Консьюмер №3 начал делать работу: Задача №26
Консьюмер №5 закончил делать работу: Задача №23
Консьюмер №5 начал делать работу: Задача №29
Консьюмер №4 закончил делать работу: Задача №22
Консьюмер №4 начал делать работу: Задача №30
Консьюмер №4 закончил делать работу: Задача №30
Консьюмер №4 начал делать работу: Задача №31
Консьюмер №1 закончил делать работу: Задача №28
Консьюмер №1 начал делать работу: Задача №32
Консьюмер №2 закончил делать работу: Задача №27
Консьюмер №2 начал делать работу: Задача №33
Консьюмер №3 закончил делать работу: Задача №26
Консьюмер №3 начал делать работу: Задача №34
Консьюмер №5 закончил делать работу: Задача №29
Консьюмер №5 начал делать работу: Задача №35
Консьюмер №2 закончил делать работу: Задача №33
Консьюмер №2 начал делать работу: Задача №36
Консьюмер №1 закончил делать работу: Задача №32
Консьюмер №1 начал делать работу: Задача №37
Консьюмер №4 закончил делать работу: Задача №31
Консьюмер №4 начал делать работу: Задача №38
Консьюмер №5 закончил делать работу: Задача №35
Консьюмер №5 начал делать работу: Задача №39
Консьюмер №3 закончил делать работу: Задача №34
Консьюмер №3 начал делать работу: Задача №40
Консьюмер №2 закончил делать работу: Задача №36
Консьюмер №1 закончил делать работу: Задача №37
Консьюмер №4 закончил делать работу: Задача №38
Консьюмер №4 завершился
Консьюмер №1 завершился
Консьюмер №2 завершился
Консьюмер №5 закончил делать работу: Задача №39
Консьюмер №5 завершился
Консьюмер №3 закончил делать работу: Задача №40
Консьюмер №3 завершился

Видно, что обработчики не остановятся, пока не обработают все данные, которые есть в коллекции.

Ваш ответ

Нажимая на кнопку «Отправить ответ», вы соглашаетесь с нашими условиями использования и подтверждаете, что прочитали и поняли наши политику конфиденциальности и нормы поведения.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.