1. 概述

本文主要分享 SkyWalking Collector Remote 远程通信服务。该服务用于 Collector 集群内部通信。

目前集群内部通信的目的，跨节点的流式处理。Remote Module 应用在 SkyWalking 架构图如下位置( 红框 ) ：
FROM https://github.com/apache/incubating-skywalking

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

collector-remote-define ：定义远程通信接口。
collector-remote-kafka-provider ：基于 Kafka 的远程通信实现。目前暂未完成。
collector-remote-grpc-provider ：基于 Google gRPC 的远程通信实现。生产环境目前使用

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

2. collector-remote-define

collector-remote-define ：定义远程通信接口。项目结构如下：

整体流程如下图：

我们按照整个流程的处理顺序，逐个解析涉及到的类与接口。

2.1 RemoteModule

org.skywalking.apm.collector.remote.RemoteModule ，实现 Module 抽象类，远程通信 Module 。

#name() 实现方法，返回模块名为 "remote" 。

#services() 实现方法，返回 Service 类名：RemoteSenderService 、RemoteDataRegisterService 。

2.2 RemoteSenderService

org.skywalking.apm.collector.remote.service.RemoteSenderService ，继承 Service 接口，远程发送服务接口，定义了 #send(graphId, nodeId, data, selector) 接口方法，调用 RemoteClient ，发送数据。

graphId 方法参数，Graph 编号。通过 graphId ，可以查找到对应的 Graph 对象。
Graph 在《SkyWalking 源码分析 —— Collector Streaming Computing 流式处理（一）》「2. apm-collector-core/graph」有详细解析。
nodeId 方法参数，Worker 编号。通过 workerId ，可以查找在 Graph 对象中的 Worker 对象，从而 Graph 中的流式处理。
Worker 在《SkyWalking 源码分析 —— Collector Streaming Computing 流式处理（一）》「3. apm-collector-stream」有详细解析。
data 方法参数，Data 数据对象。例如，流式处理的具体数据对象。
Data 在《SkyWalking 源码分析 —— Collector Storage 存储组件》「2. apm-collector-core」有详细解析。
selector 方法参数，org.skywalking.apm.collector.remote.service.Selector 选择器对象。根据 Selector 对象，使用对应的负载均衡策略，选择集群内的 Collector 节点，发送数据。
RemoteSenderService.Mode 返回值，发送模式分成 Remote 和 Local 两种方式。前者，发送数据到远程的 Collector 节点；后者，发送数据到本地，即本地处理，参见 RemoteWorkerRef#in(message) 方法。

2.3 RemoteClientService

org.skywalking.apm.collector.remote.service.RemoteClientService ，继承 Service 接口，远程客户端服务接口，定义了 #create(host, port, channelSize, bufferSize) 接口方法，创建 RemoteClient 对象。

2.4 RemoteClient

org.skywalking.apm.collector.remote.service.RemoteClient ，继承 java.lang.Comparable 接口，远程客户端接口。定义了如下接口方法：

#push(graphId, nodeId, data, selector) 接口方法，发送数据。
#getAddress() 接口方法，返回客户端连接的远程 Collector 地址。
#equals(address) 接口方法，判断 RemoteClient 是否连接了指定的地址。

2.5 CommonRemoteDataRegisterService

在说 CommonRemoteDataRegisterService 之前，首先来说下 CommonRemoteDataRegisterService 的意图。

在上文中，我们可以看到发送给 Collector 是 Data 对象，而 Data 是数据的抽象类，在具体反序列化 Data 对象之前，程序是无法得知它是 Data 的哪个实现对象。这个时候，我们可以给 Data 对象的每个实现类，生成一个对应的数据协议编号。

在发送数据之前，序列化 Data 对象时，增加该 Data 对应的协议编号，一起发送。
在接收数据之后，反序列化数据时，根据协议编号，创建 Data 对应的实现类对象。

org.skywalking.apm.collector.remote.service.CommonRemoteDataRegisterService ，通用远程数据注册服务。

id 属性，数据协议自增编号。
dataClassMapping 属性，数据类型( Class<? extends Data> )与数据协议编号的映射。
dataInstanceCreatorMapping 属性，数据协议编号与数据对象创建器( RemoteDataInstanceCreator )的映射。

2.5.1 RemoteDataRegisterService

org.skywalking.apm.collector.remote.service.RemoteDataRegisterService ，继承 Service 接口，远程客户端服务接口，定义了 #register(Class<? extends Data>, RemoteDataInstanceCreator) 接口方法，注册数据类型对应的远程数据创建器( RemoteDataRegisterService.RemoteDataInstanceCreator )对象。

CommonRemoteDataRegisterService 实现了 RemoteDataRegisterService 接口，#register(Class<? extends Data>, RemoteDataInstanceCreator) 实现方法。

