使用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文件的内容如下：

[plain]
view plain
copy

print?





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 {
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的远程调用》一文中的配置有很多不同：

provider不再是akka.remote.RemoteActorRefProvider，而是akka.cluster.ClusterActorRefProvider。这说明ActorRef将由akka.cluster.ClusterActorRefProvider提供；
增加了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

[java]
view plain
copy

print?





@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);  
        }  
  
    }  
}  

@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);
}

}
}

运行展示

初始化的代码如下：

[java]
view plain
copy

print?





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

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应用的博文如下：

《Spring与Akka的集成》；
《使用Akka的远程调用》；

《使用Akka构建集群（一）》；
《使用Akka构建集群（二）》；
《使用Akka持久化——持久化与快照》；
《使用Akka持久化——消息发送与接收》；

后记：个人总结整理的《深入理解Spark：核心思想与源码分析》一书现在已经正式出版上市，目前京东、当当、天猫等网站均有销售，欢迎感兴趣的同学购买。



京东：http://item.jd.com/11846120.html 

当当：http://product.dangdang.com/23838168.html