Как читать файл с чередованием одновременно, используя реактивные расширения
============================================================================

Я новичок в реактивных расширениях, и я хотел бы использовать его (в C #) для чтения файла, который содержит несколько потоков, которые чередуются. В основном файл в формате`ABCDABCDABCD...`, Я бы предпочел читать файл последовательно и разделять потоки (т.е.`AAA..`, `BBB..`и т. д.) и обрабатывать каждый поток параллельно, используя отдельные потоки для каждого потока.

Должна быть некоторая форма буферизации, чтобы гарантировать, что каждый поток может оставаться максимально занятым (конечно, в определенных пределах). Не все потоки обязательно начинаются одновременно, и в этом случае для задержанных потоков необходимо пропустить несколько элементов. В этом случае буферизация может сократить разрыв.

Элементы в файле маленькие (4 байта), поэтому они довольно болтливы. Поэтому я также ищу способ эффективно с этим справиться.

Я начал с создания перечислимого файла для чтения. Это может быть сделано для предоставления структуры, которая содержит идентификатор потока, или потоки могут быть разделены на основе порядка (номер элемента по модулю количества потоков). Последнее, вероятно, более эффективно, хотя.

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

Здесь способ превращения потока в`Enumerable<char>`, Поток будет применять буферизацию, это будет отправлять один результат за раз. Это можно сделать более эффективным (для отправки кусков данных назад), но в какой-то момент вам нужно обрабатывать их по одному, и это также может быть здесь. Не оптимизировать преждевременно.

Теперь предположим, что это код обработки для «вывода». наблюдаемые. Сначала я настраиваю выходные данные, а затем подписываюсь на них соответствующим образом. Обратите внимание, что я здесь использую массив, поэтому мой выходной наблюдаемый индекс - это индекс массива. Можно также использовать словарь, если индекс потока нельзя превратить в индекс, начинающийся с нуля.

Обратите внимание на использование`Subject<char>` чтобы отправить мои элементы на. Это зависит от типа вашего элемента, но`char` работает в приведенном примере. Также обратите внимание, что я задерживаю элементы только для того, чтобы доказать, что все работает. Теперь они являются независимыми потоками, и вы можете делать с ними все, что захотите.

Хорошо, учитывая поток файлов:

Теперь я могу подписаться и использовать индекс по модулю для отправки в правильный поток вывода:

Здесь есть потенциально более эффективные методы, в зависимости от того, как именно вы можете рассчитать целевой поток, но идея остается той же.

Входной файл содержит только одну строку:`ABCABCABCABCABCABC`

Выход из запуска программы:

Спустя секунду:

А потом еще секунду:



 основанное на ответе Ямена. Кажется, что он работает правильно, это означает, что последовательный чередующийся ввод разделен на несколько последовательных потоков, которые обрабатываются параллельно (многопоточность).

Тем не менее, я не уверен, что это правильная реализация (с точки зрения стиля программирования, RX-контрактов и т. Д.).