我花了一些时间进行实验，发现你可以在 Akka 中使用Observables。事实上，由于Observable 可以被认为是Future 的多变量扩展，因此您可以遵循与组合Actor 和Futures 相同的准则。实际上，官方文档和教科书（例如 Akka Concurrency，Wyatt 2013）都支持/鼓励在 Akka 中使用 Future，但有很多警告。
首先是积极的：
-
Observables 和 Futures 一样是不可变的，因此理论上它们在消息中传递应该是安全的。
-
Observable 允许您指定执行上下文，非常类似于Future。这是使用Observable.observeOn(scheduler) 完成的。您可以通过将 Akka 调度程序（例如 system.dispatcher 或 context.dispatcher）传递给 rx.lang.scala.ExecutorScheduler 构造函数，从 Akka 的 exec 上下文创建调度程序。这应该确保它们是同步的。
- 与上述相关，正在开发中的 rx-scala 增强功能 (https://github.com/Netflix/RxJava/issues/815#issuecomment-38793433)，这将允许隐式指定 observable 的调度程序。
- Futures 非常适合使用
ask 模式的 Akka。类似的模式可以用于 Observables（见本文底部）。这也解决了向远程 observables 发送消息的问题。
现在需要注意的是：
- 它们与未来有着相同的问题。例如，请参见页面底部：http://doc.akka.io/docs/akka/2.3.2/general/jmm.html。还有 Wyatt 2013 中关于期货的章节。
- 正如@mavilein 的回答，这意味着
Observable.subscribe() 不应使用Actor 的封闭范围来访问其内部状态。例如，您不应该在订阅中调用sender。相反，将其存储到一个 val 中，然后访问这个 val，如下例所示。
- Akka 使用的调度器的分辨率与 Rx 不同。它的默认分辨率为 100 毫秒（Wyatt 2013）。如果有人遇到过这可能导致的问题，请在下面发表评论！
最后，我为 Observables 实现了 ask 模式的等价物。它使用toObservable 或?? 异步返回一个Observable，由一个临时演员和一个PublishSubject 在幕后支持。请注意，源发送的消息是使用materialize() 的rx.lang.scala.Notification 类型，因此它们满足可观察合约中的complete 和error 状态。否则我们无法将这些状态发送给接收器。但是，没有什么可以阻止您发送任意类型的消息；这些将简单地调用onNext()。如果在某个时间间隔内没有收到消息，则 observable 有一个超时并以超时异常停止。
它是这样使用的：
import akka.pattern.RX
implicit val timeout = akka.util.Timeout(10 seconds)
case object Req
val system = ActorSystem("test")
val source = system.actorOf(Props[Source],"thesource")
class Source() extends Actor {
def receive : Receive = {
case Req =>
val s = sender()
Observable.interval(1 second).take(5).materialize.subscribe{s ! _}
}
}
val obs = source ?? Req
obs.observeOn(rx.lang.scala.schedulers.ExecutorScheduler(system.dispatcher)).subscribe((l : Any) => println ("onnext : " + l.toString),
(error : Throwable) => { error.printStackTrace ; system.shutdown() },
() => { println("completed, shutting system down"); system.shutdown() })
并产生这个输出：
onnext : 0
onnext : 1
onnext : 2
onnext : 3
onnext : 4
completed, shutting system down
来源如下。它是 AskSupport.scala 的修改版本。
package akka.pattern
/*
* File : RxSupport.scala
* This package is a modified version of 'AskSupport' to provide methods to
* support RX Observables.
*/
import rx.lang.scala.{Observable,Subject,Notification}
import java.util.concurrent.TimeoutException
import akka.util.Timeout
import akka.actor._
import scala.concurrent.ExecutionContext
import akka.util.Unsafe
import scala.annotation.tailrec
import akka.dispatch.sysmsg._
class RxTimeoutException(message: String, cause: Throwable) extends TimeoutException(message) {
def this(message: String) = this(message, null: Throwable)
override def getCause(): Throwable = cause
}
trait RxSupport {
implicit def toRx(actorRef : ActorRef) : RxActorRef = new RxActorRef(actorRef)
def toObservable(actorRef : ActorRef, message : Any)(implicit timeout : Timeout) : Observable[Any] = actorRef ?? message
implicit def toRx(actorSelection : ActorSelection) : RxActorSelection = new RxActorSelection(actorSelection)
def toObservable(actorSelection : ActorSelection, message : Any)(implicit timeout : Timeout): Observable[Any] = actorSelection ?? message
}
final class RxActorRef(val actorRef : ActorRef) extends AnyVal {
def toObservable(message : Any)(implicit timeout : Timeout) : Observable[Any] = actorRef match {
case ref : InternalActorRef if ref.isTerminated =>
actorRef ! message
Observable.error(new RxTimeoutException(s"Recepient[$actorRef] has alrady been terminated."))
case ref : InternalActorRef =>
if (timeout.duration.length <= 0)
Observable.error(new IllegalArgumentException(s"Timeout length must not be negative, message not sent to [$actorRef]"))
else {
val a = RxSubjectActorRef(ref.provider, timeout, targetName = actorRef.toString)
actorRef.tell(message, a)
a.result.doOnCompleted{a.stop}.timeout(timeout.duration)
}
}
def ??(message :Any)(implicit timeout : Timeout) : Observable[Any] = toObservable(message)(timeout)
}
final class RxActorSelection(val actorSel : ActorSelection) extends AnyVal {
def toObservable(message : Any)(implicit timeout : Timeout) : Observable[Any] = actorSel.anchor match {
case ref : InternalActorRef =>
if (timeout.duration.length <= 0)
Observable.error(new IllegalArgumentException(s"Timeout length must not be negative, message not sent to [$actorSel]"))
else {
val a = RxSubjectActorRef(ref.provider, timeout, targetName = actorSel.toString)
actorSel.tell(message, a)
a.result.doOnCompleted{a.stop}.timeout(timeout.duration)
}
case _ => Observable.error(new IllegalArgumentException(s"Unsupported recipient ActorRef type, question not sent to [$actorSel]"))
}
def ??(message :Any)(implicit timeout : Timeout) : Observable[Any] = toObservable(message)(timeout)
}
private[akka] final class RxSubjectActorRef private (val provider : ActorRefProvider, val result: Subject[Any]) extends MinimalActorRef {
import RxSubjectActorRef._
import AbstractRxActorRef.stateOffset
import AbstractRxActorRef.watchedByOffset
/**
* As an optimization for the common (local) case we only register this RxSubjectActorRef
* with the provider when the `path` member is actually queried, which happens during
* serialization (but also during a simple call to `toString`, `equals` or `hashCode`!).
*
* Defined states:
* null => started, path not yet created
* Registering => currently creating temp path and registering it
* path: ActorPath => path is available and was registered
* StoppedWithPath(path) => stopped, path available
* Stopped => stopped, path not yet created
*/
@volatile
private[this] var _stateDoNotCallMeDirectly: AnyRef = _
@volatile
private[this] var _watchedByDoNotCallMeDirectly: Set[ActorRef] = ActorCell.emptyActorRefSet
@inline
private[this] def watchedBy: Set[ActorRef] = Unsafe.instance.getObjectVolatile(this, watchedByOffset).asInstanceOf[Set[ActorRef]]
@inline
private[this] def updateWatchedBy(oldWatchedBy: Set[ActorRef], newWatchedBy: Set[ActorRef]): Boolean =
Unsafe.instance.compareAndSwapObject(this, watchedByOffset, oldWatchedBy, newWatchedBy)
@tailrec // Returns false if the subject is already completed
private[this] final def addWatcher(watcher: ActorRef): Boolean = watchedBy match {
case null => false
case other => updateWatchedBy(other, other + watcher) || addWatcher(watcher)
}
@tailrec
private[this] final def remWatcher(watcher: ActorRef): Unit = watchedBy match {
case null => ()
case other => if (!updateWatchedBy(other, other - watcher)) remWatcher(watcher)
}
@tailrec
private[this] final def clearWatchers(): Set[ActorRef] = watchedBy match {
case null => ActorCell.emptyActorRefSet
case other => if (!updateWatchedBy(other, null)) clearWatchers() else other
}
@inline
private[this] def state: AnyRef = Unsafe.instance.getObjectVolatile(this, stateOffset)
@inline
private[this] def updateState(oldState: AnyRef, newState: AnyRef): Boolean =
Unsafe.instance.compareAndSwapObject(this, stateOffset, oldState, newState)
@inline
private[this] def setState(newState: AnyRef): Unit = Unsafe.instance.putObjectVolatile(this, stateOffset, newState)
override def getParent: InternalActorRef = provider.tempContainer
def internalCallingThreadExecutionContext: ExecutionContext =
provider.guardian.underlying.systemImpl.internalCallingThreadExecutionContext
/**
* Contract of this method:
* Must always return the same ActorPath, which must have
* been registered if we haven't been stopped yet.
*/
@tailrec
def path: ActorPath = state match {
case null =>
if (updateState(null, Registering)) {
var p: ActorPath = null
try {
p = provider.tempPath()
provider.registerTempActor(this, p)
p
} finally { setState(p) }
} else path
case p: ActorPath => p
case StoppedWithPath(p) => p
case Stopped =>
// even if we are already stopped we still need to produce a proper path
updateState(Stopped, StoppedWithPath(provider.tempPath()))
path
case Registering => path // spin until registration is completed
}
override def !(message: Any)(implicit sender: ActorRef = Actor.noSender): Unit = state match {
case Stopped | _: StoppedWithPath => provider.deadLetters ! message
case _ =>
if (message == null) throw new InvalidMessageException("Message is null")
else
message match {
case n : Notification[Any] => n.accept(result)
case other => result.onNext(other)
}
}
override def sendSystemMessage(message: SystemMessage): Unit = message match {
case _: Terminate => stop()
case DeathWatchNotification(a, ec, at) => this.!(Terminated(a)(existenceConfirmed = ec, addressTerminated = at))
case Watch(watchee, watcher) =>
if (watchee == this && watcher != this) {
if (!addWatcher(watcher))
// NEVER SEND THE SAME SYSTEM MESSAGE OBJECT TO TWO ACTORS
watcher.sendSystemMessage(DeathWatchNotification(watchee, existenceConfirmed = true, addressTerminated = false))
} else System.err.println("BUG: illegal Watch(%s,%s) for %s".format(watchee, watcher, this))
case Unwatch(watchee, watcher) =>
if (watchee == this && watcher != this) remWatcher(watcher)
else System.err.println("BUG: illegal Unwatch(%s,%s) for %s".format(watchee, watcher, this))
case _ =>
}
@deprecated("Use context.watch(actor) and receive Terminated(actor)", "2.2") override def isTerminated: Boolean = state match {
case Stopped | _: StoppedWithPath => true
case _ => false
}
@tailrec
override def stop(): Unit = {
def ensureCompleted(): Unit = {
result.onError(new ActorKilledException("Stopped"))
val watchers = clearWatchers()
if (!watchers.isEmpty) {
watchers foreach { watcher =>
// NEVER SEND THE SAME SYSTEM MESSAGE OBJECT TO TWO ACTORS
watcher.asInstanceOf[InternalActorRef]
.sendSystemMessage(DeathWatchNotification(watcher, existenceConfirmed = true, addressTerminated = false))
}
}
}
state match {
case null => // if path was never queried nobody can possibly be watching us, so we don't have to publish termination either
if (updateState(null, Stopped)) ensureCompleted() else stop()
case p: ActorPath =>
if (updateState(p, StoppedWithPath(p))) { try ensureCompleted() finally provider.unregisterTempActor(p) } else stop()
case Stopped | _: StoppedWithPath => // already stopped
case Registering => stop() // spin until registration is completed before stopping
}
}
}
private[akka] object RxSubjectActorRef {
private case object Registering
private case object Stopped
private final case class StoppedWithPath(path : ActorPath)
def apply(provider: ActorRefProvider, timeout: Timeout, targetName: String): RxSubjectActorRef = {
val result = Subject[Any]()
new RxSubjectActorRef(provider, result)
/*timeout logic moved to RxActorRef/Sel*/
}
}
/*
* This doesn't work, need to create as a Java class for some reason ...
final object AbstractRxActorRef {
final val stateOffset = Unsafe.instance.objectFieldOffset(RxSubjectActorRef.getClass.getDeclaredField("_stateDoNotCallMeDirectly"))
final val watchedByOffset = Unsafe.instance.objectFieldOffset(RxSubjectActorRef.getClass.getDeclaredField("_watchedByDoNotCallMeDirectly"))
}*/
package object RX extends RxSupport
2015 年 9 月 10 日更新
我想在这里添加一些更简单的代码来实现?? 运算符。这与上述略有不同，因为 a) 它不支持网络数据，b) 它返回 Observable[Observable[A]]，这使得同步响应更容易。优点是它不会弄乱 Akka 内脏：
object TypedAskSupport {
import scala.concurrent.Future
import akka.actor.{ActorRef,ActorSelection}
import scala.reflect.ClassTag
implicit class TypedAskableActorRef(actor : ActorRef) {
val converted : akka.pattern.AskableActorRef = actor
def ?[R](topic : Subscribe[R])(implicit timeout : akka.util.Timeout) : Future[Observable[R]] =
converted.ask(topic).mapTo[Observable[R]]
def ??[R](topic : Subscribe[R])(implicit timeout : akka.util.Timeout, execCtx : scala.concurrent.ExecutionContext) : Observable[Observable[R]] =
Observable.from (this.?[R](topic)(timeout))
def ?[R](topic : Request[R])(implicit timeout : akka.util.Timeout) : Future[R] =
converted.ask(topic).asInstanceOf[Future[R]]
def ??[R](topic : Request[R])(implicit timeout : akka.util.Timeout, execCtx : scala.concurrent.ExecutionContext) : Observable[R] =
Observable.from { this.?[R](topic)(timeout) }
}
implicit class TypedAskableActorSelection(actor : ActorSelection) {
val converted : akka.pattern.AskableActorSelection = actor
def ?[R](topic : Subscribe[R])(implicit timeout : akka.util.Timeout) : Future[Observable[R]] =
converted.ask(topic).mapTo[Observable[R]]
def ??[R](topic : Subscribe[R])(implicit timeout : akka.util.Timeout, execCtx : scala.concurrent.ExecutionContext) : Observable[Observable[R]] =
Observable.from (this.?[R](topic)(timeout))
def ?[R](topic : Request[R])(implicit timeout : akka.util.Timeout) : Future[R] =
converted.ask(topic).asInstanceOf[Future[R]]
}
}