为什么在 groupBy 运算符之后不调用 subscribe？

Question

为什么不调用第二个示例的订阅方法？在这两个示例中，管道内的所有日志都按预期工作。工作示例：（但使用硬编码数据和创建者）：of([  {tableName: 'table1', firstName: 'name1', lastName: 'lastName1'},  {tableName: 'table1', firstName: 'name2', lastName: 'lastName2'},  {tableName: 'table2', firstName: 'name3', lastName: 'lastName3'},  {tableName: 'table2', firstName: 'name4', lastName: 'lastName4'}]).pipe(  tap(data => console.log('amount of records', data.length)),  mergeMap((searchResult) => searchResult),  groupBy(b => b.tableName),  mergeMap(group => {    console.log('groupKey', group.key);    return group.pipe(toArray());  })).subscribe(val => console.log(val));问题代码：（使用直接来自 ngrx-store 的数据和与上面的代码工作示例相同的数据结构）this.store.pipe(  select(selectSearchResult),  tap(data => console.log('amount of records', data.length)),  mergeMap((searchResult) => searchResult),  groupBy(b => b.tableName),  mergeMap(group => {    console.log('groupKey', group.key);    return group.pipe(toArray());  })).subscribe(val => console.log(val)); // <- this is not called!?我希望为每个组调用 subscribe 方法，就像上面使用硬编码数据的工作示例一样。但具有讽刺意味的是，订阅方法没有被调用？！

Answer 1

我认为问题出在这里：

Store 是一个长期存在的Observable。这意味着：它不会自行完成。

toArray收集其源流的所有通知，直到该流完成，然后才发送带有项目数组的单个通知。

否则，它如何知道数组何时准备好发送给下一个操作员？总有另一个通知可以使阵列更大。

请注意，GroupedObservable所生产groupBy将无法完成，直到它的来源是完整的。因为源store不会完成，toArray所以永远不会收到complete消息。

解决方案是：重构您的代码，以便使用from. from以一个数组为参数，为每个数组项发送一个通知，然后发送一个完整的通知，表示toArray它可以发送一个通知。这应该有效：

mergeMap((searchResult) => 

  from(searchResult).pipe(

    groupBy(b => b.tableName),

    mergeMap(group => {

      console.log('groupKey', group.key);

      return group.pipe(toArray());

  })

))

来源：RxJS 代码查看 GroupedObservable 何时完成。

Answer 2

它认为您将 groupBy 混淆为处理数组而不是流的函数，RxJs 函数适用于流，而不是流发出的值。

groupBy 不会对流发出的数组进行分组，而是对流进行分组。每次 observable 发出一个对象时，它都会被添加到组中。如果要对存储中的数组进行分组，则需要使用数组的 groupBy 函数。

这是我编写的 groupBy 函数的示例，该函数采用数组https://stackblitz.com/edit/typescript-ezydzv

map 是 RxJs 函数，它操作由流发出的对象，在 map 中，您可以将 groupBy 应用于数组。

this.store.pipe(

  select(selectSearchResult),

  tap(data => console.log('amount of records', data.length)),

  map(data => groupBy(data, { keys: ['tableName'] }))

).subscribe(val => console.log(val));