代码之家 › 专栏 › 技术社区 › zella

如何实现带跳过和条件停止的流

akka-stream

zella · 技术社区 · 7 年前

我试图实现批处理。我的算法:

1)首先,我需要数据库的请求项,初始 skip = 0 是的。如果没有项目,则完全停止处理。

  case class Item(i: Int)

  def getItems(skip: Int): Future[Seq[Item]] = {
    Future((skip until (skip + (if (skip < 756) 100 else 0))).map(Item))
  }

2)然后每件事都要做重活( parallelism = 4 )

  def heavyJob(item: Item): Future[String] = Future {
    Thread.sleep(1000)
    item.i.toString + " done"
  }

3)所有项目处理完成后,转到 skip += 100

我在尝试什么:

val dbSource: Source[List[Item], _] = Source.fromFuture(getItems(0).map(_.toList))

val flattened: Source[Item, _] = dbSource.mapConcat(identity)

val procced: Source[String, _] = flattened.mapAsync(4)(item => heavyJob(item))

procced.runWith(Sink.onComplete(t => println("Complete: " + t.isSuccess)))

但我不知道如何实现分页

1 回复 | 直到 7 年前

Ramón J Romero y Vigil 7 年前

这个 skip 递增可以用 Iterator 作为价值的基本来源:

val skipIncrement = 100

val skipIterator : () => Iterator[Int] = 
  () => Iterator from (0, skipIncrement)

这个迭代器可以用来驱动一个akka Source 获取项目并将继续处理,直到查询返回空 Seq 以下内容:

val databaseStillHasValues : Seq[Item] => Boolean = 
  (dbValues) => !dbValues.isEmpty

val itemSource : Source[Item, _] = 
  Source.fromIterator(skipIterator)
        .mapAsync(1)(getItems)
        .takeWhile(databaseStillHasValues)
        .mapConcat(identity)

这个 heavyJob 可用于 Flow 以下内容:

val heavyParallelism = 4

val heavyFlow : Flow[Item, String, _] = 
  Flow[Item].mapAsync(heavyParallelism)(heavyJob)

最后,源和流可以附加到 Sink 以下内容:

val printSink = Sink[String].foreach(t => println(s"Complete: ${t.isSuccess}"))

itemSource.via(heavyFlow)
          .runWith(printSink)

推荐文章

Falk Schuetzenmeister · Akka:如何在一个图形阶段提取值并在下一个图形阶段使用

8 年前

Teyu11 · Akka HTTP多文件上传到S3

8 年前

xDs · Alpakka MongoDB-在MongoSource中指定类型

8 年前

Roma Gafiyatullin · 阿克卡河。组合物化值。需要绘制未来地图。要使用哪个ExecutionContext/Dispatcher?

8 年前

apflieger · 如何在播放测试中从基于InputStream的结果中获取内容

8 年前

Jatin · 在project reactor或akka streams中,sink和subscriber在概念上有什么区别?

8 年前

stan · Scala RestartSink未来

8 年前

Alexander Arendar · 不完全了解来源。延迟方法或akka流文档中存在错误

8 年前

user3294786 · 使用Akka Http转换流畅的流数据并发送分块响应

8 年前

vgkowski · Akka流从外部世界获取无限流的当前值

8 年前