继续使用 mergeMap 处理 RxJs 中的错误

Question

我正在使用 RxJs 管道和 mergeMap 运算符进行一些并行 HTTP 获取。

当第一个请求失败时（假设 /urlnotexists 抛出 404 错误），它会停止所有其他请求。

我希望它继续查询所有剩余的 url，而不为此失败的请求调用所有剩余的 mergeMap。

我尝试使用 RxJs 中的 throwError 和 catchError 但没有成功。

索引.js

const { from } = require('rxjs');
const { mergeMap, scan } = require('rxjs/operators');

const request = {
  get: url => {
    return new Promise((resolve, reject) => {
      setTimeout(() => {
        if (url === '/urlnotexists') { return reject(new Error(url)); }
        return resolve(url);
      }, 1000);
    });
  }
};

(async function() {
  await from([
    '/urlexists',
    '/urlnotexists',
    '/urlexists2',
    '/urlexists3',
  ])
    .pipe(
      mergeMap(async url => {
        try {
          console.log('mergeMap 1:', url);
          const val = await request.get(url);
          return val;
        } catch(err) {
          console.log('err:', err.message);
          // a throw here prevent all remaining request.get() to be tried
        }
      }),
      mergeMap(async val => {
        // should not pass here if previous request.get() failed 
        console.log('mergeMap 2:', val);
        return val;
      }),
      scan((acc, val) => {
        // should not pass here if previous request.get() failed 
        acc.push(val);
        return acc;
      }, []),
    )
    .toPromise()
    .then(merged => {
      // should have merged /urlexists, /urlexists2 and /urlexists3
      // even if /urlnotexists failed
      console.log('merged:', merged);
    })
    .catch(err => {
      console.log('catched err:', err);
    });
})();

const { from } = require('rxjs');
const { mergeMap, scan } = require('rxjs/operators');

const request = {
  get: url => {
    return new Promise((resolve, reject) => {
      setTimeout(() => {
        if (url === '/urlnotexists') { return reject(new Error(url)); }
        return resolve(url);
      }, 1000);
    });
  }
};

(async function() {
  await from([
    '/urlexists',
    '/urlnotexists',
    '/urlexists2',
    '/urlexists3',
  ])
    .pipe(
      mergeMap(async url => {
        try {
          console.log('mergeMap 1:', url);
          const val = await request.get(url);
          return val;
        } catch(err) {
          console.log('err:', err.message);
          // a throw here prevent all remaining request.get() to be tried
        }
      }),
      mergeMap(async val => {
        // should not pass here if previous request.get() failed 
        console.log('mergeMap 2:', val);
        return val;
      }),
      scan((acc, val) => {
        // should not pass here if previous request.get() failed 
        acc.push(val);
        return acc;
      }, []),
    )
    .toPromise()
    .then(merged => {
      // should have merged /urlexists, /urlexists2 and /urlexists3
      // even if /urlnotexists failed
      console.log('merged:', merged);
    })
    .catch(err => {
      console.log('catched err:', err);
    });
})();

我希望发出并发 GET 请求，并在最后减少一个对象中各自的值。

但如果发生一些错误，我希望他们不要中断我的管道，而是记录它们。

有什么建议吗？

Answer 1

catchError如果您想使用 RxJS，您应该在使用 并发执行所有请求之前向单个请求添加错误处理和任何其他任务forkJoin。

const { of, from, forkJoin } = rxjs;
const { catchError, tap } = rxjs.operators;

// your promise factory, unchanged (just shorter)
const request = {
  get: url => {
    return new Promise((resolve, reject) => setTimeout(
      () => url === '/urlnotexists' ? reject(new Error(url)) : resolve(url), 1000
    ));
  }
};

// a single rxjs request with error handling
const fetch$ = url => {
  console.log('before:', url);
  return from(request.get(url)).pipe(
    // add any additional operator that should be executed for each request here
    tap(val => console.log('after:', val)),
    catchError(error => {
      console.log('err:', error.message);
      return of(undefined);
    })
  );
};

// concurrently executed rxjs requests
forkJoin(["/urlexists", "/urlnotexists", "/urlexists2", "/urlexists3"].map(fetch$))
  .subscribe(merged => console.log("merged:", merged));

<script src="https://unpkg.com/@reactivex/rxjs@6.5.3/dist/global/rxjs.umd.js"></script>