java - 管理 Akka Actor 系统生命周期

Question

我有一个 Actor 系统，我想要启动它并使其一直运行，直到特定 Actor(Destructor)收到 ShutdownNow 消息。当它收到这样的消息时，它基本上会执行相当于关闭钩子(Hook)的操作，关闭整个 Actor 系统，然后关闭父 JVM 进程/主线程:如下所示:

class MyApp {
    static void main(String[] args) {
        // Do some stuff before starting the actor system.

        ActorSystem actorSystem = ActorSystem.create(“myapp-actorsys”)
        ActorRef initializer = actorSystem.actorOf(Props.create(Initializer) “initializer”)
        Inbox inbox = Inbox.create(actorSystem)

        inbox.send(initializer, new Initialize())

        // Now the actor system is initialized and is running
        // Maybe its processing data off a message broker, etc.

        // Here I’d like to wait until the actor system shuts down (see Destructor below)
        while(actorSystem.isRunning()) { }

        // Now do some final tasks.
    } // <- end of main thread/parent JVM process
}

class Destructor extends UntypedActor {
    @Override
    void onReceive(Object msg) {
        if(msg instanceof ShutdownNow) {
            // Shut down the actor system such that actorSystem.isRunning()
            // will return true.
        }
    }
}

这里的想法是，父 JVM 进程将保持 Activity 状态，除非被底层操作系统(sigkill 等)中断或 Actor 系统内的某些内容发出 ShutdownNow 消息给适当的 Actor 。

如何实现这个？

Answer 1

不知道您正在使用的确切设置是什么:

在您的 Actor 系统停止之前，Java 进程不会停止。因此，您无需执行任何操作，只需从析构函数 Actor 发送关闭命令即可。

以下代码对我有用(既可以使用参与者从进程内部停止进程，也可以使用kill从外部停止进程):

import java.util.concurrent.TimeUnit;

import scala.concurrent.duration.Duration;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.actor.UntypedActor;


public class MyApp {

    public static void main(String[] args) {
        // Do some stuff before starting the actor system.
       System.out.println("Start Actor System");
       ActorSystem actorSystem = ActorSystem.create("myapp-actorsys");
       ActorRef destructor = actorSystem.actorOf(Props.create(Destructor.class));

       //Java process won't shut down before actor system stops
       //Send message after timeout to stop system

       actorSystem.scheduler().scheduleOnce(Duration.create(10, TimeUnit.SECONDS),
               destructor, new ShutdownNow(), actorSystem.dispatcher(), null);
    }

    public static class Destructor extends UntypedActor {

        public Destructor() {

        }

        @Override
        public void onReceive(Object msg) {
            if(msg instanceof ShutdownNow) {
                //Shutdown system to stop java process
                System.out.println("Stop Actor System");
                getContext().system().shutdown();
            }
        }
    }

    public static class ShutdownNow {

    }
}