javascript - การจำกัดอัตราการเรียก HTTP ที่ทำโดย rxjs

ฉันกำลังเขียนบริการที่ผู้คนสามารถวาง URL จากเพลย์ลิสต์ของ Spotify แล้วจากนั้นส่งออกเพลย์ลิสต์ไปยังบริการอื่น สำหรับแต่ละ URL ของเพลงที่ถูกวาง จะต้องมีการส่งคำขอไปยัง API ของ Spotify

โค้ดนี้:

        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. สร้าง observable จากรายการของ ITrackIdentifiers
2. ใช้รหัสของตัวระบุแทร็กเพื่อสร้าง observable ของสตริง (รหัส)
3. ลบ id ที่ซ้ำกันใด ๆ ในรายการ
4. สร้าง observable สำหรับการเรียก http แต่ละครั้งที่ทำไปยัง Spotify (และจับข้อผิดพลาด)
5. รวมผลลัพธ์จาก observable เหล่านี้ทั้งหมดเป็นสตรีมเดียวด้วย flatmap

สิ่งนี้ทำงานได้ดีจริง ๆ ยกเว้นเมื่อมีการเพิ่มแทร็กจำนวนมาก ตัวอย่างเพลย์ลิสต์ของฉันมีแทร็กมากกว่า 500 เพลง ดังนั้นทันทีจะมีการเรียก 500 ครั้งและเบราว์เซอร์ต้องจัดการกับพวกมัน/คืนค่ารายการจากแคช ทำให้เบราว์เซอร์ช้าและค้าง และ Spotify จะส่งข้อผิดพลาดจำนวนมากเพราะฉันเกินขีดจำกัดการเรียก API

ฉันต้องการที่จะสามารถมีการโทรได้เพียง 10 ครั้งพร้อมกัน การรวมกับการตั้งค่า maxConcurrent ดูเหมือนจะเป็นวิธีแก้ปัญหาที่สมบูรณ์แบบตามที่ได้มีการพูดคุยกันใน 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 พร้อมโครงการในขั้นตอนนี้:

คำตอบ #1

ปัญหาของโค้ดของคุณที่ใช้ merge คือ spotifyService.lookupTrack ไม่คืนค่าเป็น Observable แต่คืนค่าเป็น Promise ฟังก์ชันบางอย่างของ Observable เช่น flatMap สามารถจัดการกับ Promise ได้เช่นกัน แต่ความแตกต่างระหว่าง Observable และ Promise คือ Observable เป็น lazy ในขณะที่ Promise ไม่เป็น คุณสามารถสร้าง observable แบบ lazy จากฟังก์ชัน factory ของ promise โดยใช้ Observable.defer ตามที่ 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>

คำตอบ #2

ฉันสามารถทำให้มันทำงานได้บางส่วนตามที่ฉันต้องการแล้ว แต่ฉันยังคงสงสัยว่าทำไม merge ถึงไม่ทำงาน ที่นี่รายชื่อของรหัสที่ไม่ซ้ำถูกสร้างขึ้น แล้วเราก็ใช้ concatMap เพื่อสร้าง Observable สำหรับรหัสแต่ละตัว จากนั้นรอช่วงเวลาหนึ่งก่อนที่จะไปยังรายการถัดไป:

        Rx.Observable.fromArray<ITrackIdentifier>( this._allTracks )
            .pluck<string>( "id" )
            .distinct()
            .concatMap( ( id, index ) => Rx.Observable.interval( 50 ).take( 1 ).map( () => { return id } ) )
            .flatMap(
                ( trackId ) => this.spotifyService.lookupTrack( trackId ).
                catch( ( error ) => this.handleError( error ) ))
            .subscribe(
                ( result ) => this.handleTrackLookupResult( result ),
                ( error ) => this.handleError( error ),
                () => this.handleComplete()
            );

ในตัวอย่างนี้ฉันรอ 50 มิลลิวินาทีระหว่างการเรียกแต่ละครั้ง ซึ่งช่วยลดข้อผิดพลาดได้มาก

นี่คือการเช็คอินของ Github พร้อมกับโครงการในขั้นตอนนี้