AKKA學習(一)

AKKA簡介

什么是AKKA

Akka是一個由Scala編寫的,能兼容Sacala和JAVA的,用於編寫高可用和高伸縮性的Actor模型框架.它基於了事件驅動的並發處理模式,性能非常的高,並且有很高的可用性.大大的簡化了我們在應用系統中開發並發處理的過程.它在各個領域都有很好的表現.

使用AKKA的好處

就如上面簡介中所說的,AKKA把並發操作的各種復雜的東西都統一的做了封裝.我們主要關心的是業務邏輯的實現,只需要少量的關心Actor模型的串聯即可構建出高可用,高性能,高擴展的應用.

Akka for JAVA

由於AKKA是使用Scala編寫的,而Scala是一種基於JVM的語言.因此JAVA對AKKA的支持也是很不錯的.Akka自身又是采用微內核的方式來實現的,這就意味着能很容易的在自己的項目中應用AKKA,只需要引入幾個akka的Lib包即可.而官方直接就提供了Maven庫供我們在JAVA中使用AKKA.
這些AKKA的依賴包主要有:

• akka-actor:最核心的依賴包,里面實現了Actor模型的大部分東西
• akka-agent:代理/整合了Scala中的一些STM特性
• akka-camel:整合了Apache的Camel
• akka-cluster:akka集群依賴,封裝了集群成員的管理和路由
• akka-kernel:akka的一個極簡化的應用服務器,可以脫離項目單獨運行.
• akka-osgi:對OSGI容器的支持,有akka的最基本的Bundle
• akka-remote:akka遠程調用
• akka-slf4j:Akka的日志事件監聽
• akka-testkit:Akka的各種測試工具
• akka-zeromq:整合ZeroMQ
  其中最總要的就是akka-actor,最簡單的AKKA使用的話,只需要引入這個包就可以了.

Actor模型

什么是Actor

既然說AKKA是一個Actor模型框架,那么就需要搞清楚什么是Actor模型.Actor模型是由Carl Hewitt於上世紀70年代提出的,目的是為了解決分布式編程中的一系列問題而產生.
在Actor模型中,一切都可以抽象為Actor.
而Actor是封裝了狀態和行為的對象,他們的唯一通訊方式就是交換消息,交換的消息放在接收方的郵箱(Inbox)里.也就是說Actor之間並不直接通信,而是通過了消息來相互溝通,每一個Actor都把它要做的事情都封裝在了它的內部.
每一個Actor是可以有狀態也可以是無狀態的,理論上來講,每一個Actor都擁有屬於自己的輕量級線程,保護它不會被系統中的其他部分影響.因此,我們在編寫Actor時,就不用擔心並發的問題.
通過Actor能夠簡化鎖以及線程管理,Actor具有以下的特性:

• 提供了一種高級的抽象,能夠封裝狀態和操作.簡化並發應用的開發.
• 提供了異步的非阻塞的/高性能的事件驅動模型
• 超級輕量級的線程事件處理能力.

要在JAVA中實現一個Actor也非常的簡單,直接繼承akka.actor.UntypedActor類,然后實現public void onReceive(Object message) throws Exception方法即可.

Actor系統

光有一個一個獨立的Actor顯然是不行的.Akka中還有一個Actor System.
Actor System統管了Actor,是Actor的系統工廠或管理者,掌控了Actor的生命周期.

如上圖所示,我們可以通過ActorSystem.create來創建一個ActorSystem的實例.然后通過actorOf等方法來獲取ActorRef對象.ActorRef即為Actor Reference.它是Actor的一個引用,主要的作用是發送消息給它表示的Actor.而Actor可以通過訪問self()或sender()方法來獲取到自身或消息發送者的Actor引用.通過引用發送消息.在Akka中,Actor之間永遠都不能直接的通信,必須通過他們的代理ActorRef建立通信.

Actor路徑

為了實現一切事物都是Actor,為了能把一個復雜的事物划分的更細致.Akka引入了父子Actor.也就是Actor是有樹形結構的關系的.這樣的父子結構就能遞歸的把任何復雜的事物原子化.這也是Actor模型的精髓所在.這樣做不僅使任務本身被清晰地划分出結構,而且最終的Actor也能按照他們明確的消息類型以及處理流程來進行解析.這樣的遞歸結構使得消息能夠在正確的層次進行處理.

