摘要: 原创出处 http://www.iocoder.cn/SkyWalking/collector-cluster-module/ 「芋道源码」欢迎转载,保留摘要,谢谢!

本文主要基于 SkyWalking 3.2.6 正式版


🙂🙂🙂关注**微信公众号:【芋道源码】**有福利:

  1. RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表
  2. RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址
  3. 您对于源码的疑问每条留言将得到认真回复。甚至不知道如何读源码也可以请教噢
  4. 新的源码解析文章实时收到通知。每周更新一篇左右
  5. 认真的源码交流微信群。

1. 概述

本文主要分享 SkyWalking Collector Cluster Module,负责集群的管理,即 Collector 节点的注册于发现。

友情提示:建议先阅读 《SkyWalking 源码分析 —— Collector 初始化》 ,以了解 Collector 组件体系。

Cluster Module 在 SkyWalking 架构图处于如下位置( 红框 ) :

FROM https://github.com/apache/incubating-skywalking

下面我们来看看整体的项目结构,如下图所示 :

  • collector-cluster-define :定义集群管理接口。
  • collector-cluster-standalone-provider :基于 H2 的 集群管理实现。该实现是单机版,建议仅用于 SkyWalking 快速上手,生产环境不建议使用
  • collector-cluster-redis-provider :基于 Redis 的集群管理实现。目前暂未完成
  • collector-cluster-zookeeper-provider :基于 Zookeeper 的集群管理实现。生产环境推荐使用

下面,我们从接口到实现的顺序进行分享。

2. collector-cluster-define

collector-cluster-define :定义集群管理接口。项目结构如下 :

  • 交互如下图 :

  • ModuleListenerService 暴露给其他 Module 注册监听器 ( ClusterModuleListener ) 到 DataMonitor 。

  • ModuleRegisterService 暴露给其他 Module 注册组件登记( ModuleRegistration ) 到 DataMonitor 。

  • 通过实现 DataMonitor 接口,基于不同的存储器实现注册发现。

2.1 ClusterModule

org.skywalking.apm.collector.cluster.ClusterModule ,实现 Module 抽象类,集群管理 Module 。

#name() 实现方法,返回模块名为 "cluster"

#services() 实现方法,返回 Service 类名:ModuleListenerService / ModuleRegisterService 。

2.2 ModuleRegisterService

org.skywalking.apm.collector.cluster.service.ModuleRegisterService ,继承 Service 接口,模块注册服务接口

#register(moduleName, providerName, registration) 接口方法,注册模块注册信息。一般情况下,实现该接口方法,调用 DataMonitor#register(path, registration) 方法。

2.2.1 ModuleRegistration

org.skywalking.apm.collector.cluster.ModuleRegistration ,模块注册信息抽象类。不同 Module 通过实现 ModuleRegistration ,将它们注册到 ModuleRegisterService。目前子类如下 :

#buildValue() 抽象方法,获得模块注册信息( Value )。

2.3 ModuleListenerService

org.skywalking.apm.collector.cluster.service.ModuleListenerService ,继承 Service 接口,注册监听器服务接口

#addListener(listener) 接口方法,添加监听器。一般情况下,实现该接口方法,调用 DataMonitor#addListener(listener) 方法。

2.3.1 ClusterModuleListener

org.skywalking.apm.collector.cluster.ClusterModuleListener ,集群组件监听器抽象类。目前子类如下 :

构造方法,创建地址数组( addresses )。该数组的读写方法如下:

  • #addAddress(address)
  • #removeAddress(address)
  • #getAddresses()

#path() 抽象方法,返回路径。该路径即为 ClusterModuleListener 监听的**"事件"**。多个 Collector 节点的相同 Module ,通过路径分组形成集群

#serverJoinNotify(serverAddress) / #serverQuitNotify(serverAddress) 抽象方法,通知服务的加入 / 下线。目前只有 GRPCRemoteSenderService 真正( 其它都是空方法 )实现该方法,在 《SkyWalking 源码分析 —— Collector Remote 远程通信服务》「3.2 GRPCRemoteSenderService」 详细解析。

2.4 DataMonitor

org.skywalking.apm.collector.cluster.DataMonitor ,数据监接口。通过实现 DataMonitor 接口,基于不同的存储器实现注册发现。目前子类如下 :

#register(path, registration) 接口方法,注册模块注册信息。

#addListener(ClusterModuleListener) 接口方法,添加监听器。

#setClient(Client) 接口方法,设置 Client 。在 client-component 有 ZookeeperClient / H2Client / ElasticSearchClient 等多种实现。

  • BASE_CATALOG 属性,基础目录为 "/skywalking" 。例如说,在 Zookeeper 为根节点的路径。
  • #createPath(path) 接口方法,使用 Client 创建路径。
  • #setData(path) 接口方法,使用 Client 设置路径的值。

3. collector-cluster-zookeeper-provider

collector-cluster-zookeeper-provider ,基于 Zookeeper 的集群管理实现。项目结构如下 :

实际使用时,通过 application.yml 配置如下:

cluster:
zookeeper:
hostPort: localhost:2181
sessionTimeout: 100000

  • 生产环境下,推荐 Zookeeper 配置成集群。

3.1 ClusterModuleZookeeperProvider

org.skywalking.apm.collector.cluster.zookeeper.ClusterModuleZookeeperProvider ,实现 ModuleProvider 抽象类,基于 Zookeeper 的集群管理服务提供者。

#name() 实现方法,返回组件服务提供者名为 "zookeeper"

module() 实现方法,返回组件类为 ClusterModule 。

#requiredModules() 实现方法,返回依赖组件为空。


