使用响应式扩展并行调用 2 个 httpwebrequests答案

【问题标题】：Calling 2 httpwebrequests in parallel using Reactive Extensions使用响应式扩展并行调用 2 个 httpwebrequests
【发布时间】：2012-09-20 08:39:46
【问题描述】：

我正在尝试并行调用 2 个 httpwebrequest 并让它们在使用 Rx 扩展完成后调用相同的回调。但我不知道我怎么能做到这一点..这是我的代码：

    private static IObservable<Stream> GetImage(string path)
    {
        var uri = new Uri(path);
        var thumburi = new Uri(path + "_thumb.jpg");
        return Observable.Create<Stream>(o =>
                                                  {
                                                        var request = (HttpWebRequest) HttpWebRequest.Create(uri);
                                                        var readComplete =
                                                          Observable.FromAsyncPattern<WebResponse>(
                                                              request.BeginGetResponse,
                                                              request.EndGetResponse)();
                                                        var subscription = readComplete
                                                          .Select(e => e.GetResponseStream())
                                                          .Subscribe(o);



                                                      return subscription;

                                                  });
    }

【问题讨论】：

所以我知道这行得通... GetImage(blobUrl+"_thumb.jpg").Subscribe(callback); GetImage(blobUrl).Subscribe(回调);但我对此不满意

标签： httpwebrequest system.reactive

【解决方案1】：

使用最新的位和 .Net 4.5，您可以执行以下操作：

private static IObservable<byte[]> GetImages(string path)
{
    var sources = new Uri[]
    {
        var uri = new Uri(path),
        var thumburi = new Uri(path + "_thumb.jpg")
    };

    var obs = from uri in sources.ToObservable()
              from data in Observable.Using(
                   () => new WebClient(),
                   client => client.DownloadDataTaskAsync(uri).ToObservable())
              select data;

    return obs;
}

我确实想知道您是否真的只想返回数据流，而不关心哪个流对应于基础，哪个是缩略图。并行发出请求后，您将不再控制它们返回的顺序。如果需要，您可以投影一个包含 uri 和数据流的类型以消除它们的歧义。

【讨论】：

您需要使用Observable.Using 来确保WebClient 被处理掉。
@Enigmativity，谢谢。我用 Observable.Using 子句更新了这个例子。

【解决方案2】：

我猜你会将异步调用拉出到两个单独的流，然后将它们连接起来，不是吗？像这样：http://leecampbell.blogspot.com/2010/06/rx-part-5-combining-multiple.html

【讨论】：

很好的链接！但是，我认为 Concat 将失去 Rui 正在寻找的并行行为。 SelectMany 可能是他正在寻找的东西。请参阅下面的 Jim 或 Enigmativity 的答案。

【解决方案3】：

我建议这种解决方案。

使GetImage 更通用：

private static IObservable<Stream> GetImage(Uri uri)
{
    return Observable.Create<Stream>(o =>
    {
        var request = (HttpWebRequest)HttpWebRequest.Create(uri);
        var readComplete =
            Observable.FromAsyncPattern<WebResponse>(
                request.BeginGetResponse,
                request.EndGetResponse)();
        var subscription =
            readComplete
                .Select(e => e.GetResponseStream())
                .Subscribe(o);
        return subscription;
    });
}

然后添加一个特定的GetImages 方法来查询图像及其拇指：

private static IObservable<Tuple<Uri, Stream>> GetImages(string path)
{
    var uris = new []
    {
        new Uri(path + ".jpg"),
        new Uri(path + "_thumb.jpg"),
    }.ToObservable();

    return
        from uri in uris
        from stream in GetImage(uri)
        select Tuple.Create(uri, stream);
}

我假设您的 path 变量不能包含“.jpg”扩展名，否则您必须进行一些字符串操作。

现在GetImages 返回一个IObservable<Tuple<Uri, Stream>>，因为SelectMany 不保证流的返回顺序，所以我们需要使用Uri 来消除流的歧义。

让我知道这是否适合你。

【讨论】：

【解决方案4】：

为什么不直接使用 Zip？

GetStream("foo.jpg").Zip(GetStream("bar.jpg"), (foo, bar) => new { foo, bar })
    .Subscribe(fooAndBar => ...);

【讨论】：