分佈式應用框架Akka快速入門

時間 2019-12-01

原文原文鏈接

本文結合網上一些資料，對他們進行整理，摘選和翻譯而成，對Akka進行簡要的說明。引用資料在最後列出。html

1.什麼是Akka

Akka 是一個用 Scala 編寫的庫，用於簡化編寫容錯的、高可伸縮性的 Java 和 Scala 的 Actor 模型應用。java

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是一個開發庫和運行環境，能夠用於構建高併發、分佈式、可容錯、事件驅動的基於JVM的應用。使構建高併發的分佈式應用更加容易。ruby

Akka能夠以兩種不一樣的方式來使用服務器

以庫的形式：在web應用中使用，放到 WEB-INF/lib 中或者做爲一個普通的Jar包放進classpath。
以微內核的形式：你能夠將應用放進一個獨立的內核。

2.Akka的五大特性

1）易於構建並行和分佈式應用（Simple Concurrency & Distribution）
Akka在設計時採用了異步通信和分佈式架構，並對上層進行抽象，如Actors、Futures ，STM等。

2）可靠性（Resilient by Design）多線程

系統具有自愈能力，在本地/遠程都有監護。架構

3）高性能（High Performance）併發

在單機中每秒可發送50000000個消息。內存佔用小，1GB內存中可保存2500000個actors。

4）彈性，無中心（Elastic — Decentralized）

自適應的負責均衡，路由，分區，配置

5）可擴展（Extensible）

可使用Akka 擴展包進行擴展。

3.什麼場景下特別適合使用Akka？

咱們看到Akka被成功運用在衆多行業的衆多大企業，從投資業到商業銀行、從零售業到社會媒體、仿真、遊戲和賭博、汽車和交通系統、數據分析等等等等。任何須要高吞吐率和低延遲的系統都是使用Akka的候選。

Actor使你可以進行服務失敗管理（監管者），負載管理（緩和策略、超時和隔離），水平和垂直方向上的可擴展性（增長cpu核數和/或增長更多的機器）管理。

下面的連接中有一些Akka用戶關於他們如何使用Akka的描述：http://stackoverflow.com/questions/4493001/good-use-case-for-akka

全部以上這些都在這個Apache2許可的開源軟件中。

如下是Akka被部署到生產環境中的領域
事務處理 (在線遊戲，金融/銀行業，貿易，統計，賭博，社會媒體，電信):垂直擴展，水平擴展，容錯/高可用性

服務後端 (任何行業，任何應用):提供REST, SOAP, Cometd, WebSockets 等服務做爲消息總線/集成層垂直擴展，水平擴展，容錯/高可用性

併發/並行 (任何應用):運行正確，方便使用，只須要將jar包添加到現有的JVM項目中（使用Scala，java, Groovy或jruby)

仿真:主/從，計算網格，MaReduce等等.

批處理 (任何行業):Camel集成來鏈接批處理數據源 Actor來分治地批處理工做負載

通訊Hub (電信, Web媒體, 手機媒體):垂直擴展，水平擴展，容錯/高可用性

遊戲與賭博 (MOM, 在線遊戲, 賭博):垂直擴展，水平擴展，容錯/高可用性

商業智能/數據挖掘/通用數據處理:垂直擴展，水平擴展，容錯/高可用性

復瑣事件流處理:垂直擴展，水平擴展，容錯/高可用性

4. Scala語言

Scala是一種多範式的編程語言，設計初衷是要集成面向對象編程和函數式編程的各類特性。

百度百科，Scala語言介紹：

http://baike.baidu.com/link?url=86mpCDT8PbqP29vR-ZDHQeMt1grr--g42DtVMkjvru4g57_TATYquSDnjLOJl-WovbPuywyikI7q1I0ZvWjuca

百度文庫，Scala編程入門：

http://wenku.baidu.com/view/27cbf218964bcf84b9d57b83.html

5.Actors模型

Actor模型並不是什麼新鮮事物，它由Carl Hewitt於上世紀70年代早期提出，目的是爲了解決分佈式編程中一系列的編程問題。其特色以下：

系統中的全部事物均可以扮演一個Actor

Actor之間徹底獨立

在收到消息時Actor所採起的全部動做都是並行的，在一個方法中的動做沒有明確的順序

Actor由標識和當前行爲描述

Actor可能被分紅原始（primitive）和非原始（non primitive）類別

不少開發語言都提供了原生的Actor模型。例如erlang,scala等

Actor，能夠看做是一個個獨立的實體，他們之間是毫無關聯的。可是，他們能夠經過消息來通訊。

