3

让广播演员随时间动态添加/删除路由的正确方法是什么？

问题的背景：一个参与者监听特定项目的价格变化，然后它将价格变化广播给所有其他参与者（路由），他们根据内部规则行事（例如，如果价格是 X 买入或卖出）。

我是 Akka 的新手，但是在阅读了文档之后，我相信我已经弄清楚了所需的组件，但是如果您觉得我的设计或使用的组件不正确，请发表评论/回答。

我想从固定的路由列表中更改

        ActorRef actor1 = system.actorOf(new Props(LimitOrderActor.class));
        ActorRef actor2 = system.actorOf(new Props(LimitOrderActor.class));
        ActorRef actor3 = system.actorOf(new Props(LimitOrderActor.class));
        Iterable<ActorRef> routees = Arrays.asList(new ActorRef[] { actor1, actor2, actor3 });
        ActorRef actorBroadcastRouter1 = system.actorOf(new Props(TickerWatcherActor.class).withRouter(BroadcastRouter.create(routees)), "router1");

类似于动态大小的广播路由器，在广播路由器启动并运行后创建演员

        int lowerBound = 1;
        int upperBound = 10000;
        DefaultResizer resizer = new DefaultResizer(lowerBound, upperBound);
        BroadcastRouter broadcastRouter2 = new BroadcastRouter(resizer);
        ActorRef actorBroadcastRouter2 = system.actorOf(new Props(TickerWatcherActor.class).withRouter(broadcastRouter2), "router2");

        ActorRef actor4 = system.actorOf(new Props(LimitOrderActor.class).withRouter((RouterConfig) broadcastRouter2));
        ActorRef actor5 = system.actorOf(new Props(LimitOrderActor.class).withRouter((RouterConfig) broadcastRouter2));
        ActorRef actor6 = system.actorOf(new Props(LimitOrderActor.class).withRouter((RouterConfig) broadcastRouter2));

现在，演员“actorBroadcastRouter2”收到的消息不是预期的 LimitOrderActor 演员 4、5 和 6。我做错了什么？

编辑：我相信现在我正在寻找的是事件总线而不是广播路由器

   final ActorRef actor = system.actorOf(new Props(LimitOrderActor.class));
   system.eventStream().subscribe(actor, String.class);
4

2 回答 2

0

让我们结束这个问题，我在同样的问题上磕磕绊绊。这不是广播路由器的工作方式。您在 中创建新的广播路由器actor4，actor5而actor6不是在 中添加到现有的广播路由器actorBroadcastRouter2。

您想要的在 scala 中非常简单，具有 Listeners 特征，遗憾的是在使用 Java 编程 Akka 时不可用：

class MyActor extends Actor with Listeners {
  def receive = {
     case yourmessages => gossip("To All my listeners")
  } orElse listenerManagement
}

val myActor = context.actorOf(Props[MyActor])

myActor ! Listen(someActorRef)

myActor ! "pigdog"

必须做自己的 Java 版本或使用DistributedPubSubExtension更成熟的东西。

https://groups.google.com/forum/?fromgroups=#!topic/akka-user/NEOY9IxRW5I

https://github.com/akka/akka/blob/master/akka-actor/src/main/scala/akka/routing/Listeners.scala

http://doc.akka.io/docs/akka/snapshot/contrib/distributed-pub-sub.html#a-small-example-in-java

于 2013-04-25T10:19:43.097 回答
0

使用事件总线而不是广播路由器：

final ActorRef actor = system.actorOf(new Props(LimitOrderActor.class));
system.eventStream().subscribe(actor, String.class);
于 2015-08-21T16:51:44.527 回答