我正在尝试了解throttleTimevsdebounceTime以及何时使用哪个？

我有一个向上投票按钮，它向后端发出 API 请求（计算投票数）。用户可以多次提交按钮，但我想限制每秒可以按下按钮的次数。

我知道throttleTime和debounceTime运营商可以做到这一点，但哪一个我应该选择？

const upvoteClicks = fromEvent(this.el.nativeElement, 'click')
   .pipe(debounceTime(500))
   .subscribe(() => this.myService.postUpvote(this.postId));

我有一个组件，它onScrollEnd在呈现虚拟列表中的最后一个项目时触发一个事件。此事件将执行新的 API 请求以获取下一页并使用scan运算符将它们与先前的结果合并。

该组件还有一个触发onSearch事件的搜索字段。

如何在scan触发搜索事件时清除操作员之前的累积结果？或者我需要在这里重构我的逻辑吗？

const loading$ = new BehaviorSubject(false);
const offset$ = new BehaviorSubject(0);
const search$ = new BehaviorSubject(null);

const options$: Observable<any[]> = merge(offset$, search$).pipe(
  // 1. Start the loading indicator.
  tap(() => loading$.next(true)),
  // 2. Fetch new items based on the offset.
  switchMap(([offset, searchterm]) => userService.getUsers(offset, searchterm)),
  // 3. Stop the loading indicator.
  tap(() => loading$.next(false)),
  // 4. Complete the Observable when there is no 'next' link.
  takeWhile((response) => response.links.next), …

我有一个返回一定数量 id 的 API 调用。这些 id 中的每一个都用于进行新的 api 调用。这些 API 调用的结果需要组合成一个对象。

起初，我在第一个 api 调用的 .pipe(map) 运算符中使用了一个循环。在这个循环中，我进行了第二个 api 调用，并且在每个调用的 .pipe(map) 运算符中，我将在我的 angular 组件中编辑一个变量。

这不是很漂亮，我实际上想知道这是否是线程安全的。我知道 javascript 是单线程的，但是让多个异步进程混淆同一个全局变量似乎不太安全。

之后，我只是通过循环 apiCall1 返回的 Id 将第二个 api 调用返回的 observable 存储在一个数组中，并使用 forkJoin 相应地订阅和处理每个结果（参见示例）。

然而，这不是很漂亮，我想知道是否有我可以在我的管道中使用的运算符？

所以而不是（伪代码）：

  .pipe(
      map(ids=> {

        let observables = []
        for (const id of ids) {
         observables.push(this.service.getSomeStuff(id));
        }

        forkJoin(...observables).subscribe(dataArray) => {
          for (data of dataArray) {
            //Do something
          }
        });

      }),
      takeWhile(() => this.componentActive),
      catchError(error => {
        console.log(error);
        return throwError(error);
      })
    )
    .subscribe();

