scala - 是否可以在 akkastreams 中提取子流 key ？

Question

我似乎找不到关于此的任何文档，但我知道 AkkaStreams 在调用 时存储用于将流分组为子流的 key 。 groupBy 在内存中。是否可以从子流中提取这些 key ？假设我从我的主流创建了一堆子流，通过一个折叠来计算每个子流中的对象，然后将计数存储在一个类中。我可以将子流的 key 也传递给该类吗？或者有没有更好的方法来做到这一点？我需要计算每个子流的每个元素，但我还需要存储计数属于哪个组。

Answer 1

一个很好的例子显示在 stream-cookbook 中。 :

val counts: Source[(String, Int), NotUsed] = words
// split the words into separate streams first
  .groupBy(MaximumDistinctWords, identity)
  //transform each element to pair with number of words in it
  .map(_ -> 1)
  // add counting logic to the streams
  .reduce((l, r) => (l._1, l._2 + r._2))
  // get a stream of word counts
  .mergeSubstreams

然后如下:

val words = Source(List("Hello", "world", "let's", "say", "again", "Hello", "world"))
counts.runWith(Sink.foreach(println))

将打印:

(world,2)
(Hello,2)
(let's,1)
(again,1)
(say,1)

我想到的另一个例子是用余数计算数字。所以下面，例如:

Source(0 to 101)
  .groupBy(10, x => x % 4)
  .map(e => e % 4 -> 1)
  .reduce((l, r) => (l._1, l._2 + r._2))
  .mergeSubstreams.to(Sink.foreach(println)).run()

将打印:

(0,26)
(1,26)
(2,25)
(3,25)