涉及文件转换的fs2流的性能提高

问题描述

我有类似的内容（这是https://api.flutter.dev/flutter/widgets/SafeArea-class.html的示例，其中有我的添加内容，并用注释标记了这些内容）：

import cats.effect.{Blocker,ExitCode,IO,IOApp,Resource}
import fs2.{io,text,Stream}
import java.nio.file.Paths

object Converter extends IOApp {

  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)
      .balanceAvailable // my addition
      .map ( worker => // my addition
        worker // my addition
          .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))
      ) // my addition
      .take(4).parJoinUnbounded // my addition
  }

  def run(args: List[String]): IO[ExitCode] =
    converter.compile.drain.as(ExitCode.Success)
}

如果fahrenheit.txt等于例如。 300mb的原始代码执行需要几分钟。看来我的代码并没有更快。如何改善其性能？光盘运行时，有很多未使用 CPU电源，光盘是SSD，所以我不知道为什么它这么慢。我不确定我是否正确使用balance。

解决方法

罪魁祸首是text.utf8Encode，每行不必要地发出一个块。当有很多短行时，例如在示例中（每行一个温度值，108199750行），这是巨大的开销。最近已解决（拉取请求：https://github.com/typelevel/fs2/pull/2096）。下面，我基于此PR提供了一个内联解决方案（只要有人使用没有此修复程序的版本，它就很有用）：

import cats.effect.{Blocker,ExitCode,IO,IOApp,Resource}
import fs2.{io,text,Stream,Pipe,Chunk}
import java.nio.file.Paths
import java.nio.charset.Charset

object Converter extends IOApp {

  val converter: Stream[IO,Unit] = Stream.resource(Blocker[IO]).flatMap  { blocker =>
    def fahrenheitToCelsius(f: Double): Double =
      (f - 32.0) * (5.0/9.0)

    def betterUtf8Encode[F[_]]: Pipe[F,String,Byte] =
      _.mapChunks(c => c.flatMap(s => Chunk.bytes(s.getBytes(Charset.forName("UTF-8")))))

    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) // didn't finish,could be an hour
      .through(betterUtf8Encode) // 2 minutes
      .through(io.file.writeAll(Paths.get("testdata/celsius.txt"),blocker))
  }

  def run(args: List[String]): IO[ExitCode] =
    converter.compile.drain.as(ExitCode.Success)
}

在这种情况下，这需要2分钟甚至一个小时甚至更长的时间。