為了能管理父子結構的Actor,Akka又引入了Actor Path,也就是Actor路徑.
Actor路徑使用類似於URL的方式來描述一個Actor,Actor Path在一個Actor System中是唯一的.通過路徑,可以很明確的看出某個Actor的父級關系是怎樣的.

1 2 3 4 5 6 7 8

//本地Actor "akka://my-sys/user/service-a/worker1"   //遠程Actor "akka.tcp://[e-mail]:2552/user/service-b"   //集群Actor服務 "cluster://my-cluster/service-c"

以上三種就是Akka中支持的Actor路徑. 每一個通過ActorSystem創建出來的Actor都會有一個這樣的路徑.也可以通過這個路徑從ActorSystem中獲取一個Actor.

當我們創建一個ActorSystem的時候,AKKA會為該System默認的創建三個Actor,並處於不同的層次:

其中的root guardian是所有Actor的父.
而UserActor是所有用戶創建的Actor的父.它的路徑是/user,通過system.actorOf()創建出來的Actor都算是用戶的Actor,也都是這個Actor的子.
SystemActor是所有系統創建的Actor的父.它的路徑是/system,主要的作用是提供了一系列的系統的功能.

當我們查找一個Actor的時候,可以使用ActorSystem.actorSelection()方法.並且可以使用絕對路徑或者相對路徑來獲取.如果是相對路徑,那么..表示的是父Actor.比如:

ActorSelection selection = system.actorSelection( "../brother");

ActorRef actor = selection.anchor();

selection.tell(xxx);

同時,也可以通過通配符來查詢邏輯的Actor層級,比如:

ActorSelection selection = system.actorSelection( "../*");

selection.tell(xxx);

這個就表示把消息發送給當前Actor之外的所有同級的Actor.

Hello AKKA Demo

原理講了這么多,那么我們就來看一看一個最簡單的Akka的例子吧.
這個是一個最簡單的打招呼的例子,這個例子中,定義了招呼,打招呼的人兩個對象或者說消息.然后定義了執行打招呼和打印招呼兩個Actor.然后通過ActorSystem整合整個打招呼的過程.

Greet.java

/**

* 用於表示執行打招呼這個操作的消息

* @author SUN

* @version 1.0

* @Date 16/1/6 21:43

*/

public class Greet implements Serializable {

}

Greeting.java

/**

* 招呼體,里面有打的什么招呼

* @author SUN

* @version 1.0

* @Date 16/1/6 21:44

*/

public class Greeting implements Serializable {

public final String message;

public Greeting(String message) {

this.message = message;

}

}

WhoToGreet.java

/**

* 打招呼的人

* @author SUN

* @version 1.0

* @Date 16/1/6 21:41

*/

public class WhoToGreet implements Serializable {

public final String who;

public WhoToGreet(String who) {

this.who = who;

}

}

Greeter.java

/**

* 打招呼的Actor

* @author SUN

* @version 1.0

* @Date 16/1/6 21:40

*/

public class Greeter extends UntypedActor{

 

String greeting =  "";

 

@Override

public void onReceive(Object message) throws Exception {

if (message instanceof WhoToGreet)

greeting =  "hello, " + ((WhoToGreet) message).who;

else if (message instanceof Greet)

// 發送招呼消息給發送消息給這個Actor的Actor

getSender().tell( new Greeting(greeting), getSelf());

 

else unhandled(message);

}

}

GreetPrinter.java

/**

* 打印招呼

* @author SUN

* @version 1.0

* @Date 16/1/6 21:45

*/

public class GreetPrinter extends UntypedActor{

 

@Override

public void onReceive(Object message) throws Exception {

if (message instanceof Greeting)

System.out.println(((Greeting) message).message);

}

}

DemoMain.java

/**

* @author SUN

* @version 1.0

* @Date 16/1/6 21:39

*/

