使用 rxjs 和旧式回调的同步问题

Question

我有带有 4 个异步函数的老式库 - 我们称它们为 libA、libB、libC、libD（每个函数的运行时间未知/随机） - 当它们完成时，它们调用我的回调

我编写了 loadString 函数，它返回 observable（它应该包含每个回调函数的结果 - 为了简单起见，我在这里使用字符串，但它可以是结果数组）但它给出了错误的结果“AADDDDD”。预期结果是“AADDDDDCCCCBBB”。如何正确使用 rxjs？

function loadString() {
  let result = '';
  
  return Observable.create(observer =>{
    libA( n=> { result+='A'.repeat(n) });
    libB( n=> { result+='B'.repeat(n) });
    libC( n=> { result+='C'.repeat(n) });
    libD( n=> {
      result+='D'.repeat(n);
      observer.next(result);
    });        
  })
}

下面有工作的 sn-p，您可以将其复制到您的答案并开发 loadString 函数

// SET-UP
const { of, Observable } = rxjs;
const { map, switchMap, delay } = rxjs.operators;

// simulated lib function - not change this (the times are random)
function libA(callback) { setTimeout( _=>callback(2), 1000); } 
function libB(callback) { setTimeout( _=>callback(3), 3000); }
function libC(callback) { setTimeout( _=>callback(4), 2000); }
function libD(callback) { setTimeout( _=>callback(5), 1500); }


// QUESTION: how to write below function body using rxjs?
function loadString() {
  let result = '';
  
  return Observable.create(observer =>{
    libA( n=> { result+='A'.repeat(n) });
    libB( n=> { result+='B'.repeat(n) });
    libC( n=> { result+='C'.repeat(n) });
    libD( n=> {
      result+='D'.repeat(n);
      observer.next(result);
    });        
  })
}


// TEST
let s = loadString().subscribe(str=> {
   console.log(str); // wrong result: I expected "AADDDDDCCCCBBB" value 
})

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.5/rxjs.umd.min.js" integrity="sha256-85uCh8dPb35sH3WK435rtYUKALbcvEQFC65eg+raeuc=" crossorigin="anonymous"></script>

UPDAE

我不想按顺序运行 lib 函数，而是并行运行（它们向 API 发送一些请求...）

Answer 1

一个说明您的问题的组合解决方案 - 您应该只在调用最后一次回调后调用 observer.next。只是保持运行计数。下面的代码并不重要，关键是“等到最后一个被调用后再发出”的知识。

这是一个有趣的解决方案：

// SET-UP
const { of, Observable } = rxjs;
const { map, switchMap, delay } = rxjs.operators;

// simulated lib function - not change this (the times are random)
function libA(callback) { setTimeout( _=>callback(2), 1000); } 
function libB(callback) { setTimeout( _=>callback(3), 3000); }
function libC(callback) { setTimeout( _=>callback(4), 2000); }
function libD(callback) { setTimeout( _=>callback(5), 1500); }


// QUESTION: how to write below function body using rxjs?
function loadString() {
  let result = '';

  const done = (observer,count) => val => {
      result += val;
      // THIS IF STATEMENT IS WHAT YOU WERE MISSING
      if(!--count) return observer.next(result);
  }
  
  return Observable.create(observer =>{
    const complete = done(observer,4);
    libA( n=> complete('A'.repeat(n))),
    libB( n=> complete('B'.repeat(n))),
    libC( n=> complete('C'.repeat(n))),
    libD( n=> complete('D'.repeat(n)));
  })
}


// TEST
let s = loadString().subscribe(str=> {
   console.log(str); // wrong result: I expected "AADDDDDCCCCBBB" value 
})

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.5/rxjs.umd.min.js" integrity="sha256-85uCh8dPb35sH3WK435rtYUKALbcvEQFC65eg+raeuc=" crossorigin="anonymous"></script>

Answer 2

从前一个的回调中调用每个libX 函数。

function loadString() {
  let result = '';

  return Observable.create(observer => {
    libA(n => {
      result += 'A'.repeat(n);
      libB(n => {
        result += 'B'.repeat(n);
        libC(n => {
          result += 'C'.repeat(n);
          libD(n => {
            result += 'D'.repeat(n);
            observer.next(result);
          });
        });
      });
    })
  });
}

