Akka：动态地将参与者添加到路由器

Question

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

问题的背景：参与者监听特定项目的价格变化，然后向所有其他参与者（路由者）广播价格变化，他们根据内部规则行事（例如，如果价格是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");

类似于动态大小的BroadcastRouter，其中在BroadcastRouter启动并运行后创建参与者

        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);

Answer 1

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

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

Answer 2

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

在scala中，您需要的是非常简单的侦听器特性，遗憾的是，在使用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用户/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-java中的小示例