如何在 Play 框架 2 应用程序中存储 Akka 参与者列表?

2023-12-25

我有一个 Play 框架 2 应用程序,可以接收数据并通过 WebSocket 将其发送到多个客户端。我使用 Akka actor 来处理 WebSockets,就像本文档 http://www.playframework.com/documentation/2.3.x/JavaWebSockets。我也有一个WebSocketRouter扩展的类UntypedActor并包含路由逻辑(决定将系统接收到的数据传递给哪些客户端)。我知道我可以使用RouterAkka 的功能,但这对我来说不是目前的问题。问题是我必须存储所有活跃客户的列表。现在我将它存储在静态列表中WebSocketRouter班级。这是编写概念验证原型的最快方法,但它不是线程安全的,并且似乎不是“Akka 方式”。 下面是一个简化的代码示例:

WebSocket控制器:

//This controller handles the creation of WebSockets.
public class WebSocketController extends Controller {
    public static WebSocket<String> index() {
        return WebSocket.withActor(new F.Function<ActorRef, Props>() {
            public Props apply(ActorRef out) throws Throwable {
                return MessageSender.props(out);
            }
        });
    }
}

消息发送者:

//Hold a reference to the auto-created Actor that handles WebSockets 
//and also registers and unregisters itself in the router.
public class  MessageSender extends UntypedActor {

    public static Props props(ActorRef out) {
        return Props.create(MessageSender.class, out);
    }

    private final ActorRef out;

    public MessageSender(ActorRef out) {
        this.out = out;
    }

    @Override
    public void preStart() {
        WebSocketRouter.addSender(getSelf());
    }

    @Override
    public void onReceive(Object message) throws Exception {
        if (message instanceof String) {
            out.tell(message, getSelf());
        }
        else {
            unhandled(message);
        }
    }

    public void postStop() {
        WebSocketRouter.removeSender(getSelf());
    }
}

WebSocket 路由器:

public class WebSocketRouter extends UntypedActor {
    private static ArrayList<ActorRef> senders;
    static {
        senders = new ArrayList<>();
    }

    public static void addSender(ActorRef actorRef){
        senders.add(actorRef);
    }

    public static void removeSender(ActorRef actorRef){
        senders.remove(actorRef);
    }

    @Override
    public void onReceive(Object message) throws Exception {
        if (message instanceof String) {
            for (ActorRef sender : senders) {
                sender.tell(message, getSelf());
            }
        }
    }
}

我再次知道这是一个糟糕的解决方案,我正在寻找更好的解决方案。我想过创建一个线程安全的单例类来保存当前连接。我还考虑过在某个 Akka actor 的实例中保存当前连接列表,并通过 Akka 消息修改该列表,但为了这种方式工作,我必须存储一个ActorRef静态地分配给该参与者,以便可以从不同的地方访问它ActorSystems.

解决我的问题最适合 Akka 意识形态的最佳方法是什么?


而不是对 Actor 进行静态引用(WebSocketRouter在你的情况下),为什么不想出一些消息来发送它呢?这样,参与者就可以以一致的方式维护自己的内部状态。通过消息进行状态更改是 Actor 模型的主要好处之一。

在开始编写代码之前,如果这不是 100% 准确,我很抱歉,我只使用了 Akka 的 Scala 版本,并且基于对 Akka 的快速扫描Akka 文档 http://doc.akka.io/docs/akka/snapshot/java/untyped-actors.html.

所以在你的例子中,我会定义一些对象来表达加入/离开......

public class JoinMessage { } 
public class ExitMessage { } 

注意ExitMessage仅当您打算保持 WebSocket 打开并让用户停止侦听路由器时才需要。否则,路由器可以检测到 Actor 何时被终止。

然后你会改变你的MessageSender演员在加入或离开聊天室时发送这些消息......

public class MessageSender extends UntypedActor {

    public static Props props(ActorRef out) {
        return Props.create(MessageSender.class, out);
    }

    private final ActorRef out;
    private final ActorRef router;

    public MessageSender(ActorRef out) {
        this.out = out;
        this.router= getContext().actorSelection("/Path/To/WebSocketRouter");
    }

    @Override
    public void preStart() {
        router.tell(new JoinMessage(), getSelf());
    }

    @Override
    public void onReceive(Object message) throws Exception {
        if (message instanceof String) {
            out.tell(message, getSelf());
        } else {
            unhandled(message);
        }
    }    
}

然后你的路由器可以更改为在内部管理状态,而不是在 Actor 上公开内部方法(正如你所知,这不好)......

public class WebSocketRouter extends UntypedActor {
    private final Set<ActorRef> senders = new HashSet<>();

    private void addSender(ActorRef actorRef){
        senders.add(actorRef);
    }

    private void removeSender(ActorRef actorRef){
        senders.remove(actorRef);
    }

    @Override
    public void onReceive(Object message) throws Exception {
        if (message instanceof JoinMessage) {
            addSender(sender);
            getContext().watch(sender); // Watch sender so we can detect when they die.
        } else if (message instanceof Terminated) {
            // One of our watched senders has died.
            removeSender(sender);
        } else if (message instanceof String) {
            for (ActorRef sender : senders) {
                sender.tell(message, getSelf());
            }
        }
    }
}

同样,此代码旨在让您了解如何利用 Actor 模型来完成此任务。抱歉,如果 Java 不是 100% 准确,但希望您能遵循我的意图。

本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)

如何在 Play 框架 2 应用程序中存储 Akka 参与者列表? 的相关文章

随机推荐