使用Akka构建集群（一）

概述

　　Akka提供的很是吸引人的特性之一就是轻松构建自定义集群，这也是我要选择Akka的最基本缘由之一。若是你不想敲太多代码，也能够经过简单的配置构建一个很是简单的集群。本文为说明Akka集群构建的学习成本低廉，以Akka官网的例子代码出发，进行简单改造后与Spring集成，有关Spring集成的信息你能够选择阅读《Spring与Akka的集成》一文。本文所讲述的是一款十分简便的集群监听器，它经过订阅集群成员的消息，对整个集群的成员进行管理（管理的方式只是打印一行日志）。

Akka集群规范

　　根据Akka官网的描述——Akka集群特性提供了容错的、去中心化的、基于集群成员关系点对点的，不存在单点问题、单点瓶颈的服务。其实现原理为闲聊协议和失败检查。

集群概念

• 节点（node）：集群中的逻辑成员。容许一台物理机上有多个节点。由元组hostname:port:uid惟一肯定。
• 集群（cluster）：由成员关系服务构建的一组节点。
• 领导（leader）：集群中惟一扮演领导角色的节点。
• 种子节点（seed node）：做为其余节点加入集群的链接点的节点。实际上，一个节点能够经过向集群中的任何一个节点发送Join（加入）命令加入集群。

节点状态

这里以Akka官网提供的成员状态状态图为例，如图1所示。

图1

图1展现了状态转换的两个因素：动做和状态。

状态

• joining：节点正在加入集群时的状态。
• weekly up：配置了akka.cluster.allow-weakly-up-members=on时，启用的状态。
• up：集群中节点的正常状态。
• leaving/exiting：优雅的删除节点时，节点的状态。
• down：标记为已下线的状态。
• removed：墓碑状态，表示已经再也不是集群的成员。

动做 

• join：加入集群。
• leave：告知节点优雅的离开集群。
• down：标记集群为已下线。

配置

　　本节将要展现构建集群所须要的最基本的配置，几乎不会引入过多的开发成本，一个集群就构建完成了。application.conf文件的内容以下：

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的远程调用》一文中的配置有不少不一样：

1. provider再也不是akka.remote.RemoteActorRefProvider，而是akka.cluster.ClusterActorRefProvider。这说明ActorRef将由akka.cluster.ClusterActorRefProvider提供；
2. 增长了cluster配置；

cluster配置详解

首先任何一个集群都须要种子节点，做为基本的加入集群的链接点。本例中以我本地的两个节点（分别监听2551和2552端口）做为种子节点。不管配置了多少个种子节点，除了在seed-nodes中配置的第一个种子节点须要率先启动以外（不然其它种子节点没法初始化而且其它节点也没法加入），其他种子节点都是启动顺序无关的。第一个节点须要率先启动的另外一个缘由是若是每一个节点均可以率先启动，那么有可能形成一个集群出现几个种子节点都启动而且加入了本身的集群，此时整个集群实际上分裂为几个集群，形成孤岛。当你启动了超过2个以上的种子节点，那么第一个启动的种子节点是能够关闭下线的。若是第一个种子节点重启了，它将不会在本身建立集群而是向其它种子节点发送Join消息加入已存在的集群。

注意：除了akka.remote.netty.tcp.port配置项指定的端口不一样，全部加入集群节点的application.conf能够彻底同样。若是akka.remote.netty.tcp.port未指定，那么Akka会为你随机选择其余未占用的端口。

简单集群监听器

　　咱们建立一个简单的集群监听器SimpleClusterListener（其实是一个Actor，由于继承了UntypedActor），它向集群订阅MemberEvent（成员事件）和UnreachableMember（不可达成员）两种消息，来对集群成员进行管理（打印），其实现见代码清单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);
		}
	}
}

运行展现

初始化的代码以下：

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

咱们首先启动第一个种子节点，配置跟第一小节彻底一致。咱们观察SimpleClusterListener的日志输出以下图所示。

咱们再启动第二个种子节点，其配置的akka.remote.netty.tcp.port为2552，咱们观察SimpleClusterListener的日志输出以下图所示。

咱们再启动一个非种子节点，没有为其指定akka.remote.netty.tcp.port，咱们观察SimpleClusterListener的日志输出以下图所示。

能够看到新加入的节点信息被SimpleClusterListener打印出来了，细心的同窗可能发现了一些Akka集群中各个节点的状态迁移信息，第一个种子节点正在加入自身建立的集群时的状态时JOINING，因为第一个种子节点将本身率先选举为Leader，所以它还将本身的状态改变为Up。后面它还将第二个种子节点和第三个节点从JOINING转换到Up状态。

咱们中止第三个加入的节点，咱们观察SimpleClusterListener的日志输出以下图所示。

能够看到其状态首先被标记为Down，最后被转换为Removed。

总结

　　经过以上介绍相信你们对使用Akka构建集群有了基本的认识，是否是很轻松？若是想要继续了解如何使用Akka构建集群，请阅读《使用Akka构建集群（二）》。

 

后记：通过近一年的准备，《Spark内核设计的艺术 架构设计与实现》一书现已出版发行，图书如图：

 

售卖连接以下：

京东： https://item.jd.com/12302500.html