Akka流滑动窗口控制SourceQueue减少发射到接收器

Question

更新：我在test project 提出我的问题来解释我的意思

=====================================================================

我有来自数据库表的conkaune读取的Akka源，并且通过某个键组合然后减少它。然而似乎在我应用reduce函数后，数据永远不会发送到sink，它将继续减少，因为上游总是有数据到来。

我读了一些帖子，并尝试了分组和滑动，但它没有按照我的想法工作，它只将消息分组到更大的部分但从未使上游暂停并发出沉没。以下是Akka stream 2.5.2中的代码

Source reduce代码：

source = source
  .groupedWithin(100, FiniteDuration.apply(1, TimeUnit.SECONDS))
  .sliding(3, 1)
  .mapConcat(i -> i)
  .mapConcat(i -> i)
  .groupBy(2000000, i -> i.getEntityName())
  .map(i -> new Pair<>(i.getEntityName(), i))
  .reduce((l, r) ->{ l.second().setAction(r.second().getAction() + l.second().getAction()); return l;})
  .map(i -> i.second())
  .mergeSubstreams();

沉没并运行：

Sink<Object, CompletionStage<Done>> sink = 
        Sink.foreach(i -> System.out.println(i))
final RunnableGraph<SourceQueueWithComplete<Object>> run = source.toMat(sink, Keep.left());
run.run(materIalizer);

我也尝试过.takeWhile（谓词）;我使用timer来切换谓词值true和false，但似乎它只会将第一个开关设为false，当我切换回true时它不会重启上游。

请提前帮助我！

=================================================

更新

有关元素类型的信息

添加我想要的：我有类调用SystemCodeTracking包含2个属性(id, entityName)

我将有对象列表：(1, "table1"), (2, "table2"), (3, "table3"),(4, "table1"),(5, "table3")

我想groupBy entityName然后加上id，因此，我希望看到的结果如下

("table1" 1+4),("table3", 3+5),("table2", 2)

我现在正在做的代码如下

source
.groupBy(2000000, systemCodeTracking -> systemCodeTracking.getEntityName)
.map(systemCodeTracking -> new Pair<String, Integer>(systemCodeTracking.getEntityName, SystemCodeTracking.getId()))
.scan(....)

我现在的问题是如何构建扫描初始状态呢？

scan(new Pair<>("", 0), (first, second) -> first.setId(first.getId() + second.getId()))

Answer 1

所以，如果我理解一切，你想要的是：

• 首先，按ID分组
• 然后按时间窗口分组，并在这个时间窗口内，将所有systemCodeTracking.getId()相加

对于第一部分，你需要groupBy。对于第二部分groupedWithin。但是，它们的工作方式不同：第一个将为您提供子流，而第二个将为您提供一个列表流。

因此，我们必须以不同的方式处理它们。

首先，让我们为您的列表编写一个reducer：

private SystemCodeTracking reduceList(List<SystemCodeTracking> list) throws Exception {
    if (list.isEmpty()) {
        throw new Exception();
    } else {
        SystemCodeTracking building = list.get(0);
        building.setId(0L);
        list.forEach(next -> building.setId(building.getId() + next.getId()));
        return building;
    }
}

因此，对于列表中的每个元素，我们递增building.id以获得遍历整个列表时所需的值。

现在你只需要做

Source<SystemCodeTracking, SourceQueueWithComplete<SystemCodeTracking>> loggedSource = source
    .groupBy(20000, SystemCodeTracking::getEntityName) // group by name
    .groupedWithin(100, FiniteDuration.create(10, TimeUnit.SECONDS)   // for a given name, group by time window (or by packs of 100)
    .filterNot(List::isEmpty)                          // remove empty elements from the flow (if no element has passed in the last second, to avoid error in reducer)
    .map(this::reduceList)                             // reduce each list to sum the ids
    .log("====== doing reduceing ")                    // log each passing element using akka logger, rather than `System.out.println`
    .mergeSubstreams()                                 // merge back all elements with different names