Akka Stream中FileTailSource无法正常工作问题求助

你的代码存在两个关键问题导致无输出：

1. 无效的mapConcat()调用：groupBy(8,_._1).mapConcat()这里的mapConcat未传入任何参数，属于语法错误，直接阻断了流的元素传递。
2. reduce不适配无限流场景：reduce是终端操作，需要等待整个子流结束才会输出最终结果，但FileTailSource是持续监控文件的无限流，子流永远不会终止，因此reduce永远不会产生输出。

修正后的代码 #

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{FileIO, Flow, Framing, Sink, Source}
import akka.stream.alpakka.file.scaladsl.FileTailSource
import akka.util.ByteString
import java.nio.file.Paths
import scala.concurrent.duration._

object LogFileAnalyzer {
  def main(args: Array[String]): Unit = {
    implicit val system = ActorSystem("LogFileAnalyzer")
    implicit val materializer = ActorMaterializer()

    val logFile = Paths.get("./src/main/scala/log-generator.log")
    val outputFile = Paths.get("./f.txt")

    val source = FileTailSource(
      logFile,
      maxChunkSize = 4096,
      startingPosition = 0L,
      pollingInterval = 250.millis
    )
    // 按换行符分割日志行
    .via(Framing.delimiter(ByteString("\n"), maximumFrameLength = 4096, allowTruncation = false))
    // 解析日志行，提取网站信息并转为(网站, 访问计数增量)元组
    .map { line =>
      val fields = line.utf8String.split(" ")
      // 容错处理格式错误的日志行
      if (fields.length >=3) (fields(2), 1)
      else ("invalid_log", 1)
    }
    // 按网站分组，子流数量上限设为8
    .groupBy(8, _._1)
    // 使用scan实时累计访问次数，每次收到新元素就输出当前累计值
    .scan(("", 0)) { (acc, current) =>
      (current._1, acc._2 + current._2)
    }
    // 过滤初始的空累计值
    .filter(_._1.nonEmpty)
    // 将分组后的子流合并回主流
    .mergeSubstreams
    // 格式化为带换行的字节流，保证输出内容可读性
    .map { case (website, count) => ByteString(s"$website: $count\n") }

    val sink = FileIO.toPath(outputFile)
      // 设置为追加模式，避免覆盖已有统计结果
      .withAttributes(FileIO.WriteOptions.append)

    source.runWith(sink)
  }
}

关键修改说明 #

• 移除无效的mapConcat()：删除无意义的空调用，保证流的正常传递。
• 用scan替代reduce：scan支持在无限流中实时输出累计结果，符合日志监控的实时统计需求。
• 添加日志格式容错：处理字段不足的异常日志行，避免数组越界崩溃。
• 合并子流：groupBy后必须通过mergeSubstreams将子流合并回主流，否则元素无法传递到Sink。
• 文件追加写入：通过WriteOptions.append确保新统计结果追加到输出文件，而非覆盖原有内容。
• 添加换行符：让每条统计结果单独占一行，提升输出文件的可读性。

另外，若使用Akka 2.6及以上版本，ActorMaterializer已被废弃，推荐替换为：

import akka.stream.Materializer
implicit val materializer: Materializer = Materializer(system)

内容的提问来源于stack exchange，提问作者ach2ashes