akka介绍

太过爱你忘了你带给我的痛 2022-09-26 15:58 377阅读 0赞

**1、概述**

Akka是JAVA虚拟机JVM平台上构建**高并发、分布式和容错应用**的工具包和运行时。Akka用Scala语言写成，同时提供了Scala和JAVA的开发接口。

Akka处理并发的方法基于Actor模型。在Akka里，Actor之间通信的唯一机制就是消息传递。Akka框架支持两种语言Java和Scala。Akka是一个运行时与编程模型一致的系统。

**2、Akka中的Actor是什么**

Actor本质上就是接收消息并采取行动处理消息的对象。它从消息源中解耦出来，只负责正确识别接收到的消息类型，并采取相应的行动。

收到一条消息之后，一个Actor可能会采取以下一个或多个行动：  
执行一些本身的操作（例如进行计算、持久化数据、调用外部的Web服务等）；  
把消息或衍生消息转发给另外一个Actor；  
实例化一个新的Actor并把消息转发给它。

或者，如果这个Actor认为合适的话，可能会完全忽略这条消息（也就是说，它可能选择不响应）。

**3、Akka使用场景**  
任何需要高吞吐率和低延迟的系统都可以使用Akka。  
Actor使你能够进行服务失败管理（监管者），负载管理（缓和策略、超时和隔离），水平和垂直方向上的可扩展性（增加cpu核数和/或增加更多的机器）管理。

**4、准备工作**  
Akka要求你安装了 Java 1.6或更高版本。

**下载**  
下载Akka有几种方法。你可以下载包含微内核的完整发布包（包含所有的模块）. 或者也可以从Maven或sbt从Akka Maven仓库下载对akka的依赖。

**模块**  
Akka的模块化做得非常好，它为不同的功能提供了不同的Jar包。

akka-actor-2.0.jar – 标准Actor, 有类型Actor，等等  
akka-remote-2.0.jar – 远程Actor  
akka-slf4j-2.0.jar – SLF4J事件处理监听器  
akka-testkit-2.0.jar – 用于测试Actor的工具包  
akka-kernel-2.0.jar – Akka微内核，可运行一个基本的最小应用服务器  
akka--mailbox-2.0.jar – Akka可容错邮箱

要查看每个Akka模块的jar包依赖见 依赖 章节. 虽然不重要不过akka-actor 没有外部依赖 (除了 scala-library.jar JAR包).

使用发布版：[http://akka.io/downloads][http_akka.io_downloads]，下载发布包并解压.

**5、maven配置**

<dependency>
        <groupId>com.typesafe.akka</groupId>
        <artifactId>akka-actor_2.10</artifactId>
        <version>2.1.2</version>
    </dependency>

**6、akka使用**

（1）作为一个库，以普通jar包的形式放在classpath上，或放到web应用中的 WEB-INF/lib位置  
（2）作为一个独立的应用程序，使用 Microkernel（微内核），自己有一个main类来初始化Actor系统  
将Akka作为一个库。如果你是编写web应用，你估计要使用这种方式。通过添加更多的模块，可以有多种以库的形式使用Akka的方式。

**7、akka的Java API**  
**（1）Creating Actors**

/* Creating Actors xtending the UntypedActor class and implementing the onReceive method */
    import akka.actor.UntypedActor;
    import akka.event.Logging;
    import akka.event.LoggingAdapter;
    
    public class MyUntypedActor extends UntypedActor { 
      LoggingAdapter log = Logging.getLogger(getContext().system(), this);
    
      @override
      public void onReceive(Object message) throws Exception {
        if (message instanceof String)
          log.info("Received String message: {}", message);
        else
          unhandled(message);
      }
    }

UntypedActor只定义了一个抽象方法，就是上面提到的onReceive(Objectmessage), 用来实现actor的行为。

如果当前actor的行为与收到的消息不匹配，则会调用unhandled方法, 它的缺省实现是向actor系统的事件流中发布一条 akka.actor.UnhandledMessage(message, sender, recipient)。

另外，它还包括:

getSelf(); //代表本actor的 ActorRef
    getSender(); //代表最近收到的消息的发送actor，通常用于下面将讲到的回应消息中
    supervisorStrategy(); //用户可重写它来定义对子actor的监管策略
    getContext(); //暴露actor和当前消息的上下文信息

**（2）Send messages**  
tell 意思是“fire-and-forget”， 异步发送一个消息并立即返回。这是发送消息的推荐方式。不会阻塞地等待消息。它拥有最好的并发性和可扩展性。ask 异步发送一条消息并返回一个 Future代表一个可能的回应。需要采用Future的处理模式。

public class MyReceivedTimeoutUntypedActor extends UntypedActor { 
    
        public MyReceivedTimeoutUntypedActor() {
            getContext().setReceiveTimeout(Duration.parse("30 seconds"));
        }
    
        public void onReceive(Object message) {
            if (message.equals("Hello")) {
                getSender().tell("Hello world");
            } else if (message == Actors.receiveTimeout()) {
                throw new RuntimeException("received timeout");
            } else {
                unhandled(message);
            }
        }
    }

如果服务端处理消息时发生了异常而导致没有给客户端回应，那么客户端收到的结果将会收到Timeout的Failure：Failure(akka.pattern.AskTimeoutException: Timed out)。可以将异常捕获用Failure封装异常发给客户端：actor.tell(new akka.actor.Status.Failure(e))。

Future的onComplete, onResult, 或 onTimeout 方法可以用来注册一个回调，以便在Future完成时得到通知。从而提供一种避免阻塞的方法。

Future<Object> future = Patterns.ask(queryActor, param, timeout);
    
    future.onComplete(new OnComplete<Object>() {
    
        public void onComplete(Throwable failure, Object result) {
    
            if (failure != null) {
    
                // time out exception occurs
                if (failure.getMessage() != null) {
                    logger.error(failure.getMessage(), failure);
                }
    
                // mark as timeout error
                res.resume(Response.ok().entity(result).build());
    
            } else {
                res.resume(Response.ok().entity(result).build());
            }
    
            logger.info("query cost " + stopwatch.elapsedTime(TimeUnit.MILLISECONDS) + " query is " + param);
        }
    }, actorSystem.dispatcher());

[http_akka.io_downloads]: http://akka.io/downloads