如何为不支持流式传输的数据库提供响应式流 API？就像让我们说例如dynamodb。进行 get 调用时，dynamodb 将返回所有结果。因此，即使我将 get 调用包装在 Source 中，如何处理来自下游阶段的背压？另外我如何在数据库中实现写调用？我的水槽会是什么样子？对此的任何指示都会有所帮助。
【问题讨论】：
标签： scala akka amazon-dynamodb akka-stream reactive-streams
一种选择是使用ActorPublisher 实现您的数据库Source -
见：http://doc.akka.io/docs/akka/2.4.11/scala/stream/stream-integrations.html#ActorPublisher
只需混合此 trait 并实现命令接口，即可为您提供一个响应式流兼容的数据发布者，该发布者可以处理下游背压。如果订阅者下游拉取更多数据，您的发布者将收到Request 消息，并且如果它需要主动向下游推送更多数据，它将可以访问当前感知的需求。然后，您可以通过创建 Source 来将此发布者插入您的 Akka Streams 管道：
Source.actorPublisher[Data](MyPublisher.props).runWith(MySink)
为了处理底层数据库本身不是反应式的事实，您需要在ActorPublisher 中实现一些缓冲和轮询逻辑。
【讨论】：