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

简介:   序:最近对storm平台系统进行性能检测发现偶尔会出现oncebolt向另一个twobolt发送数据后,twobolt要500毫秒后才接收到进行处理。这里简单说增大twobolt的并行度即可解决,但是究其内部原因是因为storm的通信机制所导致的问题。

  序:最近对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(刘洋)
老铁,你的--->推荐,--->关注,--->评论--->是我继续写作的动力。
微信公众号号:Apache技术研究院
由于博主能力有限,文中可能存在描述不正确,欢迎指正、补充!
本文版权归作者和博客园共有,欢迎转载,但未经作者同意必须保留此段声明,且在文章页面明显位置给出原文连接,否则保留追究法律责任的权利。
相关文章
|
消息中间件 监控 Java
图解Kafka线程模型及其设计缺陷
图解Kafka线程模型及其设计缺陷
图解Kafka线程模型及其设计缺陷
|
6月前
|
消息中间件 监控 Java
✈️【Kafka技术专题】「核心原理篇」深入实战探索Kafka的Broker的原理及可靠性机制分析
✈️【Kafka技术专题】「核心原理篇」深入实战探索Kafka的Broker的原理及可靠性机制分析
175 0
|
6月前
|
消息中间件 存储 缓存
Kafka - 3.x 图解Broker总体工作流程
Kafka - 3.x 图解Broker总体工作流程
202 0
|
6月前
|
消息中间件 存储 监控
Kafka Streams:深度探索实时流处理应用程序
Apache Kafka Streams 是一款强大的实时流处理库,为构建实时数据处理应用提供了灵活且高性能的解决方案。本文将深入探讨 Kafka Streams 的核心概念、详细原理,并提供更加丰富的示例代码,以帮助大家深入理解和应用这一流处理框架。
|
消息中间件 Kafka RocketMQ
Kafka重平衡机制
当集群中有新成员加入,或者某些主题增加了分区之后,消费者是怎么进行重新分配分区再进行消费的?这里就涉及到重平衡(Rebalance)的概念,下面我就给大家讲解一下什么是 Kafka 重平衡机制,我尽量做到图文并茂通俗易懂。
1272 0
Kafka重平衡机制
|
消息中间件 存储 负载均衡
「事件驱动架构」Apache Kafka再平衡协议:再平衡协议101
「事件驱动架构」Apache Kafka再平衡协议:再平衡协议101
「事件驱动架构」Apache Kafka再平衡协议:再平衡协议101
|
消息中间件 存储 负载均衡
【Kafka从入门到放弃系列 三】Kafka架构深入——工作流程和存储机制
【Kafka从入门到放弃系列 三】Kafka架构深入——工作流程和存储机制
184 0
|
消息中间件 监控 关系型数据库
详解Kafka设计架构核心——Kafka副本机制详解
所谓的副本机制(Replication),也可以称之为备份机制,通常是指分布式系统在多台网络互联的机器上保存有相同的数据拷贝。副本机制有什么好处呢? 1. 提供数据冗余。即使系统部分组件失效,系统依然能够继续运转,因而增加了整体可用性以及数据持久性。 2. 提供高伸缩性。支持横向扩展,能够通过增加机器的方式来提升读性能,进而提高读操作吞吐量。 3. 改善数据局部性。允许将数据放入与用户地理位置相近的地方,从而降低系统延时。
详解Kafka设计架构核心——Kafka副本机制详解
|
分布式计算 搜索推荐 NoSQL