public class DemoMain {

 

public static void main(String[] args) throws Exception {

final ActorSystem system = ActorSystem.create("helloakka");

 

// 創建一個到greeter Actor的管道

final ActorRef greeter = system.actorOf(Props.create(Greeter.class), "greeter");

 

// 創建郵箱

final Inbox inbox = Inbox.create(system);

 

// 先發第一個消息,消息類型為WhoToGreet

greeter.tell( new WhoToGreet("akka"), ActorRef.noSender());

 

// 真正的發送消息,消息體為Greet

inbox.send(greeter,  new Greet());

 

// 等待5秒嘗試接收Greeter返回的消息

Greeting greeting1 = (Greeting) inbox.receive(Duration.create( 5, TimeUnit.SECONDS));

System.out.println( "Greeting: " + greeting1.message);

 

// 發送第三個消息,修改名字

greeter.tell( new WhoToGreet("typesafe"), ActorRef.noSender());

// 發送第四個消息

inbox.send(greeter,  new Greet());

 

// 等待5秒嘗試接收Greeter返回的消息

Greeting greeting2 = (Greeting) inbox.receive(Duration.create( 5, TimeUnit.SECONDS));

System.out.println( "Greeting: " + greeting2.message);

 

// 新創建一個Actor的管道

ActorRef greetPrinter = system.actorOf(Props.create(GreetPrinter.class));

 

//使用schedule 每一秒發送一個Greet消息給 greeterActor,然后把greeterActor的消息返回給greetPrinterActor

system.scheduler().schedule(Duration.Zero(), Duration.create( 1, TimeUnit.SECONDS), greeter, new Greet(), system.dispatcher(), greetPrinter);

//system.shutdown();

}

}

以上就是整個Demo的所有代碼,並不多.接下來我們就分析一下這個程序.

首先是定義的幾個消息.在Akka中傳遞的消息必須實現Serializable接口.WhoToGreet消息表示了打招呼的人,Greeting表示了招呼的內容,而Greet表示了打招呼這個動作.

接着就是兩個最重要的Actor了.GreetPrinter非常簡單,接收到消息后,判斷消息的類型,如果是Greeting招呼內容,那么就直接打印消息到控制台.而Greeter這個Actor稍微復雜點,它消費兩種不同的消息,如果是WhoToGreet,那么就把要打招呼的人記錄到自己的上下文中,如果是Greet,那么就構造出招呼的內容,並把消息反饋回sender.

最后,再來分析下DemoMain.

1. 一來,先創建了一個ActorSystem,
2. 然后創建了一個GreeterActor的實例,命名為greeter.
3. 接着,為這個Actor,顯示的創建了一個郵箱.
4. 而后,調用greeter.tell(new WhoToGreet("akka"), ActorRef.noSender());,表示給greeter這個Actor發送一個消息,消息的內容是WhoToGreet,發送者是空.這就意味着在greeter這個Actor內部,調用sender是不能獲取到發送者的.通過這個動作,就把消息限定為了單向的.
5. 再然后,通過inbox.send(greeter, new Greet());,使用郵箱顯示的發送一個Greet消息給greeter.這是給Actor發送消息的另外一種方法,這種方法通常會有更高的自主性,能完成更多更復雜的操作.但是調用起來比直接使用ActorRef來的復雜.
6. Greeting greeting1 = (Greeting) inbox.receive(Duration.create(5, TimeUnit.SECONDS));表示的就是嘗試在5秒鍾內,從Inbox郵箱中獲取到反饋消息.如果5秒內沒有獲取到,那么就拋出TimeoutException異常. 由於我們在greeter這個Actor中有處理,接收到Greet消息后,就構造一個Greeting消息給sender,因此這個地方是能夠正確的獲取到消息的反饋的.
7. 后面的操作都是一樣的,就不再重復描述.
8. 只有最后一個代碼稍微有點不一樣system.scheduler().schedule(Duration.Zero(), Duration.create(1, TimeUnit.SECONDS), greeter, new Greet(), system.dispatcher(), greetPrinter);,這個使用了ActorSystem中的調度功能.每一秒鍾給greeter這個Actor發送一個Greet消息,並指定消息的發送者是greetPrinter.這樣每隔一秒鍾,greeter就會收到Greet消息,然后構造成Greeting消息,又返回給GreetPrinter這個Actor,這個Actor接收到消息后,打印出來.形成一個環流.