点击上方“芋道源码”,选择“设为星标”
做积极的人,而不是积极废人!
源码精品专栏
摘要: 原创出处 http://www.iocoder.cn/SkyWalking/collector-streaming-first/ 「芋道源码」欢迎转载,保留摘要,谢谢!
本文主要基于 SkyWalking 3.2.6 正式版
本文主要分享 Collector Streaming 流式处理。主要包含如下部分:
apm-collector-core
模块的 graph
包,提供最精简、单节点的流式处理的封装。如下图所示:
apm-collector-stream
模块,在 graph
包的基础上,提供异步、跨节点等等的流式处理的封装。如下图所示:
免打脸大保健:笔者对流式处理非常不了解,本文可能是一本正经的胡说八道。考虑到笔者是靠脸吃饭(颜值我只服我红雷哥),所以读者老爷请爱护下笔者。
Collector Streaming 在 SkyWalking 架构图处于如下位置( 红框 ) :
FROM https://github.com/apache/incubating-skywalking
OK,下面来一本正经的代码走起!
整体类图如下:
看起来略复杂,不要方,我们先来看一个流式大数据处理框架 Apache Storm 的说明:
FROM 《流式大数据处理的三种框架:Storm,Spark和Samza》 在 Storm 中,先要设计一个用于实时计算的图状结构,我们称之为拓扑(topology)。这个拓扑将会被提交给集群,由集群中的主控节点(master node)分发代码,将任务分配给工作节点(worker node)执行。
整体交互流程如下:
整体顺序图如下:
apm-collector-stream
模块,是 NodeProcessor 接口的一种实现,处理提交给 Node 的数据。在 #onWork(message)
抽象方法里,子类可以实现该方法,根据自身需求,是否调用 #onNext(message)
方法,Next 逐个使用 NodeWay 数组提交给下面的 Node ,继续处理。下面,我们来详细分别看看如下逻辑的详细代码实现:
创建 Graph 的顺序图如下:
GraphManager#createIfAbsent(graphId, input)
方法( input
参数没用 ),创建一个 Graph 对象。Graph#addNode(WayToNode)
方法,创建该 Graph 的首个 Node 对象。Node#addNext(WayToNode)
方法,创建该 Node 的下一个 Node 对象。如下是 collector-agent-stream-provider
模块,TraceStreamGraph#createServiceReferenceGraph()
方法的代码:
public void createServiceReferenceGraph() {
QueueCreatorService<ServiceReference> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);
RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
Graph<ServiceReference> graph = GraphManager.INSTANCE.createIfAbsent(SERVICE_REFERENCE_GRAPH_ID, ServiceReference.class);
graph.addNode(new ServiceReferenceAggregationWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener))
.addNext(new ServiceReferenceRemoteWorker.Factory(moduleManager, remoteSenderService, SERVICE_REFERENCE_GRAPH_ID).create(workerCreateListener))
.addNext(new ServiceReferencePersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));
}
让我们来看看每个方法的具体代码实现。
第一步
GraphManager#createIfAbsent(graphId, input)
方法,创建一个 Graph 对象,并添加到 Graph 映射。代码如下:
INSTANCE
属性,单例。allGraphs
属性,Graph 映射。其中映射的 KEY 为每个 Graph 全局唯一编号。在 JvmMetricStreamGraph 、RegisterStreamGraph 、TraceStreamGraph 类中,枚举了实际使用的 Graph 编号们。第二步
Graph#addNode(WayToNode)
方法,创建该 Graph 的首个 Node 对象。代码如下:
id
属性,Graph 编号。entryWay
,首个提交数据给 Node 的方式。entryWay
赋值给 this.entryWay
属性。在下分享的 Graph#start(input)
方法里,我们会看到这是 Graph 启动的入口,首个提交给 Node 的方式。WayToNode#buildDestination(Graph)
方法,创建 Node 对象,并返回该 Node。在上文中,我们已经说过创建的 Node 对象,为该 Graph 的首个 Node 。WayToNode#buildDestination()
方法,创建该 WayToNode 的 Node 对象。代码如下:
destination
属性,目标 Node 。即该 WayToNode 提交数据到的 Node 。destinationHandler
属性,目标 Node 的处理器。见 `#out(INPUT)` 方法。Node 构造方法
方法,代码如下:
nodeProcessor
属性,节点处理器。next
属性,包含 WayToNode 数组,即 Node 提交数据给 Next 的 Node 数组的方式。Graph#checkForNewNode(Node)
方法,校验 Node 的 NodeProcessor 在其 Graph 里,编号唯一。Graph#checkForNewNode(Node)
方法,校验 Node 的 NodeProcessor 在 Graph 里,编号唯一,代码如下:
nodeIndex
属性,处理器编号与 Node 的映射。其中映射的 KEY 为 NodeProcessor#id()
。第三步
Node#addNext(WayToNode)
方法,创建该 Node 的下一个 Node 对象。代码如下:
WayToNode#buildDestination(Graph)
方法,创建该 Node 的下面的 Node 对象。next
属性。创建 Graph 的顺序图如下:
数据流向 | FROM | TO | 逻辑 |
---|---|---|---|
第一步 | Graph | WayToNode | |
第二步 | WayToNode | Node | |
第三步 | Node | NodeProcessor | |
第四步 | NodeProcessor | Next | 根据具体实现,若到 Next ,重复第一步 |
第一步
Graph#start(input)
方法,启动 Graph ,处理数据。代码如下:
WayToNode#in(input)
方法,输入数据给 WayToNode 。WayToNode#in(input)
抽象方法,以 DirectWay#in(input)
实现方法举例子,代码如下:
super#out(input)
方法,直接输出数据,调用 Node#execute(input)
方法,提交数据给 Node ,进行处理。第二步
Node#execute
方法,调用 NodeProcessor#process(input, next)
方法,处理数据。
第三步
NodeProcessor#process(input, next)
接口方法,以 AbstractWorker#process(input, next)
实现方法举例子,代码如下:
next
赋值给 this.next
属性。this.next
属性,用于封装的 #onNext(OUTPUT)
方法,提交数据给当前 Node 的 Next ( 下面的 Node 们 )继续处理数据。#onWork
方法里,调用 #onNext(OUTPUT)
方法,例如 `ApplicationRegisterRemoteWorker#onWork(Application)` 。第四步
Next#execute(INPUT)
方法,循环 WayToNode 数组,输入数据给 WayNode ,相当于"重回"【第一步】。
在文章的开头,我们提到了 apm-collector-stream
模块,在 graph
包的基础上,提供异步、跨节点等等的流式处理的封装。主要在 WayToNode 、NodeProcessor 的实现类上做文章。
整体类图如下:
org.skywalking.apm.collector.stream.worker.base.WorkerRef
,Worker 引用抽象类。
在 apm-collector-stream
模块里,我们会发现类的命名从 Node / NodeProcessor 转向了 Worker ?这是为什么呢?关于这一点,我们特意采访( 请教 )了官方大佬。
Worker 更具业务含义 Node / Processor 更偏技术含义
目前,WorkerRef 无具体的方法。
org.skywalking.apm.collector.stream.worker.base.LocalAsyncWorkerRef
,异步 Worker 引用实现类,提供了异步的流式处理封装。
我们回到 「2.2 Graph 创建」 的【第一步】。
LocalAsyncWorkerRef#in(INPUT)
方法,代码如下:
那么为什么会回调呢?LocalAsyncWorkerRef 实现了 org.skywalking.apm.collector.queue.base.QueueExecutor
接口,它自身被设置到 QueueEventHandler 中, 作为"事件"的执行器。
整体流程如下:
org.skywalking.apm.collector.stream.worker.base.RemoteWorkerRef
,远程 Worker 引用实现类,提供了远程跨节点的流式处理的封装。
我们再回到 「2.2 Graph 创建」 的【第一步】。
RemoteWorkerRef#in(INPUT)
方法,代码如下:
remoteSenderService
属性,远程发送服务。在 《SkyWalking 源码分析 —— Collector Remote 远程通信服务》「3.2 GRPCRemoteSenderService」 我们会详细解析它的代码实现,这里只简单介绍下。remoteWorker
属性,远程 Worker 。在下文会详细分享它的实现。RemoteSenderService#send(…)
方法,根据远程 Worker 的 Selector 选择器,选择一个 Worker 进行发送。#out(INPUT)
方法,提交数据到本地的 Worker ( Node )。整体类图如下:
AbstractWorker 的代码实现,在 「2.2 Graph 启动」 已经详细解析。
org.skywalking.apm.collector.stream.worker.base.AbstractWorkerProvider
,Worker 供应者抽象类,定义了 #workerInstance(ModuleManager)
抽象方法,用于创建 Worker 对象。
org.skywalking.apm.collector.stream.worker.base.AbstractLocalAsyncWorker
,异步 Worker 抽象类。
目前,AbstractLocalAsyncWorker 无具体的方法。
实际使用时,继承 AbstractLocalAsyncWorker 类,实现 #work(INPUT)
方法,例如:ApplicationRegisterSerialWorker 。
org.skywalking.apm.collector.stream.worker.base.AbstractLocalAsyncWorkerProvider
,LocalAsyncWorker 供应者抽象类。
#create(WorkerCreateListener)
]() 实现方法,创建 AbstractLocalAsyncWorker 和 LocalAsyncWorkerRef 对象。org.skywalking.apm.collector.stream.worker.base.AbstractRemoteWorker
,远程 Worker 抽象类,定义了 #selector()
抽象方法,获得选择器。RemoteSenderService 根据选择器,调用 RemoteClientSelector#select(...)
方法,选择好远程节点,而后进行发送数据。
实际使用时,继承 AbstractLocalAsyncWorker 类,实现 #work(INPUT)
方法,例如:ApplicationRegisterRemoteWorker 。
org.skywalking.apm.collector.stream.worker.base.AbstractRemoteWorkerProvider
,AbstractRemoteWorker 供应者抽象类。
欢迎加入我的知识星球,一起探讨架构,交流源码。加入方式,长按下方二维码噢:
已在知识星球更新源码解析如下:
如果你喜欢这篇文章,喜欢,转发。
生活很美好,明天见(。・ω・。)ノ♡