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

Question

Фон

У меня есть очень большой файл двоичных данных (20+ ГБ), который мне нужно проанализировать, обработать данные и записать вывод. У меня очень мало опыта работы с такими большими объемами данных, и, хотя у меня возникли небольшие проблемы с концепцией, как с этим справляться, у меня есть идея. Примечание. Входные данные содержат ряд записей, извлеченных из мэйнфрейма IBM, поэтому они отформатированы следующим образом:

Первые 4 байта каждой записи (строка / строка) - это RDW (слово дескриптора записи). RDW содержит длину записи (включая RDW). Из-за RDW, хотя файл представляет собой один постоянный поток байтов, я знаю, где заканчивается каждая запись. Я мог бы перевести этот двоичный файл в текстовый файл, преобразовав каждые два байта в его шестнадцатеричное представление, а также добавить символ новой строки в конце записи, но я боюсь, насколько большим будет двоичный файл размером более 20 ГБ, если переводится так.

Поскольку я хочу оставить файл в виде двоичного файла, у меня есть одна идея, как поступить:

1. Чтение файла последовательно с использованием одного «основного» потока.
2. Как только мастер достиг конца записи (используя информацию, найденную в RDW), он порождает новый «рабочий» поток, передавая потоку данные, которые он прочитал из файла.
   • Рабочий поток анализирует данные, обрабатывает данные и выводит их куда-то. (Я думаю, что буду размещать данные в базе данных SQLite.)
3. Пока рабочий поток работает, главный поток продолжает чтение файла, а когда он заканчивает чтение другой записи, он порождает второй рабочий поток для работы со второй записью. Это продолжается до тех пор, пока все записи не будут обработаны.

Проблема

К сожалению, я вижу проблему. Чтение «главного» потока будет работать намного быстрее, чем потоки, которые он порождает, что, боюсь, будет создано слишком много потоков. Чтобы предотвратить это, я вообразил это решение (в псевдокоде):

record = file.ReadRecord()
if(numberOfRunningWorkerThreads < MAX_THREADS)
    SpawnWorkerThread(record);
else
    WaitUntil(numberOfRunningWorkerThreads < MAX_THREADS)

Однако я понятия не имею, как реализовать такую ​​функциональность, особенно это последнее условие else. Я новичок в многопоточности и асинхронности, и я даже не уверен, в чем разница между этими двумя терминами.

Кто-нибудь может указать мне правильное направление?

Answer 1

Solution1:

Использовать ThreadPool . Установить MaxThreads , что

Устанавливает количество запросов к пулу потоков, которые могут быть активны одновременно. Все запросы выше этого числа остаются в очереди, пока потоки пула потоков не станут доступными.

Что-то вроде:

System.Threading.ThreadPool.SetMaxThreads(50, 1000);
// inside loop
ThreadPool.QueueUserWorkItem(ProcessRequest);
// end loop

Где ProcessRequest - ваш метод, который выполняет работу.

Solution2:

Если вы знаете количество записей: используйте Parallel.For и установите MaxDegreeOfParallelism соответственно.

Parallel.For(0, 1000, new ParallelOptions { MaxDegreeOfParallelism = 10 },
i => { 
    ProcessRequest(i);
});

Answer 2

Я полагаю, что вы ищете семафор (или, возможно, SemaphoreSlim может работать и для вас). Семафор «Ограничивает количество потоков, которые могут одновременно обращаться к ресурсу или пулу ресурсов». Семафор создан с определенным количеством слотов. Затем вы можете вызвать «WaitOne», чтобы дождаться доступного слота, и «Release», когда закончите со слотом. Если нет доступных слотов, «WaitOne» может либо ждать вечно, либо до истечения времени ожидания.

Таким образом, в вашем случае основной поток будет вызывать WaitOne для ожидания доступного слота. Затем в конце рабочих потоков вы можете вызвать Release, чтобы освободить слот.

Семафор .NET: https://msdn.microsoft.com/en-us/library/system.threading.semaphore(v=vs.110).aspx

.NET SemaphoreSlim (легкий семафор): https://msdn.microsoft.com/en-us/library/system.threading.semaphoreslim(v=vs.110).aspx