получая данные начинает их асинхронную обработку
- в показанном коде нет ничего асинхронного. Не путайте многопоточность и асинхронность.
Задача на шаблон проектирования 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 завершился
Видно, что обработчики не остановятся, пока не обработают все данные, которые есть в коллекции.