介紹

這是一個 Java、Maven、Akka 項目，演示了如何設(shè)置一個基本的 Akka 集群。

這個項目是一系列項目中的一個，它從一個簡單的 Akka Cluster 項目開始，逐步構(gòu)建到事件溯源和命令查詢職責分離的示例。

該項目系列由以下 GitHub 存儲庫組成：

akka-java-cluster (這個項目)

akka-java-集群感知

akka-java-cluster-singleton

akka-java-集群分片

akka-java-cluster-persistence

akka-java-cluster-persistence-query

每個項目都可以獨立于其他項目進行克隆、構(gòu)建和運行。

關(guān)于 Akka 聚類

根據(jù)Akka 文檔，“ Akka Cluster 提供了一種容錯分散的基于點對點的集群成員服務(wù)，沒有單點故障或單點瓶頸。它使用 gossip 協(xié)議和自動故障檢測器來做到這一點。

Akka 集群允許構(gòu)建分布式應(yīng)用程序，其中一個應(yīng)用程序或服務(wù)跨越多個節(jié)點。"

Akka 文檔中的上述段落包含了許多最初可能難以理解的概念?？紤]一些僅在兩句話中就被拋棄的術(shù)語，例如“容錯”、“去中心化”、“點對點”和“無單點故障”。最后一句幾乎是隨便說的“一個應(yīng)用程序或服務(wù)跨越多個節(jié)點”。等待; 什么?應(yīng)用程序或服務(wù)如何跨越多個節(jié)點?

答案是 Akka 提供了一個抽象層，該層由參與者在參與者系統(tǒng)中相互交互組成。Akka 是一個演員模型的實現(xiàn)。演員模型” (維基百科)將“演員”視為并發(fā)計算的通用原語。為了響應(yīng)它接收到的消息，演員可以：做出本地決策，創(chuàng)建更多演員，發(fā)送更多消息，并確定如何響應(yīng)收到的下一條消息。參與者可以修改自己的私有狀態(tài)，但只能通過消息相互影響(避免需要任何鎖)。

Akka Actor 通過異步消息相互通信。Akka Actor 系統(tǒng)在 Java 虛擬機上運行，??并且使用 Akka 集群，單個 Actor 系統(tǒng)可以在邏輯上跨越多個網(wǎng)絡(luò) JVM。這個網(wǎng)絡(luò)化的參與者系統(tǒng)抽象層使參與者可以跨節(jié)點集群透明地與每個參與者進行通信。一種思考方式是，從演員的角度來看，他們生活在一個演員系統(tǒng)中，演員系統(tǒng)在一個或多個節(jié)點上運行的事實在很大程度上隱藏在抽象層中。

ClusterListenerActor 演員

Akka Actor 是用 Java 或 Scala 實現(xiàn)的。您可以將參與者創(chuàng)建為 Java 或 Scala 類。有兩種實現(xiàn)actor的方法，無類型和有類型。這個 Akka Java 集群示例項目系列中使用了無類型的 actor。

對于那些有興趣深入了解 Actor 如何工作以及如何實現(xiàn)的細節(jié)的人來說，關(guān)于Actors的 Akka 文檔部分 是一個很好的起點。

我們將要查看的第一個 Actor 名為 ClusterListenerActor。該參與者設(shè)置為接收有關(guān)集群事件的消息。當節(jié)點加入和離開集群時，此參與者會收到有關(guān)這些事件的消息。然后將這些接收到的消息寫入記錄器。

ClusterListenerActor 提供了集群活動的簡單視圖。以下是日志輸出的示例：

