我和同事有争执。我们正在编写一个处理大量数据的.NET应用程序。它接收数据元素，根据一些标准将它们的子集分组为块，并处理这些块。
假设我们有Foo类型的数据项一个接一个地到达某个源（例如，来自网络）。我们希望收集Foo类型的相关对象的子集，从每个这样的子集构造Bar类型的对象，并处理Bar类型的对象。
我们中的一个人提出了以下设计。它的主要主题是直接从组件的接口公开IObservable<T>对象。
// ********* Interfaces **********
interface IFooSource
{
// this is the event-stream of objects of type Foo
IObservable<Foo> FooArrivals { get; }
}
interface IBarSource
{
// this is the event-stream of objects of type Bar
IObservable<Bar> BarArrivals { get; }
}
/ ********* Implementations *********
class FooSource : IFooSource
{
// Here we put logic that receives Foo objects from the network and publishes them to the FooArrivals event stream.
}
class FooSubsetsToBarConverter : IBarSource
{
IFooSource fooSource;
IObservable<Bar> BarArrivals
{
get
{
// Do some fancy Rx operators on fooSource.FooArrivals, like Buffer, Window, Join and others and return IObservable<Bar>
}
}
}
// this class will subscribe to the bar source and do processing
class BarsProcessor
{
BarsProcessor(IBarSource barSource);
void Subscribe();
}
// ******************* Main ************************
class Program
{
public static void Main(string[] args)
{
var fooSource = FooSourceFactory.Create();
var barsProcessor = BarsProcessorFactory.Create(fooSource) // this will create FooSubsetToBarConverter and BarsProcessor
barsProcessor.Subscribe();
fooSource.Run(); // this enters a loop of listening for Foo objects from the network and notifying about their arrival.
}
}
另一个提出了另一种设计，其主题是使用我们自己的发布者/订阅者接口，并仅在需要时在实现中使用Rx。
//********** interfaces *********
interface IPublisher<T>
{
void Subscribe(ISubscriber<T> subscriber);
}
interface ISubscriber<T>
{
Action<T> Callback { get; }
}
//********** implementations *********
class FooSource : IPublisher<Foo>
{
public void Subscribe(ISubscriber<Foo> subscriber) { /* ... */ }
// here we put logic that receives Foo objects from some source (the network?) publishes them to the registered subscribers
}
class FooSubsetsToBarConverter : ISubscriber<Foo>, IPublisher<Bar>
{
void Callback(Foo foo)
{
// here we put logic that aggregates Foo objects and publishes Bars when we have received a subset of Foos that match our criteria
// maybe we use Rx here internally.
}
public void Subscribe(ISubscriber<Bar> subscriber) { /* ... */ }
}
class BarsProcessor : ISubscriber<Bar>
{
void Callback(Bar bar)
{
// here we put code that processes Bar objects
}
}
//********** program *********
class Program
{
public static void Main(string[] args)
{
var fooSource = fooSourceFactory.Create();
var barsProcessor = barsProcessorFactory.Create(fooSource) // this will create BarsProcessor and perform all the necessary subscriptions
fooSource.Run(); // this enters a loop of listening for Foo objects from the network and notifying about their arrival.
}
}
你认为哪一个更好？公开IObservable<T>并使我们的组件从Rx运营商创建新的事件流，或者定义我们自己的发布者/订阅者接口并在需要时在内部使用Rx？
以下是关于设计需要考虑的一些事项：
在第一个设计中，我们接口的消费者可以在他/她的指尖上使用Rx的全部功能，并且可以执行任何Rx操作。我们中的一个声称这是一个优势，另一个则声称这是缺点。
第二种设计允许我们在后台使用任何发布者/订阅者架构。第一个设计将我们与Rx联系在一起。
如果我们希望使用Rx的功能，它需要在第二个设计中做更多的工作，因为我们需要将自定义发布者/订阅者实现转换为Rx并返回。它需要为每个希望进行一些事件处理的类编写粘合代码。
暴露IObservable<T>不会以任何方式污染Rx的设计。事实上，设计决策与公开老派.NET事件或推出自己的pub/sub机制之间的挂起完全相同。唯一的区别是IObservable<T>是一个较新的概念。
需要证据吗？看看F#，它也是一种.NET语言，但比C#年轻。在F#中，每个事件都源自IObservable<T>。老实说，我认为抽象一个非常合适的.NET pub/sub机制（即IObservable<T>）与您自己开发的pub/sub抽象是没有意义的。只需暴露IObservable<T>即可。
对我来说，滚动自己的pub/sub抽象就像是将Java模式应用到.NET代码中。不同的是，在.NET中，一直都有很好的框架支持Observer模式，根本不需要滚动自己的模式。
首先，值得注意的是，IObservable<T>是mscorlib.dll和System命名空间的一部分，因此公开它在某种程度上等同于公开IComparable<T>或IDisposable。这相当于选择.NET作为您的平台，而您似乎已经这样做了。
现在，我不想提出答案，而是想提出一个不同的问题，然后是一种不同的心态，我希望（并相信）你会从那里得到解决。
你基本上是在问：我们想在整个系统中推广Rx运营商的分散使用吗。现在很明显，这不是很有吸引力，因为你可能在概念上把Rx视为第三方库。
不管怎样，答案不在于你们两个提出的基本设计，而在于这些设计的用户。我建议将您的设计分解为抽象级别，并确保Rx运算符的使用范围仅限于一个级别。当我谈到抽象级别时，我指的是类似于OSI模型的东西，只是在同一应用程序的代码中。
在我的书中，最重要的是不要采取的设计立场"让我们创造一些将在整个系统中使用和分散的东西，因此我们需要确保在未来的岁月里只做一次，而且做得恰到好处"。我更喜欢"让我们让这个抽象层产生其他层所需的最小API，以当前实现其目标"。
关于这两种设计的简单性，实际上很难判断，因为Foo和Bar并没有告诉我太多关于用例的信息，因此也没有告诉我可读性因素（顺便说一句，不同用例的可读性因素不同）。
在第一个设计中，我们接口的消费者可以在他/她的指尖上使用Rx的全部功能，并且可以执行任何Rx操作。我们中的一个声称这是一个优势，另一个则声称这是缺点。
我同意Rx的可用性是一种优势。列出它是一个缺点的一些原因可能有助于确定如何解决这些问题。我看到的一些优势是：
- 正如Yam和Christoph所反对的那样，IObservable/IObserver在.NET 4.0的mscorlib中，因此它将（有望）成为每个人都会立即理解的标准概念，就像事件或IEnumerable一样
- Rx。一旦您需要组合、过滤或以其他方式操作潜在的多个流，这些都会非常有用。你可能会发现自己用自己的界面以某种形式重做了这项工作
- Rx。Rx库执行一个定义良好的合同，并尽可能多地执行该合同。即使您需要自己制定运营商，
Observable.Create也会执行合同（这就是为什么Rx团队不建议直接实施IObservable） - Rx库有很好的方法来确保您在需要时使用正确的线程
我已经写了我的操作员部分，图书馆没有涵盖我的情况。
第二种设计允许我们在后台使用任何发布者/订阅者架构。第一个设计将我们与Rx联系在一起。
我看不出暴露Rx的选择对如何在后台实现架构的影响（如果有的话）比使用自己的接口更大。我断言，除非绝对必要，否则你不应该发明新的酒吧/酒吧架构。
此外，Rx库可能具有可简化"发动机罩下"部件的操作员。
如果我们希望使用Rx的功能，它需要在第二个设计中做更多的工作，因为我们需要将自定义发布者/订阅者实现转换为Rx并返回。它需要为每个希望进行一些事件处理的类编写粘合代码。
是和否。如果我看到第二个设计，我会想到的第一件事是："这几乎就像IObservable；让我们写一些扩展方法来转换接口。"粘合代码只写一次，到处都用。
粘合代码很简单，但如果你认为你会使用Rx，只需公开IOobservable，省去麻烦。
进一步考虑
基本上，您的备选设计在3个关键方面与IObservable/IObserver不同。
- 无法取消订阅。这可能只是复制到问题时的疏忽。如果没有，那么如果你走这条路，就应该强烈考虑添加
- 没有定义错误向下游流动的路径（例如
IObserver.OnError） - 没有办法指示流（例如
IObserver.OnCompleted）的完成。只有当您的基础数据打算有一个终止点时，这才是相关的
您的替代设计也将回调作为操作返回，而不是将其作为接口上的方法，但我认为区别并不重要。
Rx库鼓励采用功能性方法。FooSubsetsToBarConverter类更适合作为返回IObservable<Bar>的IObservable<Foo>的扩展方法。这稍微减少了混乱（为什么要创建一个只有一个属性的类，而函数做得很好），并且更适合Rx库其余部分的链式组合。您可以将相同的方法应用于备用接口，但如果没有操作员的帮助，这可能会更加困难。
另一种选择可能是：
interface IObservableFooSource : IFooSource
{
IObservable<Foo> FooArrivals
{
get;
}
}
class FooSource : IObservableFooSource
{
// Implement the interface explicitly
IObservable<Foo> IObservableFooSource.FooArrivals
{
get
{
}
}
}
这样，只有期望IObservableFooSource的客户端才会看到RX特定的方法，而那些期望IFooSource或FooSource的客户则不会看到。