Scala应用流式读取外部进程写入的动态增长文件需求问询

刚好遇到过类似的场景，我给你整理了一个能覆盖你提到的两种边缘情况的实用方案，代码和思路都给你理得明明白白：

核心思路 #

咱们要启动一个独立线程，主要做这几件事：

1. 等待目标文件被外部进程创建出来
2. 持续读取文件新增的内容到目标OutputStream
3. 直到外部进程完全终止，且文件不再增长（确认没有新内容写入）

关键边缘情况处理 #

1. 线程启动时文件尚未存在 #

用循环+短暂休眠的方式等待文件创建，既避免CPU空转，还能加超时机制防止无限等待。

2. 读取速度快于写入速度（读到文件末尾但文件还在增长） #

记录每次读取后的文件长度，循环检查文件长度变化：

• 如果进程还在运行，只要文件长度增加，就继续读取新增内容
• 如果进程已终止，再等待一小段时间确认文件长度稳定，就结束读取

完整代码实现 #

import java.io.{File, FileInputStream, OutputStream}
import java.util.concurrent.TimeUnit
import scala.util.control.Breaks._

def readFileUntilProcessComplete(
    targetFile: File,
    outputStream: OutputStream,
    externalProcess: Process,
    waitForFileIntervalMs: Long = 100,
    waitForStableFileMs: Long = 500,
    maxWaitForFileMs: Long = 30000
): Unit = {
  // 先等待文件出现，带超时机制
  val fileWaitStart = System.currentTimeMillis()
  while (!targetFile.exists()) {
    if (System.currentTimeMillis() - fileWaitStart > maxWaitForFileMs) {
      throw new IllegalStateException(s"等待目标文件创建超时，路径：${targetFile.getAbsolutePath}")
    }
    TimeUnit.MILLISECONDS.sleep(waitForFileIntervalMs)
  }

  // 打开文件输入流，用try-finally确保资源关闭
  var fileInputStream: FileInputStream = null
  try {
    fileInputStream = new FileInputStream(targetFile)
    var lastReadPosition = 0L

    // 循环读取直到进程结束且文件稳定
    while (externalProcess.isAlive() || targetFile.length() > lastReadPosition) {
      val currentFileLength = targetFile.length()
      
      // 读取从上次位置到当前末尾的内容
      if (currentFileLength > lastReadPosition) {
        val buffer = new Array[Byte](8192)
        fileInputStream.getChannel.position(lastReadPosition)
        var bytesRead = fileInputStream.read(buffer)
        while (bytesRead != -1) {
          outputStream.write(buffer, 0, bytesRead)
          outputStream.flush() // 确保内容及时写入输出流
          bytesRead = fileInputStream.read(buffer)
        }
        lastReadPosition = currentFileLength
      }

      // 如果进程已终止，检查文件是否稳定
      if (!externalProcess.isAlive()) {
        TimeUnit.MILLISECONDS.sleep(waitForStableFileMs)
        // 再次检查文件长度，如果没变化就退出循环
        if (targetFile.length() == lastReadPosition) {
          break
        }
      } else {
        // 进程还在运行，短暂休眠后再检查
        TimeUnit.MILLISECONDS.sleep(waitForFileIntervalMs)
      }
    }
  } finally {
    if (fileInputStream != null) {
      fileInputStream.close()
    }
  }
}

使用示例 #

// 1. 启动外部进程（替换成你的实际命令和参数）
val externalProcess = new ProcessBuilder("your-external-command", "arg1", "arg2").start()

// 2. 准备输出流（这里以文件输出为例，也可以换成网络流等）
val outputStream = new java.io.FileOutputStream("your-output-file.txt")

// 3. 启动读取线程
val readerThread = new Thread(() => {
  try {
    val targetFile = new File("external-process-output.txt")
    readFileUntilProcessComplete(targetFile, outputStream, externalProcess)
  } catch {
    case e: Exception => e.printStackTrace()
  }
})
readerThread.start()

// 4. 等待进程和线程完成，确保资源正确释放
externalProcess.waitFor()
readerThread.join()
outputStream.close()

可传递的参数说明 #

你可以向这个线程传递这些参数来适配不同场景：

• targetFile: 外部进程写入的目标文件对象
• outputStream: 要复制内容到的输出流（比如网络流、文件流等）
• externalProcess: 外部进程的引用，用来判断进程是否终止
• waitForFileIntervalMs: 等待文件出现或检查文件变化的间隔时间（默认100ms）
• waitForStableFileMs: 进程终止后等待文件稳定的时间（默认500ms）
• maxWaitForFileMs: 等待文件创建的最长超时时间（默认30秒）

注意事项 #

1. 资源关闭：一定要用try-finally或者Scala 2.13+的Using来管理输入输出流，避免资源泄漏
2. 异常处理：线程里的异常要捕获处理，避免线程默默终止而你毫无察觉
3. 性能调整：根据外部进程的写入速度，调整休眠时间间隔，平衡CPU占用和读取延迟

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