另外，AgentStreamRemoteDataRegister 会调用 RemoteDataRegisterService#register(Class<? extends Data>, RemoteDataInstanceCreator) 方法，注册每个数据类型的 RemoteDataInstanceCreator 对象。注意，例如 Application::new 是 RemoteDataInstanceCreator 的匿名实现类。

2.5.2 RemoteDataIDGetter

org.skywalking.apm.collector.remote.service.RemoteDataIDGetter ，继承 Service 接口，远程数据协议编号获取器接口，定义了 #getRemoteDataId(Class<? extends Data>) 接口方法，根据数据类型获取数据协议编号。

CommonRemoteDataRegisterService 实现了 RemoteDataIDGetter 接口，#getRemoteDataId(Class<? extends Data>) 实现方法。

2.5.3 RemoteDataInstanceCreatorGetter

org.skywalking.apm.collector.remote.service.RemoteDataInstanceCreatorGetter ，继承 Service 接口，远程数据创建器的获取器接口，定义了 #getInstanceCreator(remoteDataId 接口方法，根据数据协议编号获得远程数据创建器( RemoteDataInstanceCreator )。

CommonRemoteDataRegisterService 实现了 RemoteDataInstanceCreatorGetter 接口，#getInstanceCreator(remoteDataId) 实现方法。

2.6 RemoteSerializeService

org.skywalking.apm.collector.remote.service.RemoteSerializeService ，远程通信序列化服务接口，定义了 #serialize(Data) 接口方法，序列化数据，生成 Builder 对象。

2.7 RemoteSerializeService

org.skywalking.apm.collector.remote.service.RemoteDeserializeService ，远程通信序反列化服务接口，定义了 #deserialize(RemoteData, Data) 接口方法，反序列化传输数据。

3. collector-remote-grpc-provider

collector-remote-grpc-provider ，基于 Google gRPC 的远程通信实现。

项目结构如下：

默认配置，在 application-default.yml 已经配置如下：

|

remote:
  gRPC:
    host: localhost
    port: 11800

|

3.1 RemoteModuleGRPCProvider

org.skywalking.apm.collector.remote.grpc.RemoteModuleGRPCProvider ，实现 ModuleProvider 抽象类，基于 gRPC 的组件服务提供者实现类。

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

module() 实现方法，返回组件类为 RemoteModule 。

#requiredModules() 实现方法，返回依赖组件为 cluster 、gRPC_manager 。

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

第 53 至 56 行：创建 CommonRemoteDataRegisterService 、GRPCRemoteSenderService 对象，并调用 #registerServiceImplementation() 父类方法，注册到 services 。

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

Server 相关
第 65 行：创建 gRPC Server 对象。
第 67 行：注册 RemoteCommonServiceHandler 对象到 gRPC Server 上，用于接收 gRPC 请求后的处理。
《SkyWalking 源码分析 —— Collector Server Component 服务器组件》「3. gRPC 实现」
《SkyWalking 源码分析 —— Collector gRPC Server Manager》
注册发现相关
第 70 至 71 行：创建 org.skywalking.apm.collector.remote.grpc.RemoteModuleGRPCRegistration 对象，将自己注册到集群管理。这样，自己可以被 Collector 集群节点发现，从而被调用。
第 73 至 74 行：注册 GRPCRemoteSenderService 对象到集群管理。这样，自己可以监听到 Collector 集群节点的加入或离开，从而调用。
《SkyWalking 源码分析 —— Collector Cluster 集群管理》

#notifyAfterCompleted() 实现方法，方法为空。

3.2 GRPCRemoteSenderService

org.skywalking.apm.collector.remote.grpc.service.GRPCRemoteSenderService ，继承 ClusterModuleListener 抽象类，实现 RemoteSenderService 接口，基于 gPRC 的远程发送服务实现类。

3.2.1 注册发现

通过继承 ClusterModuleListener 抽象类，实现了监听 Collector 集群节点的加入或离开。

remoteClients 属性，连接 Collector 集群节点的客户端数组。每个 Collector 集群节点，对应一个客户端。
#path() 实现方法，返回监听的目录 "/" + RemoteModule.NAME + "/" + RemoteModuleGRPCProvider.NAME 。Collector 集群中，每个节点的 Remote Server 都会注册到该目录下。
#serverJoinNotify(serverAddress) 实现方法，当新的节点加入，创建新的客户端连接。
#serverQuitNotify(serverAddress) 实现方法，当老的节点离开，移除对应的客户端连接。

3.2.2 负载均衡

RemoteModuleGRPCProvider 基于不同的选择器 ( Selector ) ，提供不同的客户端选择( org.skywalking.apm.collector.remote.grpc.service.selector.RemoteClientSelector )实现：

