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

  1. 云栖社区>
  2. 博客>
  3. 正文

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

刘洋intsmaze 2017-03-14 09:15:00 浏览358
展开阅读全文

  序:最近对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中的任务数量要合理,不要太多,也不要太少,这是一个调优的过程。

作者:intsmaze(刘洋)
出处:http://www.cnblogs.com/intsmaze/
老铁,你的--->推荐,--->关注,--->评论--->是我继续写作的动力。
微信公众号号:Apache技术研究院
由于博主能力有限,文中可能存在描述不正确,欢迎指正、补充!
本文版权归作者和博客园共有,欢迎转载,但未经作者同意必须保留此段声明,且在文章页面明显位置给出原文连接,否则保留追究法律责任的权利。

网友评论

登录后评论
0/500
评论
刘洋intsmaze
+ 关注