RxJs 分组热可观察对象的笛卡尔积答案

【问题标题】：RxJs cartesian product of grouped hot observablesRxJs 分组热可观察对象的笛卡尔积
【发布时间】：2016-01-19 06:07:55
【问题描述】：

动机：

目标是编写一个允许提取关系/规范化数据的抓取库。

为此，有：

源流：发出cheerio 文档：通过抓取url 或通过限定父cheerio 文档的范围（例如cheerio 文档-> 多个cheerio 文档，每个li 对应于父文档）
表格流：发出表格行、订阅源流组合并从给定的 Cheerio 文档中提取数据

由于复杂数据需要多个表，每个源流可能被多次订阅，需要共享，因为它包含副作用（爬取）

示例：博客

在一个简单的博客上可能有帖子、作者和类别；底层数据结构是例如

Post: {id, author, title, text}
Author: {id, name}
Category: {id, name}
author_post: {author_id, post_id}
post_category: {post_id, category_id}

为了从抓取的 html 中重新创建数据结构，我们创建了三个源流：

post：馈送帖子网址，返回帖子的cheerio文档
post.author：帖子的子级，跟随帖子作者的超链接并发出作者的cheerio文档
post.category：帖子的子级，为帖子中列出的每个类别（例如在“.categories li”中）返回一个 Cheerio 文档

要重新创建 post_category 表，必须将每个帖子与属于该帖子的每个类别组合在一起（== carthesian product）。

我的实际问题更加人为，因为子流已经发出了自己的cheerio文档以及每个父母的文档，即{post：cheerio，作者：cheerio}。

合并流的问题只出现在例如兄弟姐妹。

我也无法通过在一个流中发出父级的所有子级来规避分组问题（例如 {post, author, category}），因为更复杂的数据结构需要由祖父母分组

（如果需要，我可以提供一个示例，但这已经足够长了）。

问题：

不可能使用 groupBy 和 zip 来组合热的 observables 组。

groupBy 发出的 GroupObservables 在创建后立即发出它们的值，而 zip 等待所有压缩的 observables 发出 GroupObservables 意味着丢失任何在 zip 函数运行之前发出的值。

问题：

如何在不丢失值的情况下对热门 observable 进行分组和压缩？不能依赖时间信息（例如，在下一个父级发射之前所有子级发射），因为子级可能被异步解析（爬网）。

更多信息：

我能想象的最好的是：

Child1 和 Child2 是 parent 的映射版本，C1C2 是 Child1 和 Child2 按 parent 分组并计算这些组内的笛卡尔积的结果。

Parent: -1------2-------3--------

Child1: --a------b--c------------

Child2: ---1-2----3---4----------

C1C2:   --a1-a2---b3-c3-b4-c4----

形成笛卡尔积本身没有问题，因为实现起来很简单（取自 RxJs 上的 issue #807，无法发布更多链接）

function xprod (o1, o2) {
  return o1.concatMap(x => o2, (x, y) => [x, y]);
};

问题不在于缺少任何值。

编辑： jsbin 展示了一个简单的情况：1 个父母，2 个孩子。

参考文献：

question on RxJs Github // 删除了额外的链接，因为我只能发布 2 个

【问题讨论】：

改进的描述，但是在我看来你仍然混合了太多的问题描述和实现细节（stream is possibly subscribed to multiple times ，父母，孩子，兄弟姐妹等）。我仍然不明白你想要的输出是什么。那是表流吗？那是你给出的底层数据结构吗？
无论如何，如果您有 html 文档（帖子）并且想要生成 id, author, title, text、id 您可能会生成自己，title 和 text 我想您不会有问题要解决。要获得author（只有一个通过post 对吗？），您需要获取其他一些html 文档并提取作者信息。那是对的吗？这会处理基础数据结构中的第一个表。 category 也一样，但帖子有几个类别。您仅在获取一些 html 文档后提取的每个类别。如果这就是全部，则不需要笛卡尔积。
如果您确实想做author 和category 的笛卡尔积，我的回答仍然应该有效。稍后再谈。请先确认我的理解是否正确。
博客的例子被简化了 - 我可以详细解释它，但我希望它足以说明：我需要组合分组流并在匹配组中形成 carthesian 产品
我真的不能说你的理解是否正确，我想我现在无法更好地表达它 - 我会尽量让它更清楚。在此之前，感谢您的帮助:)。

标签： javascript rxjs cartesian-product

【解决方案1】：