03:20:29.569 INFO  cluster-akka.actor.default-dispatcher-4 akka.tcp://cluster@127.0.0.1:2551/user/clusterListener - MemberUp(Member(address = akka.tcp://cluster@127.0.0.1:2553, status = Up)) sent to Member(address = akka.tcp://cluster@127.0.0.1:2551, status = Up)
03:20:29.570 INFO  cluster-akka.actor.default-dispatcher-4 akka.tcp://cluster@127.0.0.1:2551/user/clusterListener - 1 (LEADER) (OLDEST) Member(address = akka.tcp://cluster@127.0.0.1:2551, status = Up)
03:20:29.570 INFO  cluster-akka.actor.default-dispatcher-4 akka.tcp://cluster@127.0.0.1:2551/user/clusterListener - 2 Member(address = akka.tcp://cluster@127.0.0.1:2552, status = Up)
03:20:29.570 INFO  cluster-akka.actor.default-dispatcher-4 akka.tcp://cluster@127.0.0.1:2551/user/clusterListener - 3 Member(address = akka.tcp://cluster@127.0.0.1:2553, status = Joining)

讓我們從完整的 ClusterListenerActor 源文件開始。請注意，此 actor 是作為擴展基于 Akka 的類的單個 Java 類實現(xiàn)的。

package cluster;
import akka.actor.AbstractLoggingActor;
import akka.actor.Cancellable;
import akka.actor.Props;
import akka.cluster.Cluster;
import akka.cluster.ClusterEvent;
import akka.cluster.ClusterEvent.CurrentClusterState;
import akka.cluster.Member;
import java.time.Duration;
import java.util.Optional;
import java.util.function.Consumer;
import java.util.stream.StreamSupport;
class ClusterListenerActor extends AbstractLoggingActor {
    private final Cluster cluster = Cluster.get(context().system());
    private Cancellable showClusterStateCancelable;
    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(ShowClusterState.class, this::showClusterState)
                .matchAny(this::logClusterEvent)
                .build();
    }
    private void showClusterState(ShowClusterState showClusterState) {
        log().info("{} sent to {}", showClusterState, cluster.selfMember());
        logClusterMembers(cluster.state());
        showClusterStateCancelable = null;
    }
    private void logClusterEvent(Object clusterEventMessage) {
        log().info("{} sent to {}", clusterEventMessage, cluster.selfMember());
        logClusterMembers();
    }
    @Override
    public void preStart() {
        log().debug("Start");
        cluster.subscribe(self(), ClusterEvent.initialStateAsEvents(),
                ClusterEvent.ClusterDomainEvent.class);
    }
    @Override
    public void postStop() {
        log().debug("Stop");
        cluster.unsubscribe(self());
    }
    static Props props() {
        return Props.create(ClusterListenerActor.class);
    }
    private void logClusterMembers() {
        logClusterMembers(cluster.state());
        if (showClusterStateCancelable == null) {
            showClusterStateCancelable = context().system().scheduler().scheduleOnce(
                    Duration.ofSeconds(15),
                    self(),
                    new ShowClusterState(),
                    context().system().dispatcher(),
                    null);
        }
    }
    private void logClusterMembers(CurrentClusterState currentClusterState) {
        Optional<Member> old = StreamSupport.stream(currentClusterState.getMembers().spliterator(), false)
                .reduce((older, member) -> older.isOlderThan(member) ? older : member);
        Member oldest = old.orElse(cluster.selfMember());
        StreamSupport.stream(currentClusterState.getMembers().spliterator(), false)
                .forEach(new Consumer<Member>() {
                    int m = 0;
                    @Override
                    public void accept(Member member) {
                        log().info("{} {}{}{}", ++m, leader(member), oldest(member), member);
                    }
                    private String leader(Member member) {
                        return member.address().equals(currentClusterState.getLeader()) ? "(LEADER) " : "";
                    }
                    private String oldest(Member member) {
                        return oldest.equals(member) ? "(OLDEST) " : "";
                    }
                });
    }
    private static class ShowClusterState {
        @Override
        public String toString() {
            return ShowClusterState.class.getSimpleName();
        }
    }
}

這個類是一個簡單的actor實現(xiàn)的例子。然而，這個actor的獨特之處在于它訂閱了Akka系統(tǒng)來接收集群事件消息。

@Override
public void preStart() {
    log().debug("Start");
    cluster.subscribe(self(), ClusterEvent.initialStateAsEvents(),
            ClusterEvent.ClusterDomainEvent.class);
}

