storm1.0节点间消息传递过久分析及调优

序:最近对storm平台系统进行性能检测发现偶尔会出现oncebolt向另一个twobolt发送数据后，twobolt要500毫秒后才接收到进行处理。这里简单说增大twobolt的并行度即可解决，但是究其内部原因是因为storm的通信机制所导致的问题。　　先介绍背景:一个拓扑的结构，spout(并行度:1)[处理性能:capacity 0.04],oncebolt(并行度:20)[处理性能:capacity 0.2],twobolt(并行度:100)[处理性能:capacity 0.6];整个拓扑就我预估最大的处理量就是一秒一千条。

原文和作者一起讨论:http://www.cnblogs.com/intsmaze/p/6544017.htmll

微信：intsmaze

最近对系统进行性能检测，统计整个storm系统中一条消息处理中各个IO耗时的时间，找出性能瓶颈。发现除了活动匹配中会有分布式锁以及大量的redis的IO操作，导致最多会耗时30ms,以及从Hbase中查询数据时由于hbase集群当时正在跑任务导致耗时1~2s。唯一出现的问题就是onebolt向twobolt发送数据后，某些数据耗时几百毫秒才会被twobolt接收到。这就引起了我的注意。先上一下伪代码：

public class OnceBolt extends BaseRichBolt{
    private static final long serialVersionUID = -5283595260540124273L;
    
    private OutputCollector collector;
    
    
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }
    public void execute(Tuple input) {long intsmazeTime=System.currentTimeMillis();
        collector.emit(input,new Values(intsmazeTime));
    }
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("intsmaze"));
    }
}

public class TwoBolt extends BaseRichBolt{
    private static final long serialVersionUID = -5283595260540124273L;
    
    private OutputCollector collector;
    
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }
    public void execute(Tuple input) {long intsmazeTime=input.getLong(0);
            System.out.println("耗时:"+(System.currentTimeMillis()-intsmazeTime));
    }
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    }
}

这个问题从storm内部通信来说:

每个executor有自己的接收队列和输出队列。

每个worker进程有一个独立的接收线程将外部发送过来的消息移动到对应的executor线程的接收队列中。

每个worker存在一个独立的发送线程负责从worker的传输队列中读取消息，并通过网络发送给其他worker。

每个executor有单独的线程分别来处理spout/bolt的业务逻辑，业务逻辑输出的中间数据会存放在输出队列中，当executor的输出队列中的tuple达到一定的阀值，executor的发送线程将批量获取输出队列中的tuple,并发送到work中的传输队列中。

因为oncebolt任务向自己的发送队列生产过快，且向twobolt任务的接收队列发送数据过多，导致twobolt的接收队列满了，twobolt处理不过来了。[简单说就是oncebolt生产数据的速度快于twobolt的消费速率]。这个时候就会出现twobolt处理一个oncebolt的消息要几百毫秒。这个情况是因为twobolt的处理一条消息平均要50毫秒，twobolt接收队列长度是10，刚好twobolt在从队列拉取一条消息处理时，twobolt的接收队列满了，这个时候队列中第10条消息等被处理就会阻塞10*50毫秒的。　　同时因为接收队列满了，oncebolt就会阻塞到，等twobolt接收队列有空了再去发送（很多文章说会导致消息丢失，但是我测试发现没有这种情况，只会阻塞到,这种就是流量洪峰下，storm会出现的一种情况）。这种情况是某几秒消息量过大导致产生，所以这种情况只是偶尔发送，过一会就会正常了，但是如果交易量一直很大，这个时候我们就要进行调优了，最简单的就是增大twobolt的并行度以及work数量。　　个人认为的最优并行度设置:我们可以参照每一个节点的capacity的性能指标，比如我们这里spout的指标是0.04所以就不需要再增加它的并行度和kafka的分区保持一致。oncebolt的指标是0.2,而twobolt的指标是0.6。很明显是oncebolt资源被浪费了或者twobolt的速率跟不上oncebolt，我们给oncebolt的并行度可以减少一半，比如10个。这种方式是减少资源的浪费。或者就目前的问题，增大twobolt的并行度来提示消费的速度。　　还有一个问题我说一下：storm的性能提升我们是增加work数量还是增加节点的并行度。　　这个是一个调优的过程，如果我们只启动一个work，一昧的在这个work中增加并行度，这样会导致频繁的full GC，因为一个work的2G资源供所有的任务一起用；或者我们启动10个work,每个work只启动一个任务，先不说浪费资源，首先在任务间传递消息时就一定会走网络通信这也是速率的消耗。所以是一句话，一个work中的任务数量要合理，不要太多，也不要太少，这是一个调优的过程。