#prepare(Properties) 实现方法,执行准备阶段逻辑。

  • 第 63 行 :创建 ClusterZKDataMonitor 对象。
  • 第 69 行 :创建 ZookeeperClient 对象。注意,此时并未连接 Zookeeper
  • 第 71 至 73 行 :创建 ZookeeperModuleListenerService / ZookeeperModuleRegisterService 对象,并调用 #registerServiceImplementation() 父类方法,注册到 services

#start() 实现方法,执行启动阶段逻辑。

  • 第 79 行 :调用 ZookeeperClient#initialize() 方法,初始化 ZookeeperClient ,此时会连接 Zookeeper

#notifyAfterCompleted() 实现方法,执行启动完成逻辑。

3.2 ZookeeperModuleRegisterService

org.skywalking.apm.collector.cluster.zookeeper.service.ZookeeperModuleRegisterService ,基于 Zookeeper 的模块注册服务实现类

#register(moduleName, providerName, registration) 实现方法,调用 ClusterZKDataMonitor#register(path, registration) 方法,注册模块注册信息。

3.3 ZookeeperModuleListenerService

org.skywalking.apm.collector.cluster.zookeeper.service.ZookeeperModuleListenerService ,基于 Zookeeper 的注册监听器服务实现类

#addListener(ClusterModuleListener) 实现方法,调用 ClusterZKDataMonitor#addListener(ClusterModuleListener) 方法,注册模块注册信息。

3.4 ClusterZKDataMonitor

org.skywalking.apm.collector.cluster.zookeeper.ClusterZKDataMonitor ,基于 Zookeeper 的数据监视器实现类

在看具体代码实现之前,我们先来看看 Zookeeper 是如何存储数据的,如下图所示 :

  • 紫色部分,通过调用 #createPath(path) 方法,顺着路径,逐层创建持久节点。

  • 黄色部分,通过调用 #setData(path) 方法,创建临时节点,设置 Collector 模块地址。若 Collector 集群有 N 个节点,则此处会有 N 个临时节点。

  • 打开 zkClient.sh ,我们来看一个例子 :

    [zk: localhost:2181(CONNECTED) 1] ls /skywalking
    [remote, ui, agent_jetty, agent_gRPC]
    [zk: localhost:2181(CONNECTED) 2] ls /skywalking/ui
    [jetty]
    [zk: localhost:2181(CONNECTED) 3] ls /skywalking/ui/jetty
    [localhost:12800]
    [zk: localhost:2181(CONNECTED) 4] get /skywalking/ui/jetty/localhost:12800
    /
    cZxid = 0x24
    ctime = Thu Dec 14 16:05:25 CST 2017
    mZxid = 0x24
    mtime = Thu Dec 14 16:05:25 CST 2017
    pZxid = 0x24
    cversion = 0
    dataVersion = 0
    aclVersion = 0
    ephemeralOwner = 0x16052d8b9f40006
    dataLength = 1
    numChildren = 0


#register(path, registration) 实现方法,添加到组件注册信息集合( registrations )。

#start() 方法,启动 ClusterZKDataMonitor ,将组件注册信息( registrations ) 写到 Zookeeper 中。


#addListener(listener) 实现方法,添加到监听器集合( listeners )。

#process(WatchedEvent) 实现方法,处理有 Collector 节点的组件加入或下线。总体逻辑是,从 Zookeeper 获取变更的路径下的地址数组,和本地的地址( ClusterModuleListener.addresses )比较,处理加入或移除逻辑的地址。

  • ClusterZKDataMonitor 实现 org.apache.zookeeper.Watcher 接口,所以实现该方法。
  • 该方法是 synchronized 方法,以保证不会出现并发问题。

3.5 ZookeeperClient

org.skywalking.apm.collector.client.zookeeper.ZookeeperClient ,实现 Client 接口,Zookeeper 客户端。

代码比较简单,胖友自己阅读理解。

4. collector-cluster-standalone-provider

collector-cluster-standalone-provider.ClusterStandaloneDataMonitor ,基于 H2 的 集群管理实现。该实现是单机版,建议仅用于 SkyWalking 快速上手,生产环境不建议使用。项目结构如下 :

大体实现和 collector-cluster-zookeeper-provider 差不多,差异在对 DataMonitor 的实现类 ClusterStandaloneDataMonitor 上。

在 ClusterStandaloneDataMonitor 里,实际并未使用 H2Client ,而是基于内存,胖友可以自己查看下。

5. collector-cluster-redis-provider

collector-cluster-redis-provider :基于 Redis 的集群管理实现。目前暂未完成

【TODO 4003】等实现后来写写,基于 Redis Pub Sub 保证实时性

666. 彩蛋

知识星球

有一种硬生生把很简单的东西,写的很复杂的感觉。

胖友,分享个朋友圈可好?

文章目录
  1. 1. 1. 概述
  2. 2. 2. collector-cluster-define
    1. 2.1. 2.1 ClusterModule
    2. 2.2. 2.2 ModuleRegisterService
      1. 2.2.1. 2.2.1 ModuleRegistration
    3. 2.3. 2.3 ModuleListenerService
      1. 2.3.1. 2.3.1 ClusterModuleListener
    4. 2.4. 2.4 DataMonitor
  3. 3. 3. collector-cluster-zookeeper-provider
    1. 3.1. 3.1 ClusterModuleZookeeperProvider
    2. 3.2. 3.2 ZookeeperModuleRegisterService
    3. 3.3. 3.3 ZookeeperModuleListenerService
    4. 3.4. 3.4 ClusterZKDataMonitor
    5. 3.5. 3.5 ZookeeperClient
  4. 4. 4. collector-cluster-standalone-provider
  5. 5. 5. collector-cluster-redis-provider
  6. 6. 666. 彩蛋