您當(dāng)前位置：首頁 > 互聯(lián)網(wǎng) > 使用Akka構(gòu)建集群（一）

使用Akka構(gòu)建集群（一）

來源：程序員人生發(fā)布時間：2017-03-17 09:57:34 閱讀次數(shù)：7628次

概述

Akka提供的非常吸引人的特性之1就是輕松構(gòu)建自定義集群，這也是我要選擇Akka的最基本緣由之1。如果你不想敲太多代碼，也能夠通過簡單的配置構(gòu)建1個非常簡單的集群。本文為說明Akka集群構(gòu)建的學(xué)習(xí)本錢低廉，以Akka官網(wǎng)的例子代碼動身，進(jìn)行簡單改造后與Spring集成，有關(guān)Spring集成的信息你可以選擇瀏覽《Spring與Akka的集成》1文。本文所講述的是1款10分簡便的集群監(jiān)聽器，它通過定閱集群成員的消息，對全部集群的成員進(jìn)行管理（管理的方式只是打印1行日志）。

Akka集群規(guī)范

根據(jù)Akka官網(wǎng)的描寫——Akka集群特性提供了容錯的、去中心化的、基于集群成員關(guān)系點對點的，不存在單點問題、單點瓶頸的服務(wù)。其實現(xiàn)原理為閑談協(xié)議和失敗檢查。

集群概念

節(jié)點（node）：集群中的邏輯成員。允許1臺物理機(jī)上有多個節(jié)點。由元組hostname:port:uid唯1肯定。
集群（cluster）：由成員關(guān)系服務(wù)構(gòu)建的1組節(jié)點。
領(lǐng)導(dǎo)（leader）：集群中唯1扮演領(lǐng)導(dǎo)角色的節(jié)點。
種子節(jié)點（seed node）：作為其他節(jié)點加入集群的連接點的節(jié)點。實際上，1個節(jié)點可以通過向集群中的任何1個節(jié)點發(fā)送Join（加入）命令加入集群。

節(jié)點狀態(tài)

這里以Akka官網(wǎng)提供的成員狀態(tài)狀態(tài)圖為例，如圖1所示。

圖1

圖1展現(xiàn)了狀態(tài)轉(zhuǎn)換的兩個因素：動作和狀態(tài)。

狀態(tài)

joining：節(jié)點正在加入集群時的狀態(tài)。
weekly up：配置了akka.cluster.allow-weakly-up-members=on時，啟用的狀態(tài)。
up：集群中節(jié)點的正常狀態(tài)。
leaving/exiting：優(yōu)雅的刪除節(jié)點時，節(jié)點的狀態(tài)。
down：標(biāo)記為已下線的狀態(tài)。
removed：墓碑狀態(tài)，表示已不再是集群的成員。

動作

join：加入集群。
leave：告知節(jié)點優(yōu)雅的離開集群。
down：標(biāo)記集群為已下線。

配置

本節(jié)將要展現(xiàn)構(gòu)建集群所需要的最基本的配置，幾近不會引入過量的開發(fā)本錢，1個集群就構(gòu)建完成了。application.conf文件的內(nèi)容以下：

akka {
  actor {
    provider = "akka.cluster.ClusterActorRefProvider"
  }
  remote {
    log-remote-lifecycle-events = off
    netty.tcp {
      hostname = "127.0.0.1"
      port = 2551
    }
  }

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

    #//#snippet
    # excluded from snippet
    auto-down-unreachable-after = 10s
    #//#snippet
    # auto downing is NOT safe for production deployments.
    # you may want to use it during development, read more about it in the docs.
    #
    # auto-down-unreachable-after = 10s
    
    # Disable legacy metrics in akka-cluster.
	metrics.enabled=off
  }

}

此配置文件與我在《使用Akka的遠(yuǎn)程調(diào)用》1文中的配置有很多不同：

provider不再是akka.remote.RemoteActorRefProvider，而是akka.cluster.ClusterActorRefProvider。這說明ActorRef將由akka.cluster.ClusterActorRefProvider提供；
增加了cluster配置；

cluster配置詳解

