rxjs 嵌套的可观察对象、窗口和扫描的意外行为

Question

我想在数据进入时显示分析的部分结果。为每个新值重新计算效率非常低（与“扫描”一样）。但是，在这种情况下，我可以对数据块进行分析并结合这些结果。所以我一直在使用“窗口”来分解数据，然后“扫描”来组合每个窗口计算的结果。结果本身就是一个可观察对象，因此将其作为嵌套的可观察对象发出是很自然的。此外，在使用 observables 时，该过程的下一步非常有效。

但是，我无法按预期工作。 （我确实做了一个尴尬的步骤，将内部可观察对象转换为数组，然后再转换为可观察对象。）似乎我对“窗口”和/或“扫描”有些不理解。

这里有两个示例，它们在生成嵌套 observable 的方式上有所不同。我原以为以下两个示例会给出相同的结果，但事实并非如此。

首先，我直接创建嵌套的 observable。第二，我用窗口操作创建它。然后，在这两种情况下，我都会对嵌套的 observable 应用相同的扫描。

这符合我的预期：

rxjs.from([rxjs.from([1, 2]), rxjs.from([3, 4])])
    .pipe(
        ops.scan((acc, curr) => rxjs.merge(acc, curr), rxjs.from([]))
    ).subscribe(win => win.subscribe(
        x => console.log(JSON.stringify(x)), e => console.log("error"), () => console.log("|")),
        e => console.log("outer error"), () => console.log("outer |"))

对于每个发出的 observable，我看到了前一个值的累积，然后是新的值。 1 2 | 1 2 3 4 |

我预计下一个会产生相同的结果，但事实并非如此：

rxjs.from([1, 2, 3, 4])
    .pipe(
        ops.windowCount(2),
        ops.scan((acc, curr) => rxjs.merge(acc, curr), rxjs.from([]))
    ).subscribe(win => win.subscribe(x => console.log(JSON.stringify(x)), e => console.log("error"), () => console.log("|")),
        e => console.log("outer error"), () => console.log("outer|"))

它似乎有效地忽略了扫描操作并发出原始窗口， 1 2 | 3 4 |

我错过了什么？对此的传统解决方案是什么样的？谢谢！

Answer 1

windowCount 在内部使用Subject。因此它创建并返回一个主题，然后将1 和2 发送给它以用于第一个窗口。在第一次scan 迭代中，您在1 和2 发送和接收这些值之前订阅此主题。对于以后的迭代，您在 1 和 2 已经发出之后订阅，因此您不会再次收到这些值。

有点像：

const { Subject, merge, from } = rxjs

const window1 = new Subject()
const scanResult1 = merge(from([]), window1)
scanResult1.subscribe(console.log)
window1.next(1)
window1.next(2)

console.log('|')

const window2 = new Subject()
const scanResult2 = merge(scanResult1, window2)
scanResult2.subscribe(console.log)
window2.next(3)
window2.next(4)

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

使用 bufferCount

您只需将windowCount 替换为bufferCount 即可将数组发送到scan 而不是主题。 scan 中的代码可以保持与 merge 中的代码相同，也可以处理数组，但如果你想保证值以它们进入的相同顺序发出，你应该使用 concat 而不是 merge。

rxjs.from([1, 2, 3, 4])
  .pipe(
    ops.bufferCount(2),
    ops.scan((acc, curr) => rxjs.concat(acc, curr), rxjs.from([]))
  ).subscribe(
    win => win.subscribe(
      x => console.log(JSON.stringify(x)), 
      e => console.log("error"), () => console.log("|")
    ),
    e => console.log("outer error"),
    () => console.log("outer|")
  )

使用 windowCount

您可以将shareReplay 添加到您的窗口，以便将其值重播给未来的订阅者。由于 windowCount 如果源的计数可被 windowSize 整除，则在最后发出一个空窗口，因此只有当当前窗口不为空时，您才必须映射到合并的 observable。否则你会得到两次最终结果。

from([1, 2, 3, 4]).pipe(
  windowCount(2),
  scan((acc, curr) => {
    const shared = curr.pipe(shareReplay())
    return shared.pipe(
      isEmpty(),
      switchMap(empty => empty ? EMPTY : merge(acc, shared))
    )
  }, from([]))
)

或者

from([1, 2, 3, 4]).pipe(
  windowCount(2),
  map(w => w.pipe(shareReplay())),
  concatMap(w => w.pipe(isEmpty(), filter(e => !e), mapTo(w))),
  scan((acc, curr) => merge(acc, curr), from([]))
)