Answer 3

libD 是唯一调用observer.next 的东西。一旦完成，即使其他的还没有完成，你的 observable 也会发出。所以你需要等待。试试combineLatest（虽然这不会保持秩序）？

// SET-UP
const { of, Observable, combineLatest } = rxjs;
const { map, switchMap, delay } = rxjs.operators;

// simulated lib function - not change this (the times are random)
function libA(callback) { setTimeout( _=>callback(2), 1000); } 
function libB(callback) { setTimeout( _=>callback(3), 3000); }
function libC(callback) { setTimeout( _=>callback(4), 2000); }
function libD(callback) { setTimeout( _=>callback(5), 1500); }


// QUESTION: how to write below function body using rxjs?
function loadString() {
  let result = '';
   
   return combineLatest(
     Observable.create(observer => libA(n => observer.next('A'.repeat(n)))),
     Observable.create(observer => libB(n => observer.next('B'.repeat(n)))),   
     Observable.create(observer => libC(n => observer.next('C'.repeat(n)))),          
     Observable.create(observer => libD(n => observer.next('D'.repeat(n))))
   );
 
}


// TEST
let s = loadString().subscribe(str=> {
   console.log(str); // wrong result: I expected "AADDDDDCCCCBBB" value 
})

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.5/rxjs.umd.min.js" integrity="sha256-85uCh8dPb35sH3WK435rtYUKALbcvEQFC65eg+raeuc=" crossorigin="anonymous"></script>

Answer 4

我们也可以使用bufferCount（技术来自here）

// SET-UP
const { Subject } = rxjs;
const { take, bufferCount, map, finalize } = rxjs.operators;

// simulated lib function - not change this (the times are random)
function libA(callback) { setTimeout( _=>callback(2), 1000); } 
function libB(callback) { setTimeout( _=>callback(3), 3000); }
function libC(callback) { setTimeout( _=>callback(4), 2000); }
function libD(callback) { setTimeout( _=>callback(5), 1500); }


function loadString() {
  let result = new Subject();
  
  libA( n=> { result.next('A'.repeat(n)); });     
  libB( n=> { result.next('B'.repeat(n)); });     
  libC( n=> { result.next('C'.repeat(n)); });     
  libD( n=> { result.next('D'.repeat(n)); });        

  return result.pipe(
    bufferCount(4),       // num of callbacks
    take(1),              // this will unsubscribe after 4 values received
    map( r => r.join`` ),
  );
}


// TEST
let s = loadString().subscribe(str=> {
   console.log(str); // wrong result: I expected "AADDDDDCCCCBBB" value 
}, e=>{}, ()=> console.log('final'))

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.5/rxjs.umd.min.js" integrity="sha256-85uCh8dPb35sH3WK435rtYUKALbcvEQFC65eg+raeuc=" crossorigin="anonymous"></script>

或效率较低的scan（因为它连接字符串不是一次而是每次某些任务结束时）

// SET-UP
const { Subject } = rxjs;
const { scan, take, last, map } = rxjs.operators;

// simulated lib function - not change this (the times are random)
function libA(callback) { setTimeout( _=>callback(2), 1000); } 
function libB(callback) { setTimeout( _=>callback(3), 3000); }
function libC(callback) { setTimeout( _=>callback(4), 2000); }
function libD(callback) { setTimeout( _=>callback(5), 1500); }


function loadString() {
  let result = new Subject();
  
  libA( n=> { result.next('A'.repeat(n)); });     
  libB( n=> { result.next('B'.repeat(n)); });     
  libC( n=> { result.next('C'.repeat(n)); });     
  libD( n=> { result.next('D'.repeat(n)); });        

  return result.pipe(
      take(4),
      scan((acc, value) => acc + value),
      last()
  );
}


// TEST
let s = loadString().subscribe(str=> {
   console.log(str); // wrong result: I expected "AADDDDDCCCCBBB" value 
})

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.5/rxjs.umd.min.js" integrity="sha256-85uCh8dPb35sH3WK435rtYUKALbcvEQFC65eg+raeuc=" crossorigin="anonymous"></script>