akka是一个工具包，试用于可扩展、弹性的，多核并发和跨网络交互的系统。
    可以简单理解为是一个实现了actor model的工具包和运行时。

    Akka actor实现
    actor system：

    The actor system as a collaborating ensemble of actors is the natural unit for managing shared facilities like scheduling services, configuration, logging, etc. Several actor systems with different configurations may co-exist within the same JVM without problems, there is no global shared state within Akka itself, however the most common scenario will only involve a single actor system per JVM.

    actor：

    An actor is a container for State, Behavior, a Mailbox, Child Actors and a Supervisor Strategy. All of this is encapsulated behind an Actor Reference. One noteworthy aspect is that actors have an explicit lifecycle, they are not automatically destroyed when no longer referenced; after having created one, it is your responsibility to make sure that it will eventually be terminated as well—which also gives you control over how resources are released When an Actor Terminates.

    actor lifecycle：
    actor_lifecycle.png
    Demo代码：
    flink使用的是旧版的 akka API，也就是官网上的classic akka。demo也是用classic。

    1. public class EchoActor extends AbstractActor {
    2.   private final LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this);
    4.   @Override
    5.   public Receive createReceive() {
    6.     return receiveBuilder()
    7.         .match(
    8.             String.class,
    9.             s -> {
    10.                 log.info("Received String message: {}", s);
    11.                 if(!getSender().isTerminated()){
    12.                     getSender().tell("echo:" + s, getSelf());
    13.                 }
    14.             })
    15.         .matchAny(o -> log.info("received unknown message"))
    16.         .build();
    17.   }
    18. }
    20. resouces目录添加配置文件application.conf
    21. akka {
    22.   actor {
    23.     # provider=remote is possible, but prefer cluster
    24.     provider = remote
    25.   }
    26.   remote.artery.enabled = false
    27.   remote.classic {
    28.     enabled-transports = ["akka.remote.classic.netty.tcp"]
    29.     netty.tcp {
    30.       hostname = "127.0.0.1"
    31.       #0动态选择port
    32.       port = 0
    33.     }
    34.   }
    35. }
    37. public class Server {
    38.     public static void main(String[] args) throws ExecutionException, InterruptedException {
    39.         ActorSystem actorSystem = ActorSystem.create("server");
    41.         final ActorRef deadLettersActor =
    42.                 actorSystem.actorOf(DeadLettersActor.getProps(), "deadLettersActor");
    43.         actorSystem.eventStream().subscribe(deadLettersActor, DeadLetter.class);
    45.         //创建Actor并获取ActorRef
    46.         ActorRef myactorRef = actorSystem.actorOf(Props.create(EchoActor.class), "EchoServer");
    48.         System.out.println(myactorRef.path());
    49.         //tell非阻塞异步通信
    50.         myactorRef.tell("ssss", null);
    51.         //tell非阻塞异步通信，获取结果future
    52.         CompletableFuture<Object> future1 = Patterns.ask(myactorRef, "aaaa", Duration.ofMillis(1000)).toCompletableFuture();
    53.         actorSystem.log().info("response:{}", future1.get());
    54. //        actorSystem.terminate();
    55.     }
    56. }
    58. public class Client {
    59.     public static void main(String[] args) throws ExecutionException, InterruptedException {
    60.         ActorSystem actorSystem = ActorSystem.create("client");
    61.         //根据url获取ActorSelection
    62.         ActorSelection echoServer = actorSystem.actorSelection("akka.tcp://server@127.0.0.1:11880/user/EchoServer");
    64.         System.out.println(echoServer.path());
    66.         echoServer.tell("ssss", null);
    67.         CompletableFuture<Object> future1 = Patterns.ask(echoServer, "aaaa", Duration.ofMillis(1000)).toCompletableFuture();
    68.         actorSystem.log().info("response:{}", future1.get());
    69.         actorSystem.terminate();
    70.     }
    71. }