我刚刚开始使用 fs2 流进行冒险。我想要实现的是读取一个文件(一个大文件,这就是我使用 fs2 的原因),转换它并将结果写入两个不同的文件(基于某些谓词)。一些代码(来自 https://github.com/typelevel/fs2 ),以及我的评论:
val converter: Stream[IO, Unit] = Stream.resource(Blocker[IO]).flatMap { blocker =>
def fahrenheitToCelsius(f: Double): Double =
(f - 32.0) * (5.0/9.0)
io.file.readAll[IO](Paths.get("testdata/fahrenheit.txt"), blocker, 4096)
.through(text.utf8Decode)
.through(text.lines)
.filter(s => !s.trim.isEmpty && !s.startsWith("//"))
.map(line => fahrenheitToCelsius(line.toDouble).toString)
.intersperse("\n")
.through(text.utf8Encode)
.through(io.file.writeAll(Paths.get("testdata/celsius.txt"), blocker))
/* instead of the last line I want something like this:
.through(<write temperatures higher than 10 to one file, the rest to the other one>)
*/
}
这样做的最有效方法是什么?显而易见的解决方案是让两个流具有不同的过滤器,但效率低下(将有两次通过)。
最佳答案
不幸的是,据我所知,没有简单的方法可以拆分 fs2 流成两个。
您可以做的是通过将值推送到两个队列之一来拆分您的流(第一个是低于 10 的值,第二个是大于或等于 10 的值)。如果我们使用 NoneTerminatedQueue
然后队列不会终止,直到我们放置 None
进入他们。那么我们就可以使用 dequeue
创建单独的流,直到队列未关闭。
下面的示例解决方案。我将写入文件和读取分成单独的方法:
import java.nio.file.Paths
import cats.effect.{Blocker, ExitCode, IO, IOApp}
import fs2.concurrent.{NoneTerminatedQueue, Queue}
import fs2.{Stream, io, text}
object FahrenheitToCelsius extends IOApp {
def fahrenheitToCelsius(f: Double): Double =
(f - 32.0) * (5.0 / 9.0)
//I split reading into separate method
def read(blocker: Blocker, over: NoneTerminatedQueue[IO, Double], under: NoneTerminatedQueue[IO, Double]) = io.file.readAll[IO](Paths.get("testdata/fahrenheit.txt"), blocker, 4096)
.through(text.utf8Decode)
.through(text.lines)
.filter(s => !s.trim.isEmpty && !s.startsWith("//"))
.map(line => fahrenheitToCelsius(line.toDouble))
.evalMap { value =>
if (value > 10) { //here we put values to one of queues
over.enqueue1(Some(value)) //until we put some queues are not close
} else {
under.enqueue1(Some(value))
}
}
.onFinalize(
over.enqueue1(None) *> under.enqueue1(None) //by putting None we terminate queues
)
//function write takes as argument source queue and target file
def write(s: Stream[IO, Double], blocker: Blocker, fileName: String): Stream[IO, Unit] = {
s.map(_.toString)
.intersperse("\n")
.through(text.utf8Encode)
.through(io.file.writeAll(Paths.get(fileName), blocker))
}
val converter: Stream[IO, Unit] = for {
over <- Stream.eval(Queue.noneTerminated[IO, Double]) //here we create 2 queues
under <- Stream.eval(Queue.noneTerminated[IO, Double])
blocker <- Stream.resource(Blocker[IO])
_ <- write(over.dequeue, blocker, "testdata/celsius-over.txt") //we run reading and writing to both
.concurrently(write(under.dequeue, blocker, "testdata/celsius-under.txt")) //files concurrently
.concurrently(read(blocker, over, under)) //stream runs until queue over is not terminated
} yield ()
override def run(args: List[String]): IO[ExitCode] =
converter
.compile
.drain
.as(ExitCode.Success)
}
关于scala - 将 fs2 流输出拆分为两个文件,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/64188023/