通过Scala Process执行Shell脚本且收集返回的结果

需求：通过Scala Process执行Shell脚本且收集返回的结果

package tools

import java.util.concurrent.LinkedBlockingQueue

import scala.sys.process.{Process,ProcessBuilder,ProcessLogger}

/** * * 收集Linux脚本命令返回的结果 */
object StreamProcessLogger {
  private val nonzeroException = true

  def run(processBuilder: ProcessBuilder): (Process,Stream[String]) = {
    val logger = new StreamProcessLogger
    val process = processBuilder.run(logger)
    waitForExitInAnotherThread(process,logger)
    (process,logger.stream)
  }

  private def waitForExitInAnotherThread(process: Process,logger: StreamProcessLogger) = {
    val thread = new Thread() {
      override def run() = {
        logger.setExitCode(process.exitValue())
      }
    }
    thread.start()
  }
}

private class StreamProcessLogger extends ProcessLogger {
  val queue = new LinkedBlockingQueue[Either[Int,String]]

  override def buffer[T](f: => T): T = f

  override def out(s: => String): Unit = queue.put(Right(s))

  override def err(s: => String): Unit = queue.put(Right(s))

  def stream = next()

  def setExitCode(exitCode: Int) = queue.put(Left(exitCode))

  private def next(): Stream[String] = queue.take match {
    case Left(0) => Stream.empty
    case Left(code) => if (StreamProcessLogger.nonzeroException) scala.sys.error("Nonzero exit code: " + code) else Stream.empty
    case Right(s) => Stream.cons(s,next())
  }
}

object StreamProcessLoggerHelp {
  def executeCommand(command: String,name: String) {
    if (name == "hive") {
      val (process,stream) = StreamProcessLogger.run(Process(Seq("hive","-e",command)))
      stream.foreach(info => println("hive==>" + info))
    } else {
      val (process,stream) = StreamProcessLogger.run(Process(command))
      stream.foreach(info => println("info==>" + info))
    }
  }

  def main(args: Array[String]): Unit = {

    //hive的执行
// StreamProcessLoggerHelp.executeCommand("drop table graph","hive")

    //执行脚本，比如shell脚本，Spark脚本等
// StreamProcessLoggerHelp.executeCommand("ls /Users/huiyu/ideaProject/graph","shell")
    StreamProcessLoggerHelp.executeCommand("spark-shell","shell")
  }
}

结果展示