笔者在测试使用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资源，以及自动扩容是很重要的。
网友评论