Actor 被設(shè)置為接收集群事件消息。當這些消息到達時，參與者調(diào)用編寫的方法來記錄事件并記錄集群的當前狀態(tài)。

@Override
public Receive createReceive() {
    return receiveBuilder()
            .match(ShowClusterState.class, this::showClusterState)
            .matchAny(this::logClusterEvent)
            .build();
}

隨著集群中的每個節(jié)點啟動，ClusterListenerActor 的一個實例也會啟動。然后，參與者記錄每個節(jié)點中發(fā)生的集群事件。您可以再次從每個節(jié)點的角度檢查來自每個集群節(jié)點的日志，以查看集群事件并查看集群節(jié)點的狀態(tài)。

這個怎么運作

在這個項目中，我們將從一個基于 Akka、Java 和 Maven 的示例的基本模板開始，其中包含運行 Akka 集群的代碼和配置。Maven POM 文件使用兩個插件，一個用于使用mvn:exec命令運行代碼，另一個插件構(gòu)建一個自包含 JAR 文件用于使用java -jar命令運行代碼。

當項目代碼被執(zhí)行時，動作在Runner類main方法中開始。

public static void main(String[] args) {
    if (args.length == 0) {
        startupClusterNodes(Arrays.asList("2551", "2552", "0"));
    } else {
        startupClusterNodes(Arrays.asList(args));
    }
}

該main方法調(diào)用startupClusterNodes傳遞給它一個端口列表的方法。如果未提供參數(shù)，則使用默認的三個端口集。

private static void startupClusterNodes(List<String> ports) {
    System.out.printf("Start cluster on port(s) %s%n", ports);
    ports.forEach(port -> {
        ActorSystem actorSystem = ActorSystem.create("cluster", setupClusterNodeConfig(port));
        AkkaManagement.get(actorSystem).start();
        actorSystem.actorOf(ClusterListenerActor.props(), "clusterListener");
        addCoordinatedShutdownTask(actorSystem, CoordinatedShutdown.PhaseClusterShutdown());
        actorSystem.log().info("Akka node {}", actorSystem.provider().getDefaultAddress());
    });
}

這些startupClusterNodes方法循環(huán)通過端口列表。為每個端口創(chuàng)建一個參與者系統(tǒng)。

ActorSystem actorSystem = ActorSystem.create("cluster", setupClusterNodeConfig(port));

創(chuàng)建一個演員系統(tǒng)時會發(fā)生很多事情。許多決定如何運行actor系統(tǒng)的細節(jié)是通過配置設(shè)置定義的。這個項目包括一個application.conf配置文件，它位于src/main/resources目錄中。最關(guān)鍵的配置設(shè)置之一定義了參與者系統(tǒng)主機和端口。當參與者系統(tǒng)在集群中運行時，配置還定義了每個節(jié)點將如何定位和加入集群。在這個項目中，節(jié)點使用所謂的種子節(jié)點加入集群。

cluster {
  seed-nodes = [
    "akka.tcp://cluster@127.0.0.1:2551",
    "akka.tcp://cluster@127.0.0.1:2552"]
}

讓我們來看看這個項目的集群啟動場景。在此示例中，一個 JVM 啟動時沒有運行時參數(shù)。當不帶參數(shù)調(diào)用Runner類main方法時，默認是在端口 2551、2552 和端口 0 上創(chuàng)建三個參與者系統(tǒng)(零端口會導致隨機選擇非零端口號)。

由于每個參與者系統(tǒng)都是在特定端口上創(chuàng)建的，因此它會查看種子節(jié)點配置設(shè)置。如果參與者系統(tǒng)的端口是種子節(jié)點之一，它知道它將與其他種子節(jié)點聯(lián)系以形成集群。如果參與者系統(tǒng)的端口不是種子節(jié)點之一，它將嘗試聯(lián)系種子節(jié)點之一。非種子節(jié)點需要向其中一個種子節(jié)點宣布自己并要求加入集群。

