使用Akka构建集群(一)

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
简介: 版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/beliefer/article/details/53887181 概述 Akka提供的非常吸引人的特性之一就是轻松构建自定义集群,这也是我要选择Akka的最基本原因之一。
版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/beliefer/article/details/53887181

概述

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

  1. Spring与Akka的集成》;
  2. 使用Akka的远程调用》;
  3. 使用Akka构建集群(一)》;
  4. 使用Akka构建集群(二)》;
  5. 使用Akka持久化——持久化与快照》;
  6. 使用Akka持久化——消息发送与接收》;


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


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

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


相关实践学习
日志服务之使用Nginx模式采集日志
本文介绍如何通过日志服务控制台创建Nginx模式的Logtail配置快速采集Nginx日志并进行多维度分析。
相关文章
|
1月前
|
消息中间件 存储 监控
构建高可用性Apache Kafka集群:从理论到实践
【10月更文挑战第24天】随着大数据时代的到来,数据传输与处理的需求日益增长。Apache Kafka作为一个高性能的消息队列服务,因其出色的吞吐量、可扩展性和容错能力而受到广泛欢迎。然而,在构建大规模生产环境下的Kafka集群时,保证其高可用性是至关重要的。本文将从个人实践经验出发,详细介绍如何构建一个高可用性的Kafka集群,包括集群规划、节点配置以及故障恢复机制等方面。
90 4
|
2月前
|
存储 SQL 消息中间件
Hadoop-26 ZooKeeper集群 3台云服务器 基础概念简介与环境的配置使用 架构组成 分布式协调框架 Leader Follower Observer
Hadoop-26 ZooKeeper集群 3台云服务器 基础概念简介与环境的配置使用 架构组成 分布式协调框架 Leader Follower Observer
54 0
|
7月前
|
Kubernetes 网络协议 Java
在Kubernetes上运行Flink应用程序时
【2月更文挑战第27天】在Kubernetes上运行Flink应用程序时
77 10
|
7月前
|
SQL Kubernetes Java
Flink构建问题之flink 1.11 on kubernetes构建失败如何解决
Apache Flink是由Apache软件基金会开发的开源流处理框架,其核心是用Java和Scala编写的分布式流数据流引擎。本合集提供有关Apache Flink相关技术、使用技巧和最佳实践的资源。
|
分布式计算 Java 网络安全
Flink独立集群部署和HA部署
场景描述 172.19.9.202 主节点 JobManager 主/从 172.19.9.201 从节点 TaskManager 主/从 172.19.9.203 从节点 TaskManager 主/从
262 0
|
分布式计算 Java Hadoop
ZooKeeper分布式协调服务(节点分类、事件监听、java集成)
ZooKeeper分布式协调服务(节点分类、事件监听、java集成)
ZooKeeper分布式协调服务(节点分类、事件监听、java集成)
|
Java Linux Scala
将 Akka 项目打包分布式部署 | 学习笔记
快速学习将 Akka 项目打包分布式部署
将 Akka 项目打包分布式部署 | 学习笔记
|
存储 Java 大数据
环境篇之 flink 的 HA 部署|学习笔记
快速学习环境篇之 flink 的 HA 部署
136 0
|
分布式计算 Hadoop 流计算
如何搭建Storm集群?
如何搭建Storm集群?
111 0
|
流计算
storm集群的搭建
最近也是有朋友问我storm的问题,好长时间没玩storm了,今天就来简单的说一下吧,首先我们来看一下官网的图片,storm是完全实时的,就像水龙头打开后一样,会不停的往外面流水.所以他的延迟非常的低,这也是他的特点.然后先搭建storm集群吧,storm集群的搭建也比较简单.