|
轉(zhuǎn)載請(qǐng)注明出處：http://blog.csdn.net/jmppok/article/details/17264495
本文結(jié)合網(wǎng)上一些資料,，對(duì)他們進(jìn)行整理，摘選和翻譯而成,，對(duì)Akka進(jìn)行簡(jiǎn)要的說明,。引用資料在最后列出。
1.什么是Akka
Akka 是一個(gè)用 Scala 編寫的庫(kù),，用于簡(jiǎn)化編寫容錯(cuò)的、高可伸縮性的 Java 和 scala 的 Actor 模型應(yīng)用,。
官方網(wǎng)站 （http:///）的介紹是：
Akka is a toolkit and runtime for building highly concurrent, distributed, and fault tolerant event-driven applications on the JVM.
Build powerful concurrent & distributed applications more easily.
翻譯成中文就是：Akka是一個(gè)開發(fā)庫(kù)和運(yùn)行環(huán)境,，可以用于構(gòu)建高并發(fā)、分布式,、可容錯(cuò),、事件驅(qū)動(dòng)的基于JVM的應(yīng)用。使構(gòu)建高并發(fā)的分布式應(yīng)用更加容易,。
Akka可以以兩種不同的方式來使用
- 以庫(kù)的形式：在web應(yīng)用中使用,，放到 WEB-INF/lib 中或者作為一個(gè)普通的Jar包放進(jìn)classpath。
- 以微內(nèi)核的形式：你可以將應(yīng)用放進(jìn)一個(gè)獨(dú)立的內(nèi)核,。
2.Akka的五大特性
1）易于構(gòu)建并行和分布式應(yīng)用 （Simple Concurrency & Distribution）
Akka在設(shè)計(jì)時(shí)采用了異步通訊和分布式架構(gòu),，并對(duì)上層進(jìn)行抽象，如Actors,、Futures ,，STM等。
2）可靠性（Resilient by Design）
系統(tǒng)具備自愈能力,，在本地/遠(yuǎn)程都有監(jiān)護(hù),。
3）高性能（High Performance）
在單機(jī)中每秒可發(fā)送50000000個(gè)消息。內(nèi)存占用小,，1GB內(nèi)存中可保存2500000個(gè)actors,。
4）彈性，無(wú)中心（Elastic — Decentralized）
自適應(yīng)的負(fù)責(zé)均衡,，路由,，分區(qū)，配置
5）可擴(kuò)展（Extensible）
可以使用Akka 擴(kuò)展包進(jìn)行擴(kuò)展,。
3.什么場(chǎng)景下特別適合使用Akka,？
我們看到Akka被成功運(yùn)用在眾多行業(yè)的眾多大企業(yè)，從投資業(yè)到商業(yè)銀行,、從零售業(yè)到社會(huì)媒體,、仿真,、游戲和賭博、汽車和交通系統(tǒng),、數(shù)據(jù)分析等等等等,。任何需要高吞吐率和低延遲的系統(tǒng)都是使用Akka的候選。
Actor使你能夠進(jìn)行服務(wù)失敗管理（監(jiān)管者）,，負(fù)載管理（緩和策略,、超時(shí)和隔離），水平和垂直方向上的可擴(kuò)展性（增加cpu核數(shù)和/或增加更多的機(jī)器）管理,。
下面的鏈接中有一些Akka用戶關(guān)于他們?nèi)绾问褂肁kka的描述：http:///questions/4493001/good-use-case-for-akka
所有以上這些都在這個(gè)Apache2許可的開源軟件中,。
以下是Akka被部署到生產(chǎn)環(huán)境中的領(lǐng)域
事務(wù)處理 (在線游戲，金融/銀行業(yè),，貿(mào)易,，統(tǒng)計(jì)，賭博,，社會(huì)媒體,，電信):垂直擴(kuò)展，水平擴(kuò)展,，容錯(cuò)/高可用性
服務(wù)后端 (任何行業(yè),，任何應(yīng)用):提供REST, SOAP, Cometd, WebSockets 等服務(wù) 作為消息總線/集成層 垂直擴(kuò)展，水平擴(kuò)展,，容錯(cuò)/高可用性
并發(fā)/并行 (任何應(yīng)用):運(yùn)行正確,，方便使用，只需要將jar包添加到現(xiàn)有的JVM項(xiàng)目中（使用Scala,，java, Groovy或jruby)
仿真:主/從,，計(jì)算網(wǎng)格，MaReduce等等.
批處理 (任何行業(yè)):Camel集成來連接批處理數(shù)據(jù)源 Actor來分治地批處理工作負(fù)載
通信Hub (電信, Web媒體, 手機(jī)媒體):垂直擴(kuò)展,，水平擴(kuò)展,，容錯(cuò)/高可用性
游戲與賭博 (MOM, 在線游戲, 賭博):垂直擴(kuò)展，水平擴(kuò)展,，容錯(cuò)/高可用性
商業(yè)智能/數(shù)據(jù)挖掘/通用數(shù)據(jù)處理:垂直擴(kuò)展,，水平擴(kuò)展，容錯(cuò)/高可用性
復(fù)雜事件流處理:垂直擴(kuò)展,，水平擴(kuò)展,，容錯(cuò)/高可用性
4. Scala語(yǔ)言
Scala是一種多范式的編程語(yǔ)言，設(shè)計(jì)初衷是要集成面向?qū)ο缶幊毯秃瘮?shù)式編程的各種特性,。
百度百科,，Scala語(yǔ)言介紹：
http://baike.baidu.com/link?url=86mpCDT8PbqP29vR-ZDHQeMt1grr--g42DtVMkjvru4g57_TATYquSDnjLOJl-WovbPuywyikI7q1I0ZvWjuca
百度文庫(kù)，Scala編程入門：
http://wenku.baidu.com/view/27cbf218964bcf84b9d57b83.html
5.Actors模型
Actor模型并非什么新鮮事物,，它由Carl Hewitt于上世紀(jì)70年代早期提出,，目的是為了解決分布式編程中一系列的編程問題,。其特點(diǎn)如下：
系統(tǒng)中的所有事物都可以扮演一個(gè)Actor
Actor之間完全獨(dú)立
在收到消息時(shí)Actor所采取的所有動(dòng)作都是并行的，在一個(gè)方法中的動(dòng)作沒有明確的順序
Actor由標(biāo)識(shí)和當(dāng)前行為描述
Actor可能被分成原始（primitive）和非原始（non primitive）類別
很多開發(fā)語(yǔ)言都提供了原生的Actor模型,。例如erlang,scala等
Actor,，可以看作是一個(gè)個(gè)獨(dú)立的實(shí)體，他們之間是毫無(wú)關(guān)聯(lián)的,。但是,，他們可以通過消息來通信。
一個(gè)Actor收到其他Actor的信息后,，它可以根據(jù)需要作出各種相應(yīng),。消息的類型可以是任意的，消息的內(nèi)容也可以是任意的,。這點(diǎn)有點(diǎn)像webservice了,。只提供接口服務(wù)，你不必了解我是如何實(shí)現(xiàn)的,。
一個(gè)Actor如何處理多個(gè)Actor的請(qǐng)求呢？它先建立一個(gè)消息隊(duì)列,，每次收到消息后,，就放入隊(duì)列，而它每次也從隊(duì)列中取出消息體來處理,。通常我們都使得這個(gè)過程是循環(huán)的,。讓Actor可以時(shí)刻處理發(fā)送來的消息。
6.示例（http://www./Program/java/2012/03/29/67015.shtml）
應(yīng)用場(chǎng)景：服務(wù)端要處理大量的客戶端的請(qǐng)求,，并且處理請(qǐng)求耗費(fèi)較長(zhǎng)的時(shí)間,。這時(shí)就需要使用并發(fā)處理。多線程是一種方法,，這里使用Akka框架處理并發(fā),。（以下代碼在Groovy1.7.5、akka-actors-1.2下運(yùn)行成功）
這里有三個(gè)角色：Client,、Master,、Worker
Client傻乎乎地發(fā)同步請(qǐng)求給Master，一直等到結(jié)果返回客戶端才離開,。
Master接收客戶端發(fā)來的請(qǐng)求,，然后將請(qǐng)求交給Worker處理，處理完成之后將結(jié)果返回給Client,。
Worker負(fù)責(zé)具體的業(yè)務(wù)處理,，它耗費(fèi)的事件比較長(zhǎng)。
所以這里的關(guān)鍵在于Master,，如果Master線性地“接收請(qǐng)求――調(diào)用Worker處理得到返回結(jié)果――將結(jié)果返回”,，這樣的系統(tǒng)必將歇菜,。
使用Akka可以方便地將它變成并行地。
先看看Client,，模擬同時(shí)多個(gè)客戶端給Master發(fā)請(qǐng)求
import akka.actor.ActorRef
import static akka.actor.Actors.remote
class HelloClient implements Runnable {
int seq
String serviceName
HelloClient(int seq, String serviceName) {
this.seq = seq
this.serviceName = serviceName
}
void run() {
ActorRef actor = remote().actorFor(serviceName, "10.68.15.113", 9999);
String str = "Hello--" + seq
println "請(qǐng)求-----${str}"
Object res = actor.sendRequestReply(str)
println "返回-----${res}"
}
public static void main(String[] args) {
for (int i = 0; i < 5; i++) {
Thread thread = new Thread(new HelloClient(i, "hello-service"))
thread.start() //同時(shí)啟動(dòng)5個(gè)客戶端請(qǐng)求Master
}
}
}
真正干活的Worker：
import akka.actor.UntypedActor
class HelloWorker extends UntypedActor { //Worker是一個(gè)Actor,，需要實(shí)現(xiàn)onReceive方法
@Override
void onReceive(Object o) {
println "Worker 收到消息----" + o
if (o instanceof String) {
String result = doWork(o) //調(diào)用真實(shí)的處理方法
getContext().replyUnsafe(result)//將結(jié)果返回給Master
}
}
//Worker處理其實(shí)很簡(jiǎn)單，僅僅將參數(shù)字符串改造一下而已,。只不過使其sleep了20秒,，讓它變得“耗時(shí)較長(zhǎng)”
String doWork(String str) {
Thread.sleep(1000 * 20)
return "result----" + str + " 。"
}
}
負(fù)責(zé)并發(fā)調(diào)度的Master：
import akka.actor.ActorRef
import akka.actor.Actors
import akka.actor.UntypedActor
import akka.actor.UntypedActorFactory
import akka.dispatch.Future
import akka.dispatch.Futures
import java.util.concurrent.Callable
class HelloMaster extends UntypedActor {
@Override
void onReceive(Object o) {
println "Master接收到Work消息：" + o
def clientChannel = getContext().channel() //客戶端鏈接Channel
//啟動(dòng)worker actor
ActorRef worker = Actors.actorOf(new UntypedActorFactory() {
public UntypedActor create() {
return new HelloWorker();
}
}).start();
//這里實(shí)現(xiàn)真正的并發(fā)
Future f1 = Futures.future(new Callable() {
Object call() {
def result = worker.sendRequestReply(o) //將消息發(fā)給worker actor,，讓W(xué)orker處理業(yè)務(wù),，同時(shí)得到返回結(jié)果
worker.stop()
println "Worker Return----" + result
clientChannel.sendOneWay(result) //將結(jié)果返回給客戶端
return result
}
})
println "Future call over"
}
public static void main(String[] args) { //啟動(dòng)Master進(jìn)程，綁定IP,、端口和服務(wù)
Actors.remote().start("10.68.15.113", 9999).register(
"hello-service",
Actors.actorOf(HelloMaster.class));
}
}
看看客戶端的調(diào)用日志
請(qǐng)求-----Hello--4
請(qǐng)求-----Hello--1
請(qǐng)求-----Hello--3
請(qǐng)求-----Hello--0
請(qǐng)求-----Hello--2
[GENERIC] [11-10-6 下午9:49] [RemoteClientConnected(akka.remote.netty.NettyRemoteSupport@195b6aad,/10.68.15.113:9999)]
[GENERIC] [11-10-6 下午9:49] [RemoteClientStarted(akka.remote.netty.NettyRemoteSupport@195b6aad,/10.68.15.113:9999)]
返回-----result----Hello--0 ,。
返回-----result----Hello--1 。
返回-----result----Hello--2 ,。
返回-----result----Hello--4 ,。
返回-----result----Hello--3 。
服務(wù)端的日志：
[GENERIC] [11-10-6 下午9:49] [RemoteServerClientConnected(akka.remote.netty.NettyRemoteSupport@5a4fdf11,Some(/10.68.15.113:53462))]
Master接收到Work消息：Hello--1
Future call over
Master接收到Work消息：Hello--2
Future call over
Worker 收到消息----Hello--1
[DEBUG] [11-10-6 下午9:49] [akka:event-driven:dispatcher:global-13] [HelloWorker] started
[DEBUG] [11-10-6 下午9:49] [akka:event-driven:dispatcher:global-13] [HelloWorker] started
Worker 收到消息----Hello--2
Master接收到Work消息：Hello--0
Future call over
Master接收到Work消息：Hello--3
Worker 收到消息----Hello--0
[DEBUG] [11-10-6 下午9:49] [akka:event-driven:dispatcher:global-7] [HelloWorker] started
Future call over
Master接收到Work消息：Hello--4
[DEBUG] [11-10-6 下午9:49] [akka:event-driven:dispatcher:global-7] [HelloWorker] started
Worker 收到消息----Hello--3
Future call over
Worker 收到消息----Hello--4
[DEBUG] [11-10-6 下午9:49] [akka:event-driven:dispatcher:global-7] [HelloWorker] started
Worker 將消息Hello--1處理完成
Worker 將消息Hello--2處理完成
Worker Return----result----Hello--2 ,。
Worker Return----result----Hello--1 ,。
[DEBUG] [11-10-6 下午9:49] [akka:event-driven:dispatcher:global-13] [HelloWorker] stopping
Worker 將消息Hello--0處理完成
[DEBUG] [11-10-6 下午9:49] [akka:event-driven:dispatcher:global-3] [HelloWorker] stopping
Worker Return----result----Hello--0 。
[DEBUG] [11-10-6 下午9:49] [akka:event-driven:dispatcher:global-23] [HelloWorker] stopping
Worker 將消息Hello--4處理完成
Worker 將消息Hello--3處理完成
Worker Return----result----Hello--4 ,。
Worker Return----result----Hello--3 ,。
[DEBUG] [11-10-6 下午9:49] [akka:event-driven:dispatcher:global-11] [HelloWorker] stopping
[DEBUG] [11-10-6 下午9:49] [akka:event-driven:dispatcher:global-10] [HelloWorker] stopping
可以從服務(wù)端日志看到，Master接收到Work消息后onReceive就結(jié)束了（函數(shù)最后打印Future call over）,，一連接收了5個(gè)消息,，然后Worker才收到消息并處理。最后消息處理完成好后f1的call才收到Worker Return的消息,。
這里使用Future實(shí)現(xiàn)并發(fā),。
如果不使用Future：
def result = worker.sendRequestReply(o) //將消息發(fā)給worker actor
println "Worker Return----" + result
getContext().replyUnsafe(result) // 將worker返回的消息回復(fù)給客戶端
這就成了同步處理（第一個(gè)消息處理完后才接收并處理第二個(gè)消息）。
如果在Future后調(diào)用了f1.await()或f1.get(),，也成同步的了,，因?yàn)閍wait將等待worker返回后再繼續(xù)往下執(zhí)行。
Future f1 = Futures.future(new Callable() {
Object call() {
def result = worker.sendRequestReply(o) //將消息發(fā)給worker actor
worker.stop()
println "Worker Return----" + result
clientChannel.sendOneWay(result)
return result
}
})
println "Future call over" + f1.get()
服務(wù)器日志如下：
[GENERIC] [11-10-6 下午10:06] [RemoteServerStarted(akka.remote.netty.NettyRemoteSupport@7e566633)]
[DEBUG] [11-10-6 下午10:06] [main] [HelloMaster] started
[GENERIC] [11-10-6 下午10:07] [RemoteServerClientConnected(akka.remote.netty.NettyRemoteSupport@7e566633,Some(/10.68.15.113:53571))]
Master接收到Work消息：Hello--0
[DEBUG] [11-10-6 下午10:07] [akka:event-driven:dispatcher:global-3] [HelloWorker] started
Worker 收到消息----Hello--0
Worker 將消息Hello--0處理完成
[DEBUG] [11-10-6 下午10:07] [akka:event-driven:dispatcher:global-5] [HelloWorker] stopping
Worker Return----result----Hello--0 ,。
Future call overresult----Hello--0 ,。
Master接收到Work消息：Hello--2
Worker 收到消息----Hello--2
[DEBUG] [11-10-6 下午10:07] [akka:event-driven:dispatcher:global-3] [HelloWorker] started
Worker 將消息Hello--2處理完成
Worker Return----result----Hello--2 。
Future call overresult----Hello--2 ,。
Master接收到Work消息：Hello--3
[DEBUG] [11-10-6 下午10:07] [akka:event-driven:dispatcher:global-10] [HelloWorker] stopping
[DEBUG] [11-10-6 下午10:07] [akka:event-driven:dispatcher:global-3] [HelloWorker] started
Worker 收到消息----Hello--3
Worker 將消息Hello--3處理完成
Worker Return----result----Hello--3 ,。
Future call overresult----Hello--3 。
Master接收到Work消息：Hello--4
[DEBUG] [11-10-6 下午10:08] [akka:event-driven:dispatcher:global-14] [HelloWorker] stopping
[DEBUG] [11-10-6 下午10:08] [akka:event-driven:dispatcher:global-3] [HelloWorker] started
Worker 收到消息----Hello--4
Worker 將消息Hello--4處理完成
Worker Return----result----Hello--4 ,。
Future call overresult----Hello--4 ,。
Master接收到Work消息：Hello--1
[DEBUG] [11-10-6 下午10:08] [akka:event-driven:dispatcher:global-18] [HelloWorker] stopping
[DEBUG] [11-10-6 下午10:08] [akka:event-driven:dispatcher:global-3] [HelloWorker] started
Worker 收到消息----Hello--1
Worker 將消息Hello--1處理完成
Worker Return----result----Hello--1 ,。
Future call overresult----Hello--1 。
[DEBUG] [11-10-6 下午10:08] [akka:event-driven:dispatcher:global-21] [HelloWorker] stopping
Master接收到Work消息：Hello--6
[DEBUG] [11-10-6 下午10:08] [akka:event-driven:dispatcher:global-24] [HelloWorker] started
Worker 收到消息----Hello--6
Worker 將消息Hello--6處理完成
Worker Return----result----Hello--6 ,。
Future call overresult----Hello--6 ,。
Master接收到Work消息：Hello--5
[DEBUG] [11-10-6 下午10:09] [akka:event-driven:dispatcher:global-26] [HelloWorker] stopping
Worker 收到消息----Hello--5
[DEBUG] [11-10-6 下午10:09] [akka:event-driven:dispatcher:global-24] [HelloWorker] started
需要注意的是，Akka默認(rèn)使用環(huán)境變量%AKKA_HOME%/config/akka.conf配置,，默認(rèn)配置是client的read-timeout = 10(客戶端連接10秒后將自動(dòng)斷開,，這時(shí)服務(wù)端再給客戶端發(fā)消息就發(fā)布了了。報(bào)RemoteServerWriteFailed異常),，可以將值設(shè)為0,，將一直連著不斷開。
actor的timeout默認(rèn)為5秒,，也太短了,，延長(zhǎng)(不能設(shè)為0，0為總是超時(shí)).
7.參考文檔
1)
akka官網(wǎng) http:///
2）akka簡(jiǎn)介 http://inaunix.NET/uid-25885064-id-3400549.html
3）Scala語(yǔ)言：集成面向?qū)ο蠛秃瘮?shù)式編程的特性
4）actors模型 http://janeky./blog/1504125
5）示例 http://www./Program/java/2012/03/29/67015.shtml
6）akka資料 http://www./tags/10531
|
