我正在尝试编写一段代码来消耗代码流(公司的股票交易所代码),并从每个代码的REST API中获取公司信息。
我想异步获取多家公司的信息。
我想以连续的方式将结果保存到文件中,因为整个数据集可能不适合内存。
根据akka流和资源的文档,我能够在这个主题上谷歌,我提出了以下代码(为简洁起见,省略了一些部分):
implicit val actorSystem: ActorSystem = ActorSystem("stock-fetcher-system")
implicit val materializer: ActorMaterializer = ActorMaterializer(None, Some("StockFetcher"))(actorSystem)
implicit val context = system.dispatcher
import CompanyJsonMarshaller._
val parallelism = 10
val connectionPool = Http().cachedHostConnectionPoolHttps[String](s"api.iextrading.com")
val listOfSymbols = symbols.toList
val outputPath = "out.txt"
Source(listOfSymbols)
.mapAsync(parallelism) {
stockSymbol => Future(HttpRequest(uri = s"https://api.iextrading.com/1.0/stock/${stockSymbol.symbol}/company"), stockSymbol.symbol)
}
.via(connectionPool)
.map {
case (Success(response), _) => Unmarshal(response.entity).to[Company]
case (Failure(ex), symbol) => println(s"Unable to fetch char data for $symbol") "x"
}
.runWith(FileIO.toPath(new File(outputPath).toPath, Set(StandardOpenOption.APPEND)))
.onComplete { _ =>
bufferedSource.close
actorSystem.terminate()
}
这是有问题的一行:
runWith(FileIO.toPath(new File(outputPath).toPath, Set(StandardOpenOption.APPEND)))
哪个不编译,编译器给我这个神秘的错误:
Type mismatch, expected Graph[SinkShape[Any, NotInferedMat2], actual Sink[ByeString, Future[IOResult]]
如果我将接收器更改为Sink.ignore或println(_)它可以工作。
我会感谢一些更详细的解释。
正如编译器指示的那样,类型不匹配。在致.map
...
.map {
case (Success(response), _) =>
Unmarshal(response.entity).to[Company]
case (Failure(ex), symbol) =>
println(s"Unable to fetch char data for $symbol")
"x"
}
...你要么返回Company
实例或String
,所以编译器推断最接近的超类型(或“最小上限”)为Any
。 Sink
期望ByteString
类型的输入元素,而不是Any
。
一种方法是将响应发送到文件接收器,而无需解组响应:
Source(listOfSymbols)
.mapAsync(parallelism) {
...
}
.via(connectionPool)
.map(_.entity.dataBytes) // entity.dataBytes is a Source[ByteString, _]
.flatMapConcat(identity)
.runWith(FileIO.toPath(...))