我在这里添加另一个答案，以说明我对您的问题的新理解。如果我理解错了，我稍后会删除它。

我的理解是：

有一个父 observable 生成值
child1 使用这些值来生成另一个 observable，它表示来自异步操作（$.get 或其他）的一系列值
child2 也一样
因此，父级的一个值会生成两个可观察值，而您需要这两个可观察值的笛卡尔积。

该代码应该这样做：

var parent = Rx.Observable.interval(500)
    .take(3)
    .publish();
var Obs3Val = function (x, delay) {
  return Rx.Observable.interval(delay)
    .map(function(y){return "P"+x+":"+"C" + y;})
    .take(3);
};
function emits(who){
  return function (x) {console.log(who + " emits " + x);};
}

var child1 = parent.map(x => Obs3Val(x, 200));
var child2 = parent.map(x => Obs3Val(x, 100));

Rx.Observable.zip(child1, child2)
             .concatMap(function(zipped){
  var o1 = zipped[0], o2 = zipped[1];
o1.subscribe(emits("child1"));
o2.subscribe(emits("child2"));
  return o1.concatMap(x => o2, (x, y) => [x, y]);
})
    .subscribe(function(v) {
                 console.log('cartesian product : ' +v);
               });

parent.subscribe();

parent.connect();

jsbin : https://jsbin.com/revurohoce/1/edit?js,console, https://jsbin.com/wegayacede/edit?js,console

日志：

"child2 emits P0:C0"
"child1 emits P0:C0"
"child2 emits P0:C1"
"cartesian product : P0:C0,P0:C0"
"child2 emits P0:C2"
"child1 emits P0:C1"
"cartesian product : P0:C0,P0:C1"
"cartesian product : P0:C0,P0:C2"
"child2 emits P1:C0"
"child1 emits P0:C2"
"cartesian product : P0:C1,P0:C0"
"child1 emits P1:C0"
"child2 emits P1:C1"
"cartesian product : P0:C1,P0:C1"
"child2 emits P1:C2"
"cartesian product : P0:C1,P0:C2"
"child1 emits P1:C1"
"cartesian product : P0:C2,P0:C0"
"cartesian product : P0:C2,P0:C1"
"child1 emits P1:C2"
"child2 emits P2:C0"
"cartesian product : P0:C2,P0:C2"
"child1 emits P2:C0"
"child2 emits P2:C1"
"child2 emits P2:C2"
"child1 emits P2:C1"
"cartesian product : P1:C0,P1:C0"
"cartesian product : P1:C0,P1:C1"
"child1 emits P2:C2"
"cartesian product : P1:C0,P1:C2"
"cartesian product : P1:C1,P1:C0"
"cartesian product : P1:C1,P1:C1"
"cartesian product : P1:C1,P1:C2"
"cartesian product : P1:C2,P1:C0"
"cartesian product : P1:C2,P1:C1"
"cartesian product : P1:C2,P1:C2"
"cartesian product : P2:C0,P2:C0"
"cartesian product : P2:C0,P2:C1"
"cartesian product : P2:C0,P2:C2"
"cartesian product : P2:C1,P2:C0"
"cartesian product : P2:C1,P2:C1"
"cartesian product : P2:C1,P2:C2"
"cartesian product : P2:C2,P2:C0"
"cartesian product : P2:C2,P2:C1"
"cartesian product : P2:C2,P2:C2"

【讨论】：

您好，感谢您的耐心等待。我已经更新了我的问题以提供更多信息 - 抱歉这么久。据我所知，您的代码之所以有效，是因为子项是自主的可观察对象，而不是父可观察对象的映射版本（请参阅我的示例 jsbin 链接）

【解决方案2】：

我想出的解决方案是返回 ReplaySubjects 的分组运算符的简单实现。虽然解决方案是针对我的要求的，但一般要点可能会有所帮助：

Observable.prototype.splitBy = function(keySelector) {
  const parentObservable = this;
  let group, lastKey;
  return Observable.create(observable => {
    return parentObservable.subscribe(
      value => {
        const currentKey = keySelector(value);
        if(currentKey === lastKey) {
          group.next(value);
        } else {
          if(group) group.complete();
          group = new ReplaySubject();
          observable.next(group);
          group.key = currentKey;
          group.next(value);
        }
        lastKey = currentKey;
      },
      error => observable.error(error),
      completed => {
        group.complete();
        observable.complete();
      });
  });
};

【讨论】：