首先任何1個集群都需要種子節(jié)點，作為基本的加入集群的連接點。本例中以我本地的兩個節(jié)點（分別監(jiān)聽2551和2552端口）作為種子節(jié)點。不管配置了多少個種子節(jié)點，除在seed-nodes中配置的第1個種子節(jié)點需要率先啟動以外（否則其它種子節(jié)點沒法初始化并且其它節(jié)點也沒法加入），其余種子節(jié)點都是啟動順序無關(guān)的。第1個節(jié)點需要率先啟動的另外一個緣由是如果每一個節(jié)點都可以率先啟動，那末有可能造成1個集群出現(xiàn)幾個種子節(jié)點都啟動并且加入了自己的集群，此時全部集群實際上分裂為幾個集群，造成孤島。當(dāng)你啟動了超過2個以上的種子節(jié)點，那末第1個啟動的種子節(jié)點是可以關(guān)閉下線的。如果第1個種子節(jié)點重啟了，它將不會在自己創(chuàng)建集群而是向其它種子節(jié)點發(fā)送Join消息加入已存在的集群。

注意：除akka.remote.netty.tcp.port配置項指定的端口不同，所有加入集群節(jié)點的application.conf可以完全1樣。如果akka.remote.netty.tcp.port未指定，那末Akka會為你隨機(jī)選擇其他未占用的端口。

簡單集群監(jiān)聽器

我們創(chuàng)建1個簡單的集群監(jiān)聽器SimpleClusterListener（實際上是1個Actor，由于繼承了UntypedActor），它向集群定閱MemberEvent（成員事件）和UnreachableMember（不可達(dá)成員）兩種消息，來對集群成員進(jìn)行管理（打印），其實現(xiàn)見代碼清單1所示。

代碼清單1

@Named("SimpleClusterListener")
@Scope("prototype")
public class SimpleClusterListener extends UntypedActor {
	LoggingAdapter log = Logging.getLogger(getContext().system(), this);
	Cluster cluster = Cluster.get(getContext().system());

	// subscribe to cluster changes
	@Override
	public void preStart() {
		// #subscribe
		cluster.subscribe(getSelf(), ClusterEvent.initialStateAsEvents(), MemberEvent.class, UnreachableMember.class);
		// #subscribe
	}

	// re-subscribe when restart
	@Override
	public void postStop() {
		cluster.unsubscribe(getSelf());
	}

	@Override
	public void onReceive(Object message) {
		if (message instanceof MemberUp) {
			MemberUp mUp = (MemberUp) message;
			log.info("Member is Up: {}", mUp.member());

		} else if (message instanceof UnreachableMember) {
			UnreachableMember mUnreachable = (UnreachableMember) message;
			log.info("Member detected as unreachable: {}", mUnreachable.member());

		} else if (message instanceof MemberRemoved) {
			MemberRemoved mRemoved = (MemberRemoved) message;
			log.info("Member is Removed: {}", mRemoved.member());

		} else if (message instanceof MemberEvent) {
			// ignore

		} else {
			unhandled(message);
		}

	}
}

運(yùn)行展現(xiàn)

初始化的代碼以下：

		logger.info("Start simpleClusterListener");
		final ActorRef simpleClusterListener = actorSystem.actorOf(springExt.props("SimpleClusterListener"), "simpleClusterListener");
		actorMap.put("simpleClusterListener", simpleClusterListener);
		logger.info("Started simpleClusterListener");

我們首先啟動第1個種子節(jié)點，配置跟第1小節(jié)完全1致。我們視察SimpleClusterListener的日志輸出以下圖所示。

我們再啟動第2個種子節(jié)點，其配置的akka.remote.netty.tcp.port為2552，我們視察SimpleClusterListener的日志輸出以下圖所示。

我們再啟動1個非種子節(jié)點，沒有為其指定akka.remote.netty.tcp.port，我們視察SimpleClusterListener的日志輸出以下圖所示。

可以看到新加入的節(jié)點信息被SimpleClusterListener打印出來了，仔細(xì)的同學(xué)可能發(fā)現(xiàn)了1些Akka集群中各個節(jié)點的狀態(tài)遷移信息，第1個種子節(jié)點正在加入本身創(chuàng)建的集群時的狀態(tài)時JOINING，由于第1個種子節(jié)點將自己率先選舉為Leader，因此它還將自己的狀態(tài)改變成Up。后面它還將第2個種子節(jié)點和第3個節(jié)點從JOINING轉(zhuǎn)換到Up狀態(tài)。

我們停止第3個加入的節(jié)點，我們視察SimpleClusterListener的日志輸出以下圖所示。