javascript - rxjs 进行的限速 http 调用

Question

我正在编写一项服务，人们可以在其中粘贴 Spotify 播放列表中的 url，然后将播放列表导出到不同的服务中。对于粘贴在请求中的每个轨道 url，需要向 Spotify api 发出请求。

这段代码:

        Rx.Observable.fromArray<ITrackIdentifier>( this._allTracks )
            .pluck<string>( "id" )
            .distinct()
            .flatMap(
                ( trackId ) => this.spotifyService.lookupTrack( trackId ).
                    catch( ( error ) => this.handleError( error ) ))
            .subscribe(
                ( result ) => this.handleTrackLookupResult( result ),
                ( error ) => this.handleError( error ),
                () => this.handleComplete()
            );

1. 从 ITrackIdentifiers 列表中创建一个 observable
2. 采用轨道标识符的 id 来创建字符串 (ids) 的可观察对象
3. 删除列表中所有重复的 ID
4. 为 spotify 的每个 http 调用创建一个 observable(并捕获错误)
5. 使用平面图将所有这些可观察结果合并到一个流中

除了添加大量轨道时，这实际上工作正常。我的一个示例播放列表有超过 500 首轨道，因此立即进行了 500 次调用，浏览器需要处理它们/从缓存中返回项目，因此浏览器速度很慢并锁定并且当我超过 api 调用限制时 spotify 返回大量错误.

我希望只能同时运行 10 个调用。 Merge with maxConcurrent set 似乎是在 Stackoverflow 上讨论的完美解决方案.

这看起来像这样:

        Rx.Observable.fromArray<ITrackIdentifier>( this._allTracks )
            .pluck<string>( "id" )
            .distinct()
            .map(
                ( trackId ) => this.spotifyService.lookupTrack( trackId ).
                    catch( ( error ) => this.handleError( error ) ))
            .merge(10)
            .subscribe(
                ( result ) => this.handleTrackLookupResult( result ),
                ( error ) => this.handleError( error ),
                () => this.handleComplete()
            );

但这根本行不通。在 chrome 网络调试器中，您可以看到同时进行的所有调用，并且大部分都在排队等待很长时间，直到它们失败。

为什么这不起作用？我还能如何解决这个问题？

这是 Github checkin与现阶段的项目:

Answer 1

您的代码使用 merge 的问题是 spotifyService.lookupTrack 不返回 Observable 而是 Promise。 Observable 的一些函数，如 flatMap 将 handle Promises as well ，但是 Observable 和 Promise 的区别在于 Observable 是惰性的，而 Promise 不是.您可以使用 Observable.defer 从 promise 工厂函数创建一个惰性可观察对象，正如 user3743222 所建议的那样。这个小例子是用 JavaScript 而不是 TypeScript 编写的，所以它可以在这里运行。

console.log = x => {var d = document,b=d.body,p=d.createElement('pre'); p.style.margin = "0"; p.appendChild(d.createTextNode(''+x)); b.appendChild(p);  window.scrollTo(0, b.scrollHeight); };

function log_delay(timeout, value) {
  return new Promise(resolve => {
    console.log('Start: ' + value);
    setTimeout(() => {
      console.log('End: ' + value);
      resolve(value);
    }, timeout);
  });
}

Rx.Observable.range(0, 6)
.map(x => Rx.Observable.defer(
  () => log_delay(1000, x)
  .catch(e => console.log('Inner catch'))
))
.merge(2)
.subscribe(
  s => console.log('Result: ' + s),
  s => console.log('Error: ' + s),
  s => console.log('Complete')
);

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/4.0.7/rx.all.js"></script>