Scala future 及其回调在相同的执行上下文中工作
Scala future and its callback works in the same execution context
我通过 Akka actor 调用 def activateReward 并且执行 OracleClient.rewardActivate(user) 有时非常慢(数据库不在我的职责范围内,属于另一家公司)。
当数据库速度慢时,线程池耗尽并且无法有效地分配更多线程给 运行 回调 future.onComplete 因为回调和 futures 在相同的执行上下文中工作。
请告知如何从为期货分配的线程异步执行回调中的代码OracleClient.rewardActivate(用户)
class RewardActivatorHelper {
private implicit val ec = new ExecutionContext {
val threadPool = Executors.newFixedThreadPool(1000)
def execute(runnable: Runnable) {threadPool.submit(runnable)}
def reportFailure(t: Throwable) {throw t}
}
case class FutureResult(spStart:Long, spFinish:Long)
def activateReward(msg:Msg, time:Long):Unit = {
msg.users.foreach {
user =>
val future:Future[FutureResult] = Future {
val (spStart, spFinish) = OracleClient.rewardActivate(user)
FutureResult(spStart, spFinish)
}
future.onComplete {
case Success(futureResult:FutureResult) =>
futureResult match {
case res:FutureResult => Logger.writeToLog(Logger.LogLevel.DEBUG,s"started:${res.spStart}finished:${res.spFinish}")
case _ => Logger.writeToLog(Logger.LogLevel.DEBUG, "some error")
}
case Failure(e:Throwable) => Logger.writeToLog(Logger.LogLevel.DEBUG, e.getMessage)
}
}
}
}
您可以通过以下方式为 onComplete
回调显式指定执行上下文,而不是隐式指定执行上下文:
import java.util.concurrent.Executors
import scala.concurrent.duration.Duration
object Example extends App {
import scala.concurrent._
private implicit val ec = new ExecutionContext {
val threadPool = Executors.newFixedThreadPool(1000)
def execute(runnable: Runnable) {threadPool.submit(runnable)}
def reportFailure(t: Throwable) {throw t}
}
val f = Future {
println("from future")
}
f.onComplete { _ =>
println("I'm done.")
}(scala.concurrent.ExecutionContext.Implicits.global)
Await.result(f, Duration.Inf)
}
这当然不会解决数据库跟不上的根本问题,但无论如何了解它可能是件好事。
澄清一下:我让 onComplete
回调由标准 global
执行上下文处理。您可能想创建一个单独的。
我通过 Akka actor 调用 def activateReward 并且执行 OracleClient.rewardActivate(user) 有时非常慢(数据库不在我的职责范围内,属于另一家公司)。
当数据库速度慢时,线程池耗尽并且无法有效地分配更多线程给 运行 回调 future.onComplete 因为回调和 futures 在相同的执行上下文中工作。
请告知如何从为期货分配的线程异步执行回调中的代码OracleClient.rewardActivate(用户)
class RewardActivatorHelper {
private implicit val ec = new ExecutionContext {
val threadPool = Executors.newFixedThreadPool(1000)
def execute(runnable: Runnable) {threadPool.submit(runnable)}
def reportFailure(t: Throwable) {throw t}
}
case class FutureResult(spStart:Long, spFinish:Long)
def activateReward(msg:Msg, time:Long):Unit = {
msg.users.foreach {
user =>
val future:Future[FutureResult] = Future {
val (spStart, spFinish) = OracleClient.rewardActivate(user)
FutureResult(spStart, spFinish)
}
future.onComplete {
case Success(futureResult:FutureResult) =>
futureResult match {
case res:FutureResult => Logger.writeToLog(Logger.LogLevel.DEBUG,s"started:${res.spStart}finished:${res.spFinish}")
case _ => Logger.writeToLog(Logger.LogLevel.DEBUG, "some error")
}
case Failure(e:Throwable) => Logger.writeToLog(Logger.LogLevel.DEBUG, e.getMessage)
}
}
}
}
您可以通过以下方式为 onComplete
回调显式指定执行上下文,而不是隐式指定执行上下文:
import java.util.concurrent.Executors
import scala.concurrent.duration.Duration
object Example extends App {
import scala.concurrent._
private implicit val ec = new ExecutionContext {
val threadPool = Executors.newFixedThreadPool(1000)
def execute(runnable: Runnable) {threadPool.submit(runnable)}
def reportFailure(t: Throwable) {throw t}
}
val f = Future {
println("from future")
}
f.onComplete { _ =>
println("I'm done.")
}(scala.concurrent.ExecutionContext.Implicits.global)
Await.result(f, Duration.Inf)
}
这当然不会解决数据库跟不上的根本问题,但无论如何了解它可能是件好事。
澄清一下:我让 onComplete
回调由标准 global
执行上下文处理。您可能想创建一个单独的。