说明

在使用 Akka 中的 FileIO 读取文件内容，并写入其他文件时，可能会报下面的错误：

[ERROR] [04/22/2023 19:26:01.435] [sys-akka.stream.default-blocking-io-dispatcher-19] [akka://sys/user/StreamSupervisor-0/flow-0-2-fileSink] Tearing down FileSink(D:\akka-quickstart\target\classes\logexample\test_out.txt) due to upstream error
akka.stream.scaladsl.Framing$FramingException: Stream finished but there was a truncated final frame in the buffer
    at akka.stream.scaladsl.Framing$DelimiterFramingStage$$anon$2.tryPull(Framing.scala:167)
    at akka.stream.scaladsl.Framing$DelimiterFramingStage$$anon$2.doParse(Framing.scala:185)
    at akka.stream.scaladsl.Framing$DelimiterFramingStage$$anon$2.onUpstreamFinish(Framing.scala:157)
    at akka.stream.impl.fusing.GraphInterpreter.processEvent(GraphInterpreter.scala:628)
    at akka.stream.impl.fusing.GraphInterpreter.execute(GraphInterpreter.scala:542)
    at akka.stream.impl.fusing.GraphInterpreterShell.runBatch(ActorGraphInterpreter.scala:471)
    at akka.stream.impl.fusing.GraphInterpreterShell.receive(ActorGraphInterpreter.scala:433)
    at akka.stream.impl.fusing.ActorGraphInterpreter.akka$stream$impl$fusing$ActorGraphInterpreter$$processEvent(ActorGraphInterpreter.scala:603)
    at akka.stream.impl.fusing.ActorGraphInterpreter$$anonfun$receive$1.applyOrElse(ActorGraphInterpreter.scala:618)
    at akka.actor.Actor$class.aroundReceive(Actor.scala:484)
    at akka.stream.impl.fusing.ActorGraphInterpreter.aroundReceive(ActorGraphInterpreter.scala:529)
    at akka.actor.ActorCell.receiveMessage(ActorCell.scala:526)
    at akka.actor.ActorCell.invoke(ActorCell.scala:495)
    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257)
    at akka.dispatch.Mailbox.run(Mailbox.scala:224)
    at akka.dispatch.Mailbox.exec(Mailbox.scala:234)
    at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

错误复现

  @Test
  public void test() throws InterruptedException {
    Source<ByteString, CompletionStage<IOResult>> source = FileIO.fromPath(
        Paths.get(System.getProperty("user.dir") + "/target/classes/logexample/test.txt"));
    // 读取文件内容
    Sink<ByteString, CompletionStage<IOResult>> sink = FileIO.toPath(
        Paths.get(System.getProperty("user.dir") + "/target/classes/logexample/test_out.txt"));
    // 指定写出文件位置
    Flow<ByteString, String, NotUsed> flowToString = Framing.delimiter(
        ByteString.fromString("\r\n"), 100).map(ByteString::utf8String);

    Flow<String, ByteString, NotUsed> flowToByte = Flow.of(String.class)
        .map(x -> ByteString.fromString(x.toString() + "\r\n"));

    ActorSystem system = ActorSystem.create("sys");
    ActorMaterializer materializer = ActorMaterializer.create(system);

    RunnableGraph<CompletionStage<IOResult>> graph = source.via(flowToString).via(flowToByte)
        .to(sink);
    CompletionStage<IOResult> run = graph.run(materializer);
    run.thenAccept(System.out::println);
    Thread.sleep(10000);
  }

其中，test.txt 是一个文本内容源文件，内容任意。test_out.txt 是一个输出内容的文本文件。

原因

delimiter 函数中的第三个参数 allowTruncation 是 akka.stream.javadsl.FramingTruncation 类型的枚举值，用于设置流是否允许出现不完整字节序列的情况。

如果将其设置为 FramingTruncation.ALLOW，则表示允许出现不完整帧的情况，并将返回仅包含完整帧的流。而如果将其设置为 FramingTruncation.DISALLOW，则在发现不完整帧时将会抛出异常并停止流。

通常情况下，如果您的数据源将始终生成完整的帧，那么您可以将 allowTruncation 设置为 FramingTruncation.DISALLOW，这样可以确保帧的完整性。但是，如果您的数据源可能会生成不完整的帧，则应将其设置为 FramingTruncation.ALLOW，以避免抛出异常并保留完整帧。

def delimiter(delimiter: akka.util.ByteString, maximumFrameLength: scala.Int, allowTruncation: akka.stream.javadsl.FramingTruncation): akka.stream.javadsl.Flow[akka.util.ByteString, akka.util.ByteString, akka.NotUsed] 

解决方法

给 Framing.delimiter 函数的第三个参数赋值 FramingTruncation.ALLOW 即可。

 Flow<ByteString, String, NotUsed> flowToString = Framing.delimiter(
        ByteString.fromString("\r\n"), 100, FramingTruncation.ALLOW).map(ByteString::utf8String);