gpt4 book ai didi

scala - Akka 流。一次控制 Akka Streams 中正在处理的项目数量

转载 作者:行者123 更新时间:2023-12-04 19:30:05 31 4
gpt4 key购买 nike

Akka 流显着减少了我的样板代码并包含许多有用的功能。但是,我需要能够限制处理项目的速度。问题是我正在提供一个附加到资源链接的 Hazelcast 队列,以便随着时间的推移(从单个在线站点)下载,但进入队列的链接数量可能会增长得非常大。理想情况下,一次运行的请求不超过 50-60 个。 Akka Streams 中是否有允许我限制一次处理的项目数量的功能?

另一个限制是在与某些网站交互时需要复杂的状态管理、代码处理和其他功能。 Akka Http 在这里无法提供帮助。我的网络代码完全是用 Jsoup 和 Apache Http 组件编写的,偶尔会调用基于 JavaFX 的服务器来呈现脚本。

我目前尝试使用文档中描述的缓冲区控制输入速率,如下所示:

val sourceGraph: Graph[SourceShape[(FlowConfig, Term)], NotUsed] = new HazelcastTermSource(conf.termQueue, conf)
val source = Source.fromGraph(sourceGraph)
val (killSwitch, last) = source
.buffer(conf.crawlStreamConf.maxCrawlConcurrency, OverflowStrategy.backpressure)
.viaMat(new DownloadFlow())(Keep.both)
.map(x => println(x))
.to(Sink.ignore).run()

最佳答案

您正在寻找的机制是 mapAsync (或 mapAsyncUnordered ,如果不需要保留顺序 - 就像你的例子一样)。
这些组合器采用 parallelism用于限制阶段可以运行的并行任务数量的参数。

它应该成为您 DownloadFlow 的一部分.
假设您的 DownloadFlow运行异步代码,你可以这样构造它:

def download(input: Input): Future[Output] = ???

val downloadFlow: Flow[Input, Output, NotUsed] = Flow[Input].mapAsyncUnordered(50)(download)

val (killSwitch, last) = source
.buffer(conf.crawlStreamConf.maxCrawlConcurrency, OverflowStrategy.backpressure)
.viaMat(downloadFlow)(Keep.both)
.map(x => println(x))
.to(Sink.ignore).run()

由于您的下载流程具有有意义的物化值(value),因此它可能会稍微复杂一些,但希望您能理解。

docs了解更多信息。

关于scala - Akka 流。一次控制 Akka Streams 中正在处理的项目数量,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/44447390/

31 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com