c# - 从 Web 查询集合中可观察到的响应式扩展

Question

相关： 列表上的异步函数

我有一个看起来像这样的函数：

public async Task<decimal> GoToWeb(string Sym){}

我想每分钟一次通过字符串列表定期调用它。如何将其转换为反应式扩展 Observable？

Answer 1

由于反应式框架可以为您处理异步，您可以尝试以下操作：

var syms = new List<string>() { "ANZ", "BHP", };

var query =
    from i in Observable.Interval(TimeSpan.FromSeconds(1.0))
    from sym in syms.ToObservable()
    from d in GoToWeb(sym).ToObservable()
    select new
    {
        Symbol = sym,
        Value = d,
    };

您需要添加对命名空间的引用以System.Reactive.Threading.Tasks获取ToObservable()任务的扩展。

这是否满足您的需求？

Answer 2

我是一个 Rx 菜鸟，所以可能有更好的选择，但是在不知道您希望为每批完成的网络调用发生什么的情况下，我避免使用 linq 语法（尽管我更喜欢在可能的情况下使用它）。

我在 LINQPad 中使用 Rx-Main(Prerelease) 的 NuGet 包进行了此操作

void Main()
{
    var eachMinuteSequence = Observable.Timer(TimeSpan.Zero, TimeSpan.FromMinutes(1));

    var symbols = new[] { "GOOG", "MSFT", "AAPL" };

    Action<long> eachMinuteAction = async _ =>
    {
        var tasks = StartWebCalls(symbols);
        var results = await Task.WhenAll(tasks);
        // do something with the results
    };
    eachMinuteSequence.Subscribe(eachMinuteAction);
}

// Define other methods and classes here
public Task<decimal>[] StartWebCalls(IEnumerable<string> stockSymbols)
{
    return (
        from symbol in stockSymbols
        select GoToWeb(symbol)
    ).ToArray();
}

public async Task<decimal> GoToWeb(string Sym){throw new NotImplementedException();}