如何使用反应式扩展同时读取交错文件

我是反应式扩展的新手，我想使用它（在 c# 中）来读取包含多个交错流的文件。基本上文件的格式是ABCDABCDABCD...。我更喜欢按顺序读取文件并分离流（即AAA.., BBB..等）并并行处理每个流，为每个流使用单独的线程。

必须有某种形式的缓冲来确保每个流可以尽可能保持忙碌（当然在限制范围内）。并非所有流都必须同时开始，在这种情况下，必须为延迟的流跳过许多元素。在这种情况下，缓冲可能会弥补这一差距。

文件中的元素很小（4 个字节），因此内容相当丰富。因此，我也在寻找一种有效的方法来处理这个问题。

我首先创建一个可枚举来读取文件。这可以提供一个包含流 ID 的结构，或者可以根据顺序（元素号对流数取模）来分隔流。不过后者可能更有效。

这个问题到处都是“这取决于”，特别是当你谈论性能和效率但提供了一个有点做作的例子时。也就是说，与真实文件相比，您的示例文件非常简单。不过，如果有用的话，我会尝试提供一些建议。

这是将流转换为流的方法Enumerable<char>。流将应用缓冲，这将一次发送一个结果。这可以提高效率（发送回数据块），但在某些时候您需要一次处理一个数据块，它也可能在这里。不要过早优化。

IEnumerable<char> ReadBytes(Stream stream)
{
    using (StreamReader reader = new StreamReader(stream))
    {
        while (!reader.EndOfStream)
            yield return (char)reader.Read();
    }
}

现在，假设这是“输出”可观察量的处理代码。首先，我设置输出可观察量，然后根据需要订阅它们。请注意，我在这里使用数组，因此我的输出可观察索引是数组索引。如果流索引无法转换为从零开始的索引，也可以使用字典。

var outputs = Enumerable.Repeat(0, 3).Select(_ => new Subject<char>()).ToArray();                                                                                                     

outputs[0].Delay(TimeSpan.FromSeconds(2)).Subscribe(x => Console.WriteLine("hi: {0}", x));
outputs[1].Delay(TimeSpan.FromSeconds(1)).Subscribe(x => Console.WriteLine("ho: {0}", x));
outputs[2].Subscribe(x => Console.WriteLine("he: {0}", x));

注意使用Subject<char>发送我的元素。这取决于元素的类型，但是char在给出的示例中有效。另请注意，我延迟元素只是为了证明一切正常。它们现在是独立的流，您可以对它们做任何您想做的事情。

好的，给定一个文件流：

var file = @"C:\test.txt";
var buffer = 32;
var stream = new FileStream(file, FileMode.Open, FileAccess.Read, FileShare.Read, buffer);

我现在可以订阅并使用模索引发送到正确的输出流：

ReadBytes(stream)
.ToObservable(Scheduler.ThreadPool)
.Select((x,i) => new { Key = (i % 3), Value = x }) // you can change it up here
.Subscribe(x => outputs[x.Key].OnNext(x.Value));

这里可能有更有效的方法，具体取决于您如何计算目标流，但想法保持不变。

输入文件仅包含一行：ABCABCABCABCABCABC

运行程序的输出是：

he: C
he: C
he: C
he: C
he: C
he: C

一秒钟后：

ho: B
ho: B
ho: B
ho: B
ho: B
ho: B

然后又一秒钟：

hi: A
hi: A
hi: A
hi: A
hi: A
hi: A