Vert.x源码-创建集群-白红宇

Vert.x源码-创建集群

阅读量：7045 次

发布时间：2019-06-28

本文共 6259 字，大约阅读时间需要 20 分钟。

在当前的最新版本中，Vert.x官方只实现了利用Hazelcast来创建集群。当然，如果可以的话，也可以通过ClusterManager接口实现或引入需要的集群管理工具。（3.3.0已经提供了Ignite的技术预览版，期待早日实现）。本文将说明Vert.x是如何利用Hazelcast来创建和管理集群的，同时你也会了解到Vertx如何创建单机实例。

集群创建

在创建Vert.x集调用群时，调用方法和创建单机实例是有差异的。集群需要调Vertx.clusteredVertx异步方法创建。集群可以完全新建和引入已有的Hazelcast实例二种方式来创建。如下：

1.新建实例

ClusterManager mgr = new HazelcastClusterManager();

2.引入Hazelcast实例

ClusterManager mgr = new HazelcastClusterManager(hazelcastInstance);

详情可以参考官方手册。

新建集群过程

调用Vertx.clusteredVertx静态方法后，Vert.x会利用Vertx工厂方法创建Vertx实例。如下

其中简单直白的使用 new VertxImpl();来创建Vertx实例。

VertxFactoryImpl.clusteredVertx(VertxOptions options, final Handler
    
     
      > resultHandler) {    options.setClustered(true);//设置参数，启用集群    new VertxImpl(options, resultHandler);//创建Vertx实例}

图1启动集群

在VertxImpl的构造方法中，若需要创建集群，则执行：

VertxImpl(VertxOptions options, Handler
    
     
      > resultHandler) {    // some code    if (options.isClustered()) {      this.clusterManager = getClusterManager(options);//1.获取集群管理对象      this.clusterManager.setVertx(this);//2. 设置实例      this.clusterManager.join(ar -> {//3. 加入集群        if (ar.failed()) {          log.error("Failed to join cluster", ar.cause());        } else {          // Provide a memory barrier as we are setting from a different thread          synchronized (VertxImpl.this) {            haManager = new HAManager(this, deploymentManager, clusterManager, options.getQuorumSize(),                                      options.getHAGroup(), haEnabled);            createAndStartEventBus(options, resultHandler);          }        }      });    } else {      this.clusterManager = null;      createAndStartEventBus(options, resultHandler);    }    // some code  }

图2

这里会分3部来创建集群，首先调用getClusterManager来获取集群的配置管理实例。如下：

getClusterManager(VertxOptions options) {    if (options.isClustered()) {      if (options.getClusterManager() != null) {//判断是否已经创建集群管理对方        return options.getClusterManager();//若已创建，直接使用这个对象。      } else {//若无创建，执行新建过程。        ClusterManager mgr;        String clusterManagerClassName = System.getProperty("vertx.cluster.managerClass");/*通过系统参数设置集群管理对象*/        if (clusterManagerClassName != null) {//clusterManagerClassName变量指定的类名存在，开始加载          // We allow specify a sys prop for the cluster manager factory which overrides ServiceLoader          try {            Class
     clazz = Class.forName(clusterManagerClassName);            mgr = (ClusterManager)clazz.newInstance();          } catch (Exception e) {            throw new IllegalStateException("Failed to instantiate " + clusterManagerClassName, e);          }        } else {//clusterManagerClassName指定的变量null,使用默认加载器。          ServiceLoader
    
      mgrs = ServiceLoader.load(ClusterManager.class);          if (!mgrs.iterator().hasNext()) {            throw new IllegalStateException("No ClusterManagerFactory instances found on classpath");          }          mgr = mgrs.iterator().next();        }        return mgr;      }    } else {      return null;    }  }

图3，获取集群管理类

从源码看，getClusterManager并没有什么特殊的地方。首先检查用户在创建Vertx实例之前，是否创建了集群的管理对象ClusterManager。创建了，则使用这个管理对象，没有创建则自行新建一个。

注意

String clusterManagerClassName = System.getProperty("vertx.cluster.managerClass");

