正确使用scalaz Future进行异步执行

Question

我想我还不完全了解scalaz Futures的工作原理.我正在尝试将一个项目从scala期货移植到scalaz实现,但问题是scalaz Future的性能较低.最简单的示例是使用Spray在身份验证请求上加载配置文件.

功能本身:

def loadProfile[A: CollectionProvider: JsonFormat](id: String) = future {
  remote.findOne[A]("id" :> id) match {
    case Some(profile) ? \/-(profile)
    case None          ? -\/(ProfileNotFoundRejection(id))
  }
}

scalaz版本不同只是一个符号,我打电话Future.apply的scalaz.concurrent.现在加载一些html页面的Spray路由:

get {
  path("profile" / "id" ~ Segment) { id ?
    onSuccess(loadProfile[User](id)) {
      case \/-(profile) ? complete(html.page(profile))
      case -\/(pnfr)    ? reject(pnfr)
    }
  }
}

与此同时loadProfile,scalaz版本仅在方法调用中有所不同:

get {
  path("profile" / "id" ~ Segment) { id ?
    ctx => loadProfile[User](id).runAsync {
      case \/-(profile) ? ctx.complete(html.page(profile))
      case -\/(pnfr)    ? ctx.reject(pnfr)
    }
  }
}

但是scala Future版本的请求在(大约)143ms内完成,而scalaz版本在260ms内完成.所以我不太关心这个特殊的请求,但关于异步执行和服务的可伸缩性一般来说,正如我在scalaz中所理解的那样Future我必须手动将执行分叉到一个单独的线程,所以它按顺序执行？scalaz的未来用法是否有任何好的介绍/教程？

Answer 1

scala 和 scalaz future非常不同：

斯卡拉

import scala.concurrent._
import scala.concurrent.ExecutionContext.Implicits._

// creating two slow futures:
val f: Future[Unit] = Future { println("f " + Thread.currentThread().getName()); Thread.sleep(10000);  }
val g: Future[Unit] = Future { println("g " + Thread.currentThread().getName()); Thread.sleep(10000);  }

// and after moment asking for success
f onSuccess { case _ => println("f s1 " + Thread.currentThread().getName()) }
g onSuccess { case _ => println("g s1 " + Thread.currentThread().getName()) }
f onSuccess { case _ => println("f s2 " + Thread.currentThread().getName()) }
g onSuccess { case _ => println("g s2 " + Thread.currentThread().getName()) }

f创建后立即得到输出g

f ForkJoinPool-1-worker-5
g ForkJoinPool-1-worker-3

约 10 秒后停止输出

f s1 ForkJoinPool-1-worker-5
g s1 ForkJoinPool-1-worker-5
f s2 ForkJoinPool-1-worker-5
g s2 ForkJoinPool-1-worker-5

斯卡拉兹

import scalaz.concurrent._ // z!
import scala.concurrent.ExecutionContext.Implicits._

// creating two slow futures:
val f: Future[Unit] = Future { println("f " + Thread.currentThread().getName()); Thread.sleep(10000);  }
val g: Future[Unit] = Future { println("g " + Thread.currentThread().getName()); Thread.sleep(10000);  }

创建f和后g，什么也没有发生。我们有：

f: scalaz.concurrent.Future[Unit] = Async(<function1>)
g: scalaz.concurrent.Future[Unit] = Async(<function1>)

但运行后我们看到了差异：

f runAsync { _ => println("f s1 " + Thread.currentThread().getName()) }
g runAsync { _ => println("g s1 " + Thread.currentThread().getName()) }
f runAsync { _ => println("f s2 " + Thread.currentThread().getName()) }
g runAsync { _ => println("g s2 " + Thread.currentThread().getName()) }

我们得到结果：

f pool-4-thread-2
g pool-4-thread-1
f pool-4-thread-4
g pool-4-thread-3

f s2 pool-4-thread-4
g s2 pool-4-thread-3
g s1 pool-4-thread-1
f s1 pool-4-thread-2

有两点值得一提：

• 期货f并g再次执行。没有价值记忆。
• 回调runAsync与第一次计算在同一线程中执行。这是因为我们没有明确分叉。

很难说为什么它们在你的例子中表现不同。无论如何，大部分时间都应该花在上面remove.findOne。您希望使用scala.concurrent.blocking阻塞调用来帮助ExecutorService避免陷入线程饥饿（在这两种情况下）。