scala.concurrent.Future[T]。Publisher特质，您可以将其混合到某些Actor子类中。这可以定义一些状态来跟踪订阅者、处理Subscribe和Unsubscribe消息，并提供某种方便的方法来广播消息: /**
* Sends a copy of the supplied event object to every subscriber of
* the event object class and superclasses.
*/
protected[this] def publish[T](event: T) {
for (subscriber <- subscribersFor(event)) subscriber ! event
}
这些只是一些基于最近项目中类似做法的想法。如果您需要更具体的指导，请详细说明您的用例。此外，如果您确实有兴趣在Scala中探索Actor，Akka用户列表是一个很好的资源，可以解答一般性问题。
// this is from the spray-client docs
val pipeline: HttpRequest => Future[HttpResponse] = sendReceive
val response: Future[HttpResponse] = pipeline(Get("http://spray.io/"))
import scala.concurrent._
import spray.can.client.HttpClient
import spray.client.HttpConduit
import spray.client.HttpConduit._
import spray.http.{HttpRequest, HttpResponse, FormData}
import spray.httpx.unmarshalling.Unmarshaller
import spray.io.IOExtension
type Pipeline = (HttpRequest) => Future[HttpResponse]
// this is basically spray-client boilerplate
def createPipeline(system: ActorSystem, host: String, port: Int): Pipeline = {
val httpClient = system.actorOf(Props(new HttpClient(IOExtension(system).ioBridge())))
val conduit = system.actorOf(props = Props(new HttpConduit(httpClient, host, port)))
sendReceive(conduit)
}
private var pipeline: Pipeline = _
// unmarshalls to a specific type, e.g. a case class representing a datamodel
private def unmarshallingPipeline[T](implicit ec:ExecutionContext, um:Unmarshaller[T]) = (pipeline ~> unmarshal[T])
// for requests that don't return any content. If you get a successful Future it worked; if there's an error you'll get a failed future from the errorFilter below.
private def unitPipeline(implicit ec:ExecutionContext) = (pipeline ~> { _:HttpResponse => () })
// similar to unitPipeline, but where you care about the specific response code.
private def statusPipeline(implicit ec:ExecutionContext) = (pipeline -> {r:HttpResponse => r.status})
// if you want standard error handling create a filter like this
// RemoteServerError and RemoteClientError are custom exception classes
// not shown here.
val errorFilter = { response:HttpResponse =>
if(response.status.isSuccess) response
else if(response.status.value >= 500) throw RemoteServerError(response)
else throw RemoteClientError(response)
}
pipeline = (createPipeline(system, "yourHost", 8080) ~> errorFilter)
def ping()(implicit ec:ExecutionContext, um:Unmarshaller[String]): Future[String] =
unmarshallingPipeline(Get("/ping"))
def form(formData: Map[String, String])(implicit ec:ExecutionContext, um:Unmarshaller[DataModel]): Future[DataModel] =
unmarshallingPipeline(Post("/form"), FormData(formData))
import scala.util.{Failure, Success}
API.ping() foreach(println) // will print out "pong" when response comes back
API.form(Map("a" -> "b") onComplete {
case Success(dataModel) => println("Form accepted. Server returned DataModel: " + dataModel)
case Failure(e) => println("Oh noes, the form didn't go through! " + e)
}
这个例子看起来像是“可观察对象”模式的一个很好的实现。该模式源自于 .NET响应式扩展，但也适用于 Java 和 Scala。该库由Netflix提供，具有非常好的质量。
这个模式有一个良好的理论基础---从范畴论意义上来说，它是迭代器的对偶。但更重要的是，它有很多实用的想法。特别是在处理时间方面非常出色，例如你可以限制希望获取的事件速率。
通过可观察对象，您可以在非常高的层次上处理事件。在 .NET 中，它看起来很像 SQL 查询。您可以注册某些事件（“FROM”），筛选它们（“WHERE”），最后进行处理（“SELECT”）。在 Scala 中，您可以使用标准的单调 API （map、filter、flatMap）和当然还有“for 表达式”。
一个示例可能如下：
stackoverflowQuestions.filter(_.tag == "Scala").map(_.subject).throttleLast(1 second).subscribe(println _)
观察者模式可以解决许多与事件相关的问题：
您的API应为每个事件源提供一个观察者。对于过程调用，您提供一个函数，将该函数调用映射到观察者。此函数将调用远程过程并通过观察者提供结果。
将以下依赖项添加到build.sbt中：
libraryDependencies += "com.netflix.rxjava" % "rxjava-scala" % "0.15.0"
private val callbackFunc : (rx.lang.scala.Observer[String]) => rx.lang.scala.Subscription = { o =>
val listener = {
case Value(s) => o.onNext(s)
case Error(e) => o.onError(o)
}
remote.subscribe(listener)
// Return an interface to cancel the subscription
new Subscription {
val unsubscribed = new AtomicBoolean(false)
def isUnsubscribed: Boolean = unsubscribed.get()
val asJavaSubscription: rx.Subscription = new rx.Subscription {
def unsubscribe() {
remote.unsubscribe(listener)
unsubscribed.set(true)
}
}
}
在coursera上，Martin Odersky等人提供了一个非常好的课程，涵盖了Observables和其他反应技术。