一個Actor收到其餘Actor的信息後，它能夠根據須要做出各類相應。消息的類型能夠是任意的，消息的內容也能夠是任意的。這點有點像webservice了。只提供接口服務，你沒必要了解我是如何實現的。

一個Actor如何處理多個Actor的請求呢？它先創建一個消息隊列，每次收到消息後，就放入隊列，而它每次也從隊列中取出消息體來處理。一般咱們都使得這個過程是循環的。讓Actor能夠時刻處理髮送來的消息。

6.示例（http://www.th7.cn/Program/java/2012/03/29/67015.shtml）

應用場景：服務端要處理大量的客戶端的請求，而且處理請求耗費較長的時間。這時就須要使用併發處理。多線程是一種方法，這裏使用Akka框架處理併發。（如下代碼在Groovy1.7.五、akka-actors-1.2下運行成功）

這裏有三個角色：Client、Master、Worker
Client傻乎乎地發同步請求給Master，一直等到結果返回客戶端才離開。
Master接收客戶端發來的請求，而後將請求交給Worker處理，處理完成以後將結果返回給Client。
Worker負責具體的業務處理，它耗費的事件比較長。

因此這裏的關鍵在於Master，若是Master線性地「接收請求――調用Worker處理獲得返回結果――將結果返回」，這樣的系統必將歇菜。
使用Akka能夠方便地將它變成並行地。

先看看Client，模擬同時多個客戶端給Master發請求
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 "請求-----${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()        //同時啓動5個客戶端請求Master
        }
    }
}

真正幹活的Worker：
import akka.actor.UntypedActor

class HelloWorker extends UntypedActor {    //Worker是一個Actor，須要實現onReceive方法
    @Override
    void onReceive(Object o) {
        println "Worker 收到消息----" + o
        if (o instanceof String) {
            String result = doWork(o)        //調用真實的處理方法
            getContext().replyUnsafe(result)//將結果返回給Master
        }
    }
    //Worker處理其實很簡單，僅僅將參數字符串改造一下而已。只不過使其sleep了20秒，讓它變得「耗時較長」
    String doWork(String str) {
        Thread.sleep(1000 * 20)
        return "result----" + str + " 。"
    }
}

負責併發調度的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
        //啓動worker actor
        ActorRef worker = Actors.actorOf(new UntypedActorFactory() {
            public UntypedActor create() {
                return new HelloWorker();
            }
        }).start();

        //這裏實現真正的併發
        Future f1 = Futures.future(new Callable() {
            Object call() {
                def result = worker.sendRequestReply(o)            //將消息發給worker actor，讓Worker處理業務，同時獲得返回結果
                worker.stop()
                println "Worker Return----" + result
                clientChannel.sendOneWay(result)                //將結果返回給客戶端
                return result
            }
        })

println "Future call over"
}

    public static void main(String[] args) {    //啓動Master進程，綁定IP、端口和服務
        Actors.remote().start("10.68.15.113", 9999).register(
                "hello-service",
                Actors.actorOf(HelloMaster.class));
    }
}

看看客戶端的調用日誌
請求-----Hello--4
請求-----Hello--1
請求-----Hello--3
請求-----Hello--0
請求-----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 。

服務端的日誌：
[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

能夠從服務端日誌看到，Master接收到Work消息後onReceive就結束了（函數最後打印Future call over），一鏈接收了5個消息，而後Worker才收到消息並處理。最後消息處理完成好後f1的call才收到Worker Return的消息。
這裏使用Future實現併發。

若是不使用Future：
def result = worker.sendRequestReply(o) //將消息發給worker actor
println "Worker Return----" + result
getContext().replyUnsafe(result) // 將worker返回的消息回覆給客戶端
這就成了同步處理（第一個消息處理完後才接收並處理第二個消息）。

若是在Future後調用了f1.await()或f1.get()，也成同步的了，由於await將等待worker返回後再繼續往下執行。
Future f1 = Futures.future(new Callable() {
    Object call() {
        def result = worker.sendRequestReply(o)       //將消息發給worker actor
        worker.stop()
        println "Worker Return----" + result
        clientChannel.sendOneWay(result)
        return result
    }
})

println "Future call over" + f1.get()

服務器日誌以下：
[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默認使用環境變量%AKKA_HOME%/config/akka.conf配置，默認配置是client的read-timeout = 10(客戶端鏈接10秒後將自動斷開，這時服務端再給客戶端發消息就發佈了了。報RemoteServerWriteFailed異常)，能夠將值設爲0，將一直連着不斷開。
actor的timeout默認爲5秒，也過短了，延長(不能設爲0，0爲老是超時).