hashCodeSelector 属性，HashCodeSelector ，基于数据的哈希码。
foreverFirstSelector 属性，ForeverFirstSelector ，基于客户端数组的顺序，选择第一个。
rollingSelector 属性，RollingSelector ，基于客户端数组的顺序，顺序向下选择。
#send(graphId, nodeId, data, selector) 方法，代码如下：
第 63 、66 、69 行：根据选择器，调用 RemoteClientSelector#select(clients, data) 方法，选择客户端。
第 64 、67 、70 行：调用 #sendToRemoteWhenNotSelf(remoteClient, graphId, nodeId, data) 方法，发送请求数据。
第 76 至 77 行：当选择的客户端连接的是本地时，不发送数据，交给本地处理，参见 RemoteWorkerRef#in(message) 方法。
第 78 至 81 行：当选择的客户端连接的是远程时，调用 RemoteClient#push(graphId, nodeId, data) 方法，发送数据。

3.3 GRPCRemoteClientService

org.skywalking.apm.collector.remote.grpc.service.GRPCRemoteClientService ，实现 RemoteClientService 接口，基于 gRPC 的远程客户端服务实现类。

#create(host, port, channelSize, bufferSize) 实现方法，创建 GRPCRemoteClient 对象。

3.4 GRPCRemoteClient

友情提示：本小节会涉及较多 gRPC 相关的知识，建议不熟悉的胖友自己 Google ，补充下姿势。

org.skywalking.apm.collector.remote.grpc.service.GRPCRemoteClient ，实现 RemoteClient 接口，基于 gRPC 的远程客户端实现类。

client 属性，GRPCClient 对象。相比来说，GRPCRemoteClient 偏业务的封装，内部调用 GRPCClient 对象。
carrier 属性，DataCarrier 对象，本地消息队列。GRPCRemoteClient 在被调用发送数据时，先提交到本地队列，异步消费进行发送到远程 Collector 节点。DataCarrier 在《SkyWalking 源码分析 —— DataCarrier 异步处理库》详细解析。
第 63 行：调用 DataCarrier#consume(IConsumer, num) 方法，设置消费者为 RemoteMessageConsumer 对象。

#push(graphId, nodeId, data) 实现方法，异步发送消息到远程 Collector 。

第 73 行：调用 RemoteDataIDGetter#getRemoteDataId(Class<? extends Data>) 方法，获得数据协议编号。
第 76 至 80 行：创建传输数据( RemoteMessage.Builder ) 对象。RemoteMessage 通过 Protobuf 创建定义，如下图所示：

第 83 行：调用 DataCarrier#produce(data) 方法，发送数据到本地队列。

RemoteMessageConsumer ，批量消费本地队列的数据，逐条发送数据到远程 Collector 节点。

#consume(List<RemoteMessage>) 实现方法，代码如下：
第 100 行：创建 StreamObserver 对象。StreamObserver 主要是 gPRC 相关的 API 的调用。
第 101 至 103 行：调用 io.grpc.stub.StreamObserver#onNext(RemoteMessage) 方法，逐条发送数据。
第 106 行：调用 io.grpc.stub.StreamObserver#onCompleted() 方法，全部请求数据发送完成。

3.5 RemoteCommonServiceHandler

org.skywalking.apm.collector.remote.grpc.handler.RemoteCommonServiceHandler ，实现 org.skywalking.apm.collector.server.grpc.GRPCHandler 接口，继承 RemoteCommonServiceGrpc.RemoteCommonServiceImplBase 抽象类，远程通信通用逻辑处理器。

其中，RemoteCommonServiceGrpc.RemoteCommonServiceImplBase 在 RemoteCommonService.proto 文件的定义如下图：

#call(StreamObserver<Empty>) 实现方法，代码如下：

#onNext(RemoteMessage) 方法，处理每一条消息，代码如下：
第 65 行：调用 RemoteDataInstanceCreatorGetter#getInstanceCreator(remoteDataId) 方法，获得数据协议编号对应的 RemoteDataInstanceCreator 对象。然后，调用 RemoteDataInstanceCreator#createInstance(id) 方法，创建数据协议编号对应的 Data 实现类对应的对象。
第 70 行：调用 GraphManager#findGraph(graphId) 方法，获得 graphId 对应的 Graph 对象。然后，调动 GraphNodeFinder#findNext(nodeId) 方法，获得 Next 对象。
第 71 行：调用 Next#execute(Data) 方法，继续流式处理。

3.6 GRPCRemoteSerializeService

org.skywalking.apm.collector.remote.grpc.service.GRPCRemoteSerializeService ，实现 RemoteSerializeService 接口，基于 gRPC 的远程通信序列化服务实现类。

3.7 GRPCRemoteDeserializeService

org.skywalking.apm.collector.remote.grpc.service.GRPCRemoteDeserializeService ，实现 GRPCRemoteDeserializeService 接口，基于 gRPC 的远程通信反序列化服务实现类。

4. collector-remote-grpc-provider

collector-remote-kafka-provider ：基于 Kafka 的远程通信实现。

SkyWalking 源码分析 —— Collector Remote 远程通信服务