scala - 使用 akka 流对数据库进行 Tcp 请求

Question

我正在尝试使用 akka-streams 的 Tcp 客户端向数据库发送查询，但我不明白我错过了什么。

所以我有两种类型 Query 和 Response 可以完美地转换为 akka 的 ByteString 或从 ByteString 转换。因此，我正在使用 val conn = Tcp().outgoingConnection("localhost", 28015) 创建客户端连接，这为我提供了一个 Flow[ByteString, ByteString, Future[OutgoingConnection]] ，到目前为止一切顺利。所以我假设源是我对查询的请求，我找不到用查询源提供此流程的最佳方法，而是像 Source(Future.successful(query)) 一样构造它，并将其连接到流 source.via(flow)，这为我提供了另一个 Source[Response, Unit]。在这里我不明白如何获得 Future[Response]，尝试了几个组合器，但它给了我 Materialized 值，我不完全理解它是如何关联的流中的值/类型。

Answer 1

首先:它是什么类型的数据库，为什么要直接通过 TCP 访问它？您确定这会按照您缩进的方式工作吗？您是否能够处理传入响应的框架？

关于从 Source[Response, Unit] 中获取 Future[Response] 的问题，就像使用 Sink 运行 Source 一样简单.head，即像这样:val res: Future[Result] = source.runWith(Sink.head)(你需要一个implicit val mat = ActorMaterializer()当然在范围内)。

我强烈建议您花一些时间在 Akka Streams documentation 上在深入使用 Streams 之前。