问题标签 [system.reactive]

我正在考虑使用 Rx（反应式框架）来异步查询给定关键字的 Twitter 搜索 API。我想构建的示例应用程序应该能够在控制台中显示新推文。

你认为有可能做到这一点吗？它会比使用标准编程技术更简单吗？

你会怎么做？

谢谢 ！杰里米

更新

看起来 Jon Skeet 是对的（大惊喜！），问题在于我对Average扩展提供连续平均值的假设（它没有）。

对于我所追求的行为，我编写了一个简单的ContinuousAverage扩展方法，我在这里包括了它的实现，以便其他可能想要类似东西的人受益：

我正在考虑继续为其他明显的候选人做类似上述的事情——所以, ContinuousCount, ContinuousSum, ContinuousMin, ContinuousMax... 也许ContinuousVariance还有ContinuousStandardDeviation？对此有什么想法吗？

原始问题

我到处使用Rx Extensions ，感觉我已经掌握了基本的想法。

现在有些奇怪：我的印象是，如果我写了这个：

我会得到两个IObservable<double>对象（bids和avgOfBids）；一个基本上是来自 my 的所有市场出价流MarketDataProvider，另一个是这些出价的平均值流。

所以是这样的：

看来我的avgOfBids对象没有做任何事情。我错过了什么？我想我可能误解了Average实际应该做什么。（这似乎也适用于所有类似聚合的扩展方法IObservable<T>——例如Max，、Count等）

在一个IObservable序列中（在 Reactive Extensions for .NET 中），我想获取先前和当前元素的值，以便我可以比较它们。我在网上找到了一个类似于下面的示例，它完成了任务：

它工作正常，除了它两次评估序列，我想避免。您可以看到使用以下代码对其进行了两次评估：

输出显示的调试行数是序列中元素数的两倍。

我理解为什么会发生这种情况，但到目前为止，我还没有找到不评估序列两次的替代方案。如何仅通过一个序列评估将先前和当前结合起来？

我正在尝试了解反应式扩展 (Rx) 的正确用例。不断出现的示例是 UI 事件（拖放、绘图），以及 Rx 适用于异步应用程序/操作（如 Web 服务调用）的建议。

我正在开发一个需要为 REST 服务编写小型客户端 API 的应用程序。我需要调用四个 REST 端点，其中三个是为了获取一些参考数据（机场、航空公司和状态），第四个是主要服务，可为您提供给定机场的航班时间。

我创建了公开三个参考数据服务的类，方法如下所示：

在我的 GetFlights 方法中，我希望每个航班都保存一个参考，它是从机场出发的，以及运营该航班的航空公司。为此，我需要 GetAirports 和 GetAirlines 提供的数据。每个机场、航空公司和状态都将被添加到一个字典（即字典）中，以便我可以在解析每个航班时轻松设置参考。

我当前的实现现在看起来像这样：

当前的实现基于 Sergey 的回答，并使用 ForkJoin 来确保顺序执行，并且我引用的数据在航班之前被加载。这个实现比我之前的实现必须触发“ReferenceDataLoaded”事件更优雅。

有人可以解释 Observable 和 ConnectableObservable 之间的区别吗？Rx Extensions 文档非常稀少，我不明白 ConnectableObservable 在什么情况下有用。

此类用于重播/修剪方法。

Is there any equivalent of Reactive Extensions (.NET) for Java?

About Rx (Reactive Extensions)

Rx is a library for composing asynchronous and event-based programs using observable collections.

I am aware of rule engines such as Drools from JBOSS, but is there some other way that is closer to the Microsoft .NET approach?

可能重复：
反应式扩展使用的好例子

我已经玩了一段时间的响应式扩展，但主要限于在 WPF 前端处理/组合用户驱动的事件。

这是一种非常强大的异步编程新方法，我很好奇其他人在用它做什么，你认为它可以在哪里改进我们目前做事的方式？

这就是我的意思。假设我正在使用一个公开事件的 API，但这些事件不遵循标准EventHandler或EventHandler<TEventArgs>签名。一个事件可能如下所示，例如：

现在，通常，如果我想IObservable<TEventArgs>从事件中获取一个，我可以这样做：

但这行不通，因为Update事件不是EventHandler<UpdateEventArgs>——事实上，没有——它UpdateEventArgs基本上只是它自己的事情。

显然，我可以定义我自己的派生自EventArgs(ie, UpdateEventArgs) 的类，编写另一个类来包装提供Update事件的对象，为包装类提供它自己的 Update事件，即an EventHandler<UpdateEventArgs>，并从中IObservable<UpdateEventArgs>获取an。但这是令人讨厌的工作量。

有什么方法可以IObservable<[something]>从这样的“非标准”事件中创建一个，还是我不走运？

更新：从乔恩斯基特的回答中，我被推向以下重载的方向Observable.FromEvent：

不过，我不得不承认，我很难理解那Func(Of EventHandler(Of TEventArgs), TDelegate)部分。这对我来说似乎倒退了（？）。显然，我只是缺少一些东西......

无论如何，如果它有帮助，我认为这就是等效的 C# 代码的样子（老实说：我不确定这一点。尽管我自己通常更喜欢 C#，但这段代码是一个人的工作我的同事，他们主要用 VB.NET 编写；而 VB.NET 允许使用多种语法来声明事件）：

这里棘手的部分是，似乎某种派生类EventArgs是必要的，无论如何。在我正在使用的 API 中，没有这样的类。所以，最起码，我得写一个。但这应该是相当微不足道的（基本上是一个属性：）BaseEvent。

最后，我假设这个重载所需的代码在 C# 中看起来像这样：

首先：我有这个直吗？其次：我是否正确地说使用 VB 9，如果不编写自己的方法就真的无法完成上述任务吗？

我几乎觉得我一开始就从完全错误的角度来处理这个问题。但我真的不确定。

在 Reactive Extensions 的 System.Interactive.dll (v1.0.2521.0) 中，EnumerableEx既有Memoize方法又有MemoizeAll方法。两者的 API 文档是相同的：

创建一个枚举原始枚举仅一次并缓存其结果的枚举。

然而，这些方法显然不相同。如果我使用Memoize，我的 enumerable 在我第一次枚举它时具有值，而第二次似乎是空的。如果我使用MemoizeAll，那么我会从任何一种方法的描述中得到我期望的行为——我可以根据需要多次枚举结果，并且每次都得到相同的结果，但源只枚举一次。

谁能告诉我这些方法之间的预期区别是什么？用例是Memoize什么？这似乎是一个相当无用的方法，文档非常混乱。

我想充分利用.NET (Rx) 的反应式扩展，并希望获得一些关于执行一些基本任务的输入。为了说明我想要做什么，我有一个人为的例子，我有一个带有异步事件的外部组件：

该组件通过调用启动BeginStart()。此方法立即返回，稍后，当组件完成启动时，将触发该Started事件。

我想通过包装组件来创建一个同步启动方法并等待Started事件被触发。到目前为止，这是我想出的：

我想摆脱ManualResetEvent，并且我希望 Rx 有一个解决方案。但是怎么做？