这行代码，这说明可以通过JVM环境参数（-Dvertx.cluster.managerClass=[className]）来指定Vertx加载集群管理对象类。这在官方手册中并没有任何一个地方说明。

如果指定了managerClass，则会使用默认加载方式加载指定的类，并转换成ClusterManager接口。

如果没有指定managerClass，则使用默认集群加载类启动集群。

ServiceLoader<ClusterManager> mgrs = ServiceLoader.load(ClusterManager.class);

ServiceLoader是Java在1.6定义的聚群接口类，有点类似于spring的Ioc容器。其过程也是加载类。详细说明请查阅一文，解释得很清楚。

可以看到在vertx-hazelcast-[vertsion].jar包中，META-INF/services/io.vertx.core.spi.cluster.ClusterManager指定了ClusterManager要ServiceLoader加载HazelcastClusterManager。

io.vertx.spi.cluster.hazelcast.HazelcastClusterManager

回到图2，Vert.x接下来使用

clusterManager.setVertx(this)

将vertx实例设置到集群管理类中。随后调用

clusterManager.join

来加入集群。下面是clusterManager.join的源码

synchronized void join(Handler
    
     
      > resultHandler) {    vertx.executeBlocking(fut -> {      if (!active) {//确保只初始化一次        active = true;        if (customHazelcastCluster) {//当使用的是用户自己创建的Hazelcast实例时          nodeID = hazelcast.getLocalEndpoint().getUuid();//获取节点编号          membershipListenerId = hazelcast.getCluster().addMembershipListener(this);//获取当前节点监听成员变换的事件的ID          fut.complete();          return;        }        if (conf == null) {//获取Hazelcast的Config          conf = loadConfigFromClasspath();          if (conf == null) {            log.warn("Cannot find cluster configuration on classpath and none specified programmatically. Using default hazelcast configuration");          }        }        //新建hazelcast实例        hazelcast = Hazelcast.newHazelcastInstance(conf);        nodeID = hazelcast.getLocalEndpoint().getUuid();        membershipListenerId = hazelcast.getCluster().addMembershipListener(this);        fut.complete();      }    }, resultHandler);  }

图4，新建hazelcast实例

如果用户自己创建并传入Hazelcast实例，ClusterManager只是简单的从中获取需要的参数。如果未创建实例，则ClusterManager会自行创建。

首先，loadConfigFromClasspath会用来加载本地的配置文件。

Config loadConfigFromClasspath() {    Config cfg = null;    try (InputStream is = getConfigStream();         InputStream bis = new BufferedInputStream(is)) {      if (is != null) {        cfg = new XmlConfigBuilder(bis).build();//创建HazelcastConfig      }    } catch (IOException ex) {      log.error("Failed to read config", ex);    }    return cfg;  }

图5，加载HazelcastConfig

getConfigStream用来读取配置文件。

InputStream getConfigStream() {    ClassLoader ctxClsLoader = Thread.currentThread().getContextClassLoader();    InputStream is = null;    if (ctxClsLoader != null) {      is = ctxClsLoader.getResourceAsStream(CONFIG_FILE);    }    if (is == null) {      is = getClass().getClassLoader().getResourceAsStream(CONFIG_FILE);      if (is == null) {        is = getClass().getClassLoader().getResourceAsStream(DEFAULT_CONFIG_FILE);      }    }    return is;  }

图6，读取配置文件

如图5、图6的源码。getConfigStream会先加载classpath下的cluster.xml（CONFIG_FILE）文件。如果不存在，则加载jar包内的default-cluster.xml（DEFAULT_CONFIG_FILE）文件。读取完毕后，loadConfigFromClasspath使用Hazelcast的XmlConfigBuilder来构建HazelcastConfig。而后会用这个Config初始化Hazelcast。

集群创建成功后，会初始化一个HAManager实例，用于做verticle迁移。后面在详细说明HA模式。

最后，在VertxImpl中，会调用createAndStartEventBus方法在集群环境运行的EventBus。

原文地址：

转载于:https://my.oschina.net/chkui/blog/678347

你可能感兴趣的文章