是否有一个运算符使它像这样：

  .pipe( …

在 rxjs 流中，我使用distinctUntilChangedlodashisEqual来过滤掉重复值。但是，它似乎没有按预期工作。取下面的代码片段

import { isEqual } from 'lodash-es';

let cachedValue: any;

function testFn(observableVal: Observable<any>) {
  return observableVal
    .pipe(
      distinctUntilChanged(isEqual),
      tap(val => {
        const equal = isEqual(cachedValue, val);
        console.log('"output":', equal, cachedValue, val);
        cachedValue = val;
      })
    )
}

在这个例子中，我希望函数const equal内部tap永远不会=== true. 我希望这distinctUntilChanged(isEqual)会过滤掉任何值，其中isEqual(cachedValue, val) === true-->const equal === false总是意味着。但是，控制台输出显示：

"output": false undefined [ContactList]
"output": true [ContactList] [ContactList]
"output": true [ContactList] [ContactList]
"output": true …

有以下（只是一个简单的例子）：

observable.pipe(map( s => s.anything ))
          .pipe(filter(t => t > 5))
          .pipe(map( t => t+5))
          .subscribe( XXX )

为什么我应该使用 1 个管道？

    observable.pipe(
                   map( s => s.anything ), filter(t => t > 5),map( t => t+5))
              .subscribe( XXX )

对我来说，第一种情况下的代码更漂亮、更易读。但不知道这是否会影响行为。

我正在迁移到rxjs 6.0.0-ucandoit-rc.6.在5.5.2我使用的版本ErrorObservable中创建了错误的observable.

我正在使用这里推荐的方式:https://github.com/ReactiveX/rxjs/blob/master/doc/pipeable-operators.md

因为throw是一个关键词,你可以在'rxjs/observable/throw'导入{_throw}之后使用_throw.

但是,不建议在rxjs6中导入此方法.

什么是正确的方法导入_throw或ErrorObservable？

举个例子：

 this.http.get('/getdata').pipe(delay(2000))

我希望此请求至少需要 2 秒才能完成，但不会超过请求完成所需的时间。

换句话说：

1. 如果请求需要1s完成，我希望 observable 在2s内完成。

2. 如果请求需要3s才能完成，我希望 observable 在3s 而不是5s 内完成。

是否还有其他管道delay()可以实现我不知道的这一点，或者是否有必要的话可以为此构建自定义管道？

用例是显示一个加载器，但是如果请求完成得太快，当加载器只是“闪烁”一瞬间时它看起来不太好

使用带有一个参数的 Pipe 函数与根本不使用 Pipe 相比，有什么区别吗？

我目前正在实施本文中的 takeUntil 取消订阅策略。在这个SO问题的“官方解决方案”中，takeUntil运算符是通过管道发送的。但是，在此页面上使用 takeUntil 时没有使用管道。

因此，我想知道使用带有单个 Rx 运算符的 Pipe 与根本不使用 Pipe 是否有任何区别（内存泄漏/性能等）。

private destroy$ = new Subject();
...
this.potatoService.getPotato()
   .pipe(
    takeUntil(this.destroy$)
   ).subscribe(...

相对于

this.potatoService.getPotato()
    .takeUntil(this.destroy$)
    .subscribe(...

抽象问题

有什么办法可以mergeMap按照外部 observable 的原始顺序使用 a 的结果，同时仍然允许内部 observable 并行运行？

更详细的解释

让我们看看两个合并映射运算符：

• mergeMap

  ...它需要一个映射回调，以及可以同时运行的内部 observable 的数量：

    of(1, 2, 3, 4, 5, 6).pipe(
        mergeMap(number => api.get('/double', { number }), 3)
    );
  

  Run Code Online (Sandbox Code Playgroud)

  在此处查看实际操作：https : //codepen.io/JosephSilber/pen/YzwVYNb?editors=1010

  这将分别触发1、2和 的3 个并行请求3。一旦其中一个请求完成，它就会触发另一个对4. 依此类推，始终保持 3 个并发请求，直到处理完所有值。

  但是，由于先前的请求可能在后续请求之前完成，因此产生的值可能会乱序。所以而不是：

    [2, 4, 6, 8, 10, 12]
  

  Run Code Online (Sandbox Code Playgroud)

  ...我们实际上可能会得到：

    [4, 2, 8, 10, 6, 12] // or any other permutation
  

  Run Code Online (Sandbox Code Playgroud)

• concatMap

  ...输入concatMap。此运算符确保所有 observables 都按原始顺序连接，以便：

    of(1, 2, 3, 4, 5, 6).pipe( …

  Run Code Online (Sandbox Code Playgroud)

我正在尝试实现此处描述的内容：https ://www.prestonlamb.com/blog/rxjs-cache-and-refresh-in-angular

换句话说，我想在给定时间（假设 1 分钟）内缓存一个可观察值。当在给定时间之后进行订阅时，应再次检索数据并再次缓存 1 分钟。

预期结果示例：

T 00:00: Request (1) => RETRIEVE data
T 00:10: Request (2) => data from cache
T 00:35: Request (3) => data from cache
T 00:50: Request (4) => data from cache
T 01:10: Request (5) => RETRIEVE data
T 01:15: Request (6) => data from cache
T 01:30: Request (7) => data from cache
T 02:30: Request (8) => RETRIEVE data

shareReplay 运算符可以正常工作，可以在给定时间内缓存数据，但在给定时间过后我无法重新启动它。

使用 shareRelay(1, 1000) 运算符的示例：

T 00:00: Request (1) …