笔者在测试使用storm时,意外的出现了oom,后台报错如下
导出堆,进行分析时,发现如下:
要分析此问题,首先要知道storm内部的消息传递线程模型到底是什么样的,此时的storm正在进行2.0的开发,虽然还没有发版,但是内核部分,storm已经基本都改成了java实现,github地址:
https://github.com/apache/storm/
根据官网的介绍可知,storm的消息流动是从worker(worker是storm的task节点运行的逻辑节点)开始的,我们先从worker开始分析storm的消息流动,官网介绍如下:
http://storm.apache.org/releases/2.0.0-SNAPSHOT/Message-passing-implementation.html
worker的start方法内,进行了很多工作的处理,包括worker与其他worker节点的连接建立,本地task节点的监控,与supervisor的心跳发送,其中与tuple传输相关部分的代码:
其中workerState的初始化:
workerState是非常重要的一个类,是专门处理底层传输的,并且一个worker内只有一个,由worker内的spout,bolt实例共享。
workerState也会保存根据配置worker内的executor,taskId。
1,mkExecutor这个方法会生成系统配置里已定的boltExecutor以及 SpoutExecutor实例,我们在这里也可以看到,
每个实例内引用的都是同一个workerState方法,
2,Executor的idToTask是一个Map的结构,表明了这个executor负责执行哪些task实例。
3,mkexecutor之后,马上对返回的这个executor实例,执行了execute:
我们可以认为,其中的Utils.asyncLoop()对executorTransfer的call方法进行了无线循环处理。
说白了,execute方法的执行,就是系统起了一个单独的线程,一直调用executorTransfer的call方法,
同时,不管是spoutExecutor还是BoltExecutor,每一个executor都有一个executorTransfer。
下面我们来看executorTransfer的call方法,到底做了什么
可以简单理解为,call方法对私有的batchTransferQueue进行消费,如果batchTransferQueue有数据了,则保存消费索引,对已消费的数据重新组装,丢到workerState里,让workerState负责传输给目的task。
这里的WorkerState实例,是构造Exexutor时传入的,还是worker内生成的那个,workerState实例,整个Worker就一个。
这是为什么呢? 我们看下workerState的transfer方法
其中:
workerState在传输时,上层调用会组装好数据包AddressTuple,表明一个tuple,发送给哪个目的地。所以可以认为workerState是专门负责底层传输的工作类,他不管谁给我的数据,只要指明了目的地,我就帮你传输。
这里workerState内部使用了一个使用基于Disruptor打造的并发框架队列来保存需要传输的消息。
worker内的这段程序表明了,workerState内的transferQueue一旦有数据,则会调用自身的sendTuplesToRemoteWorker方法进行远程传输,顺便说一句,此处目前是使用netty实现远程数据的传输。
上述是storm消息流程模型的部分代码解读,笔者研读了相关部分,大致画了一个草图如下
有了大致的消息流动流程,我们再来分析,上面的堆内存oom现象,根据图可知,堆内有900多w个taskMessage对象,排名第二的是
AddressedTuple,我是bolt节点出的异常,所以我们从上面线程模型图的最下面开始分析:
我们看这个Worker内的registerCallBacks()
这个方法实现了启动nettyserver后,收到数据包之后调用workerState的传输方法进行数据传输
transferLocal方法会把每一个数据包,发布到对应的目的缓冲队列,目的缓冲队列就是每个BoltExecutor或SpoutExecutor内部维护的,根据上图可见tuple之后的流向。
此处我们可以看到server在接受到消息后,会对元数据TaskMessage进行解码,解码后还要生成新的AddressedTuple,再由WorkerState进行传输,由此,我们可以断定,是这里出现的瓶颈,发射点发射数据太快了,bolt点根本来不及消费。
其实这也算必然的了,我写的spout点是一个死循环一直发随机数,没有任何sleep,几分钟,就是几百兆的数据量了。
不过根据此我们也可以看出,做好资源隔离,合理配置bolt,spout资源,以及自动扩容是很重要的。