【问题标题】:Passing implicit ExecutionContext to contained objects/called methods将隐式 ExecutionContext 传递给包含的对象/调用的方法
【发布时间】:2013-07-24 20:35:28
【问题描述】:

我正在使用 Scala 2.10 期货创建一个异步库。库的构造函数采用一系列实现特定特征的用户定义对象,然后库类上的方法将一些数据一个接一个地发送到用户定义对象中。我希望用户在设置主实例时为异步操作提供ExecutionContext,然后根据需要将该上下文传递给用户定义的对象。简化(伪?)代码:

case class Response(thing: String)

class LibraryObject(stack: Seq[Processor])(implicit context: ExecutionContext) {
  def entryPoint(data: String): Future[Response] = {
    val response = Future(Response(""))
    stack.foldLeft(response) { (resp, proc) => proc.process(data, resp) }
  }
}

trait Processor {
  def process(data: String, resp: Future[Response]): Future[Response]
}

可能会这样使用:

class ThingProcessor extends Processor {
  override def process(data: String, response: Future[Response]) = {
    response map { _.copy(thing = "THE THING") }
  }
}

class PassThroughProcessor extends Processor {
  override def process(request: Request, response: Future[Response]) = {
    response
  }
}

object TheApp extends App {
  import ExecutionContext.Implicits.global

  val stack = List(
    new ThingProcessor,
    new PassThroughProcessor
  )
  val libObj = new LibraryObject(stack)

  val futureResponse = libObj.entryPoint("http://some/url")

  // ...
}

我收到ThingProcessor 的编译错误:

找不到隐式 ExecutionContext,要么自己需要,要么导入 ExecutionContext.Implicits.global

我的问题是,我如何在不让用户(谁将编写类) 担心它——也就是说,我希望用户确实 不必 必须键入:

class MyFirstProcessor(implicit context: ExecutionContext)

override def process(...)(implicit context: ExecutionContext) = { ... }

【问题讨论】:

    标签: scala scala-2.10 executioncontext


    【解决方案1】:

    隐式作用域包括伴生对象和基类的类型参数。

    或者,library.submit(new library.Processor { def process() ... })。

    这可行,但不是我的第一个想法,这是更聪明的:

    import concurrent._
    import concurrent.duration._
    
    class Library(implicit xc: ExecutionContext = ExecutionContext.global) {
      trait Processor {
        implicit val myxc: ExecutionContext = xc
        def process(i: Future[Int]): Future[Int]
      }
    
      def submit(p: Processor) = p process future(7)
    }
    
    object Test extends App {
      val library = new Library
      val p = new library.Processor {
        def process(i: Future[Int]) = for (x <- i) yield 2 * x
      }
      val res = library submit p
      val z = Await result (res, 10.seconds)
      Console println z
    }
    

    更新:

    import concurrent._
    import concurrent.duration._
    import java.util.concurrent.Executors
    
    class Library()(implicit xc: ExecutionContext = ExecutionContext.global) {
      trait Processor {
        implicit val myxc: ExecutionContext = xc
        def process(i: Future[Int]): Future[Int]
      }
    
      def submit(p: Processor) = p process future(7)
    }
    
    object ctx {
      val xc = ExecutionContext fromExecutorService Executors.newSingleThreadExecutor
    }
    object library1 extends Library
    object library2 extends Library()(ctx.xc)
    object p extends library1.Processor {
      def process(i: Future[Int]) = for (x <- i) yield 2 * x
    }
    object q extends library2.Processor {
      def process(i: Future[Int]) = for (x <- i) yield 3 * x
    }
    
    object Test extends App {
      val res = library1 submit p
      //val oops = library2 submit p
      //val oops = library1 submit q
      val z = Await result (res, 10.seconds)
      Console println z
      Console println (Await result (library2 submit q, 10.seconds))
      ctx.xc.shutdownNow()
    }
    

    这并不难:

    class Library(implicit xc: ExecutionContext = ExecutionContext.global) {
    
      def submit(p: Processor): Future[Int] = p dueProcess future(7)
    }
    trait Processor {
      implicit var myxc: ExecutionContext = _
      def dueProcess(i: Future[Int])(implicit xc: ExecutionContext) = {
        myxc = xc
        process(i)
      }
      protected def process(i: Future[Int]): Future[Int]
    }
    
    object ctx {
      val xc = ExecutionContext fromExecutorService Executors.newSingleThreadExecutor
    }
    object Test extends App {
      def db() = Console println (new Throwable().getStackTrace mkString ("TRACE [\n  ", "\n  ", "\n]"))
      val library = new Library()(ctx.xc)
      val p = new Processor {
        protected def process(i: Future[Int]) = for (x <- i) yield { db(); 2 * x }
      }
      val res = library submit p
      val z = Await result (res, 10.seconds)
      Console println z
      ctx.xc.shutdownNow()
    }
    

    【讨论】:

    • 这显然不适用于单例对象...你能想出一个解决方法吗?
    • @pagoda_5b 你的意思是喜欢更新吗?如果我没有想清楚,请提前道歉。
    • 是的,看起来很漂亮,我喜欢它。
    • @pagoda_5b 如果不需要静态链接到库,则更新。如果一个 Lib 本质上是一个数据源,也许你想选择提交到哪里。对于静态链接,我在想 Processor[A <: library with singleton>
    猜你喜欢
    • 2014-03-03
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    相关资源
    最近更新 更多