下面是使用默認端口 2551、2552 和 0 的示例啟動場景。在端口 2551 上創(chuàng)建了一個參與者系統(tǒng);查看配置它知道它是一個種子節(jié)點。端口 2551 上的種子節(jié)點參與者系統(tǒng)嘗試聯(lián)系端口 2552 上的參與者系統(tǒng)，即另一個種子節(jié)點。當創(chuàng)建端口 2552 上的參與者系統(tǒng)時，它會經(jīng)歷相同的過程，在這種情況下，2552 會嘗試與 2551 聯(lián)系并加入。當在隨機端口(例如端口 24242)上創(chuàng)建第三個參與者系統(tǒng)時，它會從配置中知道它不是種子節(jié)點，在這種情況下，它會嘗試與種子參與者系統(tǒng)之一進行通信，宣布自己并加入集群。

您可能已經(jīng)注意到，在上面的示例中，三個參與者系統(tǒng)是在單個 JVM 中創(chuàng)建的。雖然每個 JVM 運行多個參與者系統(tǒng)是可以接受的，但更常見的情況是每個 JVM 運行一個參與者系統(tǒng)。

讓我們看一個稍微現(xiàn)實一點的例子。使用提供的akka腳本啟動一個三節(jié)點集群。

./akka cluster start 3

每個節(jié)點都在單獨的 JVM 中運行。在這里，我們有三個參與者系統(tǒng)，它們在三個 JVM 中獨立啟動。這三個參與者系統(tǒng)遵循與之前相同的啟動場景，結(jié)果它們形成了一個集群。

當然，最常見的場景是每個參與者系統(tǒng)都是在不同的 JVM 中創(chuàng)建的，每個 JVM 都運行在不同的服務(wù)器、虛擬服務(wù)器或容器上。同樣，相同的啟動過程發(fā)生在各個參與者系統(tǒng)通過網(wǎng)絡(luò)找到彼此并形成集群的地方??。

讓我們回到創(chuàng)建actor系統(tǒng)的那一行代碼。

ActorSystem actorSystem = ActorSystem.create("cluster", setupClusterNodeConfig(port));

從這個簡短的描述中，您可以看到在actor系統(tǒng)抽象層中發(fā)生了很多事情，而這個啟動過程的總結(jié)只是冰山一角，這是抽象層應(yīng)該做的，它們隱藏了復雜性。

一旦多個參與者系統(tǒng)加入一個集群，從在這個虛擬參與者系統(tǒng)中運行的參與者的角度來看，它們會形成一個單一的虛擬參與者系統(tǒng)。當然，單個參與者實例物理上駐留在特定 JVM 內(nèi)的特定集群節(jié)點中，但在接收和發(fā)送參與者消息時，節(jié)點邊界是透明的并且?guī)缀跸Я?。正是這種透明性是構(gòu)建“一個應(yīng)用程序或服務(wù)跨越多個節(jié)點”的基礎(chǔ)。

此外，通過添加更多節(jié)點來擴展集群的靈活性是消除單點瓶頸的機制。當集群中的現(xiàn)有節(jié)點無法處理當前負載時，可以添加更多節(jié)點來擴展容量。失敗也是如此。一個或多個節(jié)點的丟失并不意味著整個集群出現(xiàn)故障?？梢蕴鎿Q故障節(jié)點，并且可以將在故障節(jié)點上運行的參與者重新定位到其他節(jié)點。

希望本概述能夠闡明 Akka 如何提供“無單點故障或單點瓶頸”以及“ Akka 集群如何允許構(gòu)建分布式應(yīng)用程序，其中一個應(yīng)用程序或服務(wù)跨越多個節(jié)點。 ”

安裝

git clone https://github.com/mckeeh3/akka-java-cluster.git
cd akka-java-cluster
mvn clean package

Maven 命令構(gòu)建項目并創(chuàng)建一個自包含的可運行 JAR。

運行集群(Mac、Linux)

該項目包含一組腳本，可用于啟動和停止單個集群節(jié)點或啟動和停止節(jié)點集群。

