scala - 对 future/并行代码有用的堆栈跟踪

Question

在线程中运行代码时，堆栈跟踪通常不会显示汇编/定义代码的逻辑流程， 但执行堆栈跟踪。这通常不是我调试和上下文所需的信息。

ZIO 在提供错误原因和错误位置的背景信息方面做得非常出色。

但是，如果您有一个不太可能移植到 zio 的遗留项目，我如何获得类似的异常/上下文信息，而不必重写我所有的 future 和 monad？

Answer 1

ZIO 堆栈跟踪确实令人惊叹。我们添加了一个与此相关的非常差的方法，名为 withDiagnostic。这会记录调用该方法的文件和行号，并返回一个新的 future，该 future 映射 future 返回的任何异常以包含详细信息。

它使用scalactic Position typeclass获取源位置。

object FutureUtils {

  implicit class RichFuture[A](future: Future[A]) {
    def withDiagnostic(implicit pos: Position, ec: ExecutionContext): Future[A] = {
      FutureUtils.withDiagnostic(future)
    }
  }


  def withDiagnostic[A](future: Future[A])(implicit pos: Position, ec: ExecutionContext): Future[A] = {
    future.recoverWith {
      case t: Throwable =>
        val diagnosticInfo = s"(${pos.fileName}:${pos.lineNumber})"
        if (t.getMessage.endsWith(diagnosticInfo)) Future.failed(t)
        else Future.failed(new Exception(t.getMessage + " " + diagnosticInfo, t))
    }
  }
}

这是在 Ammonite repl 中使用它的一个简短示例。请注意，在最后一个示例中，for 理解中的文件和行号包含在错误消息 java.lang.Exception:Where did I failed? 中。 (cmd10.sc:3) 在本例中，cmd10.sc:3 是 ammonite 生成的用于解析此行的文件。

Welcome to the Ammonite Repl 1.7.1
(Scala 2.12.10 Java 1.8.0_131)
If you like Ammonite, please support our development at www.patreon.com/lihaoyi
@ import org.scalactic.source._
import org.scalactic.source._

@ import scala.concurrent._
import scala.concurrent._

@ import scala.concurrent.duration._
import scala.concurrent.duration._

@ object FutureUtils {

      implicit class RichFuture[A](future: Future[A]) {
        def withDiagnostic(implicit pos: Position, ec: ExecutionContext): Future[A] = {
          FutureUtils.withDiagnostic(future)
        }
      }


      def withDiagnostic[A](future: Future[A])(implicit pos: Position, ec: ExecutionContext): Future[A] = {
        future.recoverWith {
          case t: Throwable =>
            val diagnosticInfo = s"(${pos.fileName}:${pos.lineNumber})"
            if (t.getMessage.endsWith(diagnosticInfo)) Future.failed(t)
            else Future.failed(new Exception(t.getMessage + " " + diagnosticInfo, t))
        }
      }
    }
defined object FutureUtils

@ import FutureUtils._
import FutureUtils._

@ implicit val ec: scala.concurrent.ExecutionContext = scala.concurrent.ExecutionContext.global
ec: ExecutionContext = scala.concurrent.impl.ExecutionContextImpl@434514d8

@ def futureSuccessful(i: Int): Future[Int] = Future.successful(i)
defined function futureSuccessful

@ def futureFailed: Future[Int] = Future.failed(new Exception("Where did I fail?"))
defined function futureFailed

@ val f1 = for {
      i <- futureSuccessful(3)
      j <- futureFailed
    } yield i + j
f1: Future[Int] = Future(<not completed>)

@ Await.result(f1, 1.second)
java.lang.Exception: Where did I fail?
  ammonite.$sess.cmd7$.futureFailed(cmd7.sc:1)
  ammonite.$sess.cmd8$.$anonfun$f1$1(cmd8.sc:3)
  ammonite.$sess.cmd8$.$anonfun$f1$1$adapted(cmd8.sc:2)
  scala.concurrent.Future.$anonfun$flatMap$1(Future.scala:307)
  scala.concurrent.impl.Promise.$anonfun$transformWith$1(Promise.scala:41)
  scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64)
  java.util.concurrent.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1402)
  java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
  java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
  java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
  java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)


@ val f2 = for {
      i <- futureSuccessful(3).withDiagnostic
      j <- futureFailed.withDiagnostic
    } yield i + j
f2: Future[Int] = Future(<not completed>)

@ Await.result(f2, 1.second)
java.lang.Exception: Where did I fail? (cmd10.sc:3)
  ammonite.$sess.cmd3$FutureUtils$$anonfun$withDiagnostic$1.applyOrElse(cmd3.sc:15)
  ammonite.$sess.cmd3$FutureUtils$$anonfun$withDiagnostic$1.applyOrElse(cmd3.sc:11)
  scala.concurrent.Future.$anonfun$recoverWith$1(Future.scala:417)
  scala.concurrent.impl.Promise.$anonfun$transformWith$1(Promise.scala:41)
  scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64)
  java.util.concurrent.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1402)
  java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
  java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
  java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
  java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
java.lang.Exception: Where did I fail?
  ammonite.$sess.cmd7$.futureFailed(cmd7.sc:1)
  ammonite.$sess.cmd10$.$anonfun$f2$1(cmd10.sc:3)
  ammonite.$sess.cmd10$.$anonfun$f2$1$adapted(cmd10.sc:2)
  scala.concurrent.Future.$anonfun$flatMap$1(Future.scala:307)
  scala.concurrent.impl.Promise.$anonfun$transformWith$1(Promise.scala:41)
  scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64)
  java.util.concurrent.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1402)
  java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
  java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
  java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
  java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)

我们非常依赖这个来使用 akka 类型的询问模式来测试 akka 参与者。下面是一个测试方法返回对询问的响应的示例(请注意，它包含一个隐式 pos 参数，现在调用 askEnqueueJob 的文件和行将出现在消息中.

def askEnqueueJob(referenceId: ReferenceId, tenant: Tenant)(implicit pos: Position): Future[EnqueueJobResponse] = {
  val job = newEnqueuedJob(referenceId = referenceId, tenant = tenant)

  withDiagnostic {
    tenantActor ? (sender => ProcessJobRequest(EnqueueJob(job, sender)))
  }
}