时间戳

首页 > 解决方案 > 期货/并行代码的有用堆栈跟踪

scala - 期货/并行代码的有用堆栈跟踪

问题描述

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

ZIO 在提供有关问题所在和位置的背景方面做得很好。

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

标签： scalazio

解决方案

ZIO 堆栈跟踪确实令人惊叹。我们通过一个名为withDiagnostic. 这记录了调用该方法的文件和行号，并返回一个新的未来，该未来映射未来返回的任何异常以包含详细信息。

它使用scallacticPosition类型类来获取源位置。

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 comprehension 中的文件和行号包含在错误消息中java.lang.Exception: Where did I fail? (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 actor 输入的 ask 模式 akka。这是一个测试方法返回对询问的响应的示例（注意它包含一个implicit 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)))
  }
}

推荐阅读