提供主腳本./akka以運行節(jié)點集群或啟動和停止單個節(jié)點。用于./akka node start [1-9] | stop啟動和停止單個節(jié)點以及./akka cluster start [1-9] | stop啟動和停止節(jié)點集群。clusterand start 選項將node在端口 2551 到 2559 上啟動 Akka 節(jié)點。stdinand輸出都使用文件命名約定stderr發(fā)送到目錄中的文件。/tmp/tmp/-N.log

在端口 2551 上啟動節(jié)點 1，在端口 2552 上啟動節(jié)點 2。

./akka node start 1
./akka node start 2

在端口 2553 上停止節(jié)點 3。

./akka node stop 3

在端口 2551、2552、2553 和 2554 上啟動一個由四個節(jié)點組成的集群。

./akka cluster start 4

停止所有當前正在運行的集群節(jié)點。

./akka cluster stop

您可以使用該./akka cluster start [1-9]腳本啟動多個節(jié)點，然后使用./akka node start [1-9]and./akka node stop [1-9] 啟動和停止單個節(jié)點。

使用該./akka node tail [1-9]命令tail -f創(chuàng)建節(jié)點 1 到 9 的日志文件。

該命令使用Akka 管理 擴展 Cluster Http Management./akka cluster status以 JSON 格式顯示當前正在運行的集群的狀態(tài) 。

運行集群(Windows，命令行)

以下 Maven 命令在端口 2551、2552 和 radmonly 選擇的端口上運行帶有 3 個 Akka 演員系統(tǒng)的signle JVM。

mvn exec:java

使用 CTRL-C 停止。

要在特定端口上運行，請使用以下-D選項傳入命令行參數(shù)。

mvn exec:java -Dexec.args="2551"

默認無參數(shù)等價于以下內(nèi)容。

mvn exec:java -Dexec.args="2551 2552 0"

運行測試的一種常見方法是在多個命令窗口中啟動單個 JVM。這模擬了運行多節(jié)點 Akka 集群。例如，在 4 個命令窗口中運行以下 4 個命令。

mvn exec:java -Dexec.args="2551" > /tmp/$(basename $PWD)-1.log

mvn exec:java -Dexec.args="2552" > /tmp/$(basename $PWD)-2.log

mvn exec:java -Dexec.args="0" > /tmp/$(basename $PWD)-3.log

mvn exec:java -Dexec.args="0" > /tmp/$(basename $PWD)-4.log

這將運行一個 4 節(jié)點 Akka 集群，在端口 2551 和 2552 上啟動 2 個節(jié)點，這些節(jié)點是配置的集群種子節(jié)點和application.conf文件。以及隨機選擇的端口號上的 2 個節(jié)點。可選重定向> /tmp/$(basename $PWD)-4.log是基于項目目錄名稱將日志輸出推送到文件名的示例。

為方便起見，在 Linux 命令 shell 中定義以下別名。

alias p1='cd ~/akka-java/akka-java-cluster'
alias p2='cd ~/akka-java/akka-java-cluster-aware'
alias p3='cd ~/akka-java/akka-java-cluster-singleton'
alias p4='cd ~/akka-java/akka-java-cluster-sharding'
alias p5='cd ~/akka-java/akka-java-cluster-persistence'
alias p6='cd ~/akka-java/akka-java-cluster-persistence-query'
alias m1='clear ; mvn exec:java -Dexec.args="2551" > /tmp/$(basename $PWD)-1.log'
alias m2='clear ; mvn exec:java -Dexec.args="2552" > /tmp/$(basename $PWD)-2.log'
alias m3='clear ; mvn exec:java -Dexec.args="0" > /tmp/$(basename $PWD)-3.log'
alias m4='clear ; mvn exec:java -Dexec.args="0" > /tmp/$(basename $PWD)-4.log'

p1-6 別名命令是 cd 到六個項目目錄之一的快捷方式。m1-4 別名命令使用適當?shù)亩丝趩雍?Akka 節(jié)點。Stdout 也被重定向到 /tmp 目錄。