时间戳

首页 > 解决方案 > withLatestFrom 意外行为

rxjs - withLatestFrom 意外行为

问题描述

我对操作员 withLatestFrom 有意外行为。

输出

map a
a
map a <== why a is mapped again ?
map b
b


const { Subject, operators } = window.rxjs
const { map, withLatestFrom } = operators


const createA = new Subject()
const createB = new Subject()

const a = createA.pipe(
  map(() => console.log('map a'))
)

const b = createB.pipe(
  withLatestFrom(a),
  map(() => console.log('map b'))
)

a.subscribe(() => { console.log('a') })
b.subscribe(() => { console.log('b') })

createA.next()
createB.next()
<script src="https://unpkg.com/@reactivex/rxjs@6.6.3/dist/global/rxjs.umd.js"></script>

标签： rxjs

解决方案

这里的问题不withLatestFrom()在于订阅的工作方式，而在于订阅的工作方式。Observables 是惰性的，在您订阅之前不会运行。每个新订阅都会再次运行observable。

const stream$ = from([1,2,3]);
stream$.subscribe(console.log) // output: 1 2 3
stream$.subscribe(console.log) // output: 1 2 3

在这种情况下，`from([1,2,3)] 运行了两次。如果我改变我的流，我所做的任何事情都会发生在每个订阅者身上。

const stream$ = from([1,2,3]).pipe(
  tap(_ => console.log("hi"))
);
stream$.subscribe(console.log) // output: hi 1 hi 2 hi 3
stream$.subscribe(console.log) // output: hi 1 hi 2 hi 3

难题的最后一块是：内部withLatestFrom()订阅你给它的流。就像显式.subscribe() 运行observable 一样，withLatestFrom()订阅后也是如此。

您可以使用shareReplay缓存最新的值并重播它们，而不是再次运行 observable。这是管理多播流的一种方法：

const createA = new Subject()
const createB = new Subject()

const a = createA.pipe(
  tap(() => console.log('tap a')),
  shareReplay(1)
)

const b = createB.pipe(
  withLatestFrom(a),
  tap(() => console.log('tap b'))
)

a.subscribe(() => { console.log('a') })
b.subscribe(() => { console.log('b') })

createA.next()
createB.next()

现在a.subscribe()和withLatestFrom(a)都得到一个缓冲值，只有在执行时才会运行createA.next()。

顺便说一句，将值映射为空是一种不好的习惯。考虑以下代码：

from([1,2,3]).pipe(
  map(val => console.log(val))
).subscribe(val => console.log(val));

这将输出

1
undefined
2
undefined
3
undefined

因为您实际上将每个值都映射为空。tap另一方面，不会改变可观察的源，因此它是一个更好的工具，用于调试和/或不改变流的副作用

from([1,2,3]).pipe(
  tap(val => console.log(val))
).subscribe(val => console.log(val));

这将输出

1
1
2
2
3
3

推荐阅读