c#使用流式数据处理

在 C# 中，流式数据处理通常利用 .NET Framework 的 `System.IObservable` 和 `System.IObserver` 接口来实现，这是基于观察者模式的一种设计。以下是一些关键点：

1. **Observable**：这表示一个可观察的数据源，如事件、网络请求或文件读取操作。你可以创建一个 Observable 对象，它会按照一定的时间间隔或数据变化发出通知。

using System.IO;
using System.Linq;

var observable = Observable.FromEventPattern<EventArgs>(fileStream, "DataAvailable");

2. **Observer**：这是一个接口，用于处理 Observable 发送的数据。当数据可用时，Observer 会被调用更新方法。

class DataConsumer : IObserver<byte[]>
{
    public void OnNext(byte[] data)
    {
        // 数据到达时处理这里
    }

    public void OnError(Exception error)
    {
        // 错误发生时处理这里
    }

    public void OnCompleted()
    {
        // 所有数据处理完成时处理这里
    }
}

3. **LINQ to Events**：如果你正在处理的是事件，C# 的 LINQ to Events 功能可以帮助简化 Observable 的创建。

dataObservable.Subscribe(new DataConsumer());

4. **Rx.NET**：Reactive Extensions (Rx) 是一个强大的库，提供了更高级的功能来处理异步数据流，包括过滤、映射、合并等。

using System.Reactive.Linq;

observable.Throttle(TimeSpan.FromSeconds(1)) // 每隔一秒处理一次数据
    .Where(data => data.Length > 100) // 过滤掉长度小于100的数据
    .Subscribe(data => Console.WriteLine($"Received: {BitConverter.ToString(data)}"));

流式处理在 C# 中有助于减少内存占用，提高性能，特别是在处理大量或实时数据时尤其有用。