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技术研究院
由于博主能力有限,文中可能存在描述不正确,欢迎指正、补充!
本文版权归作者和博客园共有,欢迎转载,但未经作者同意必须保留此段声明,且在文章页面明显位置给出原文连接,否则保留追究法律责任的权利。
相关文章
|
6月前
|
消息中间件 缓存 监控
Flink背压原理以及解决优化
Flink背压原理以及解决优化
423 0
|
5月前
|
消息中间件 运维 数据管理
Kafka 如何基于 KRaft 实现集群最终一致性协调
Kafka 3.3.1 引入了 KRaft 元数据管理组件,替代 Zookeeper,以简化集群一致性维护,支持更大规模集群并减轻运维复杂性。在 Zookeeper 模式下,需同时运维 ZK 和 Broker,而 KRaft 模式仅需 3 个节点即可构成最小生产集群,且通信协调基于 Raft 协议,增强了一致性。KRaft 模式中,Controller 使用单线程处理请求,通过 KRaft 保持内存状态与多节点一致性。此外,Broker 根据 KRaft 记录更新元数据,实现声明式管理,提高集群协调效率。KRaft 的引入是集群协调机制的演进,采用事件驱动模型实现元数据的一致性。
300 1
Kafka 如何基于 KRaft 实现集群最终一致性协调
|
6月前
|
消息中间件 监控 Java
✈️【Kafka技术专题】「核心原理篇」深入实战探索Kafka的Broker的原理及可靠性机制分析
✈️【Kafka技术专题】「核心原理篇」深入实战探索Kafka的Broker的原理及可靠性机制分析
191 0
|
流计算
125 Storm 消息容错机制
125 Storm 消息容错机制
45 0
|
存储 消息中间件 传感器
超越Storm,SparkStreaming——Flink如何实现有状态的计算
超越Storm,SparkStreaming——Flink如何实现有状态的计算
196 0
超越Storm,SparkStreaming——Flink如何实现有状态的计算
|
分布式计算 搜索推荐 NoSQL
|
消息中间件 数据采集 监控
图解Kafka中的数据采集和统计机制
图解Kafka中的数据采集和统计机制
图解Kafka中的数据采集和统计机制
|
存储 大数据 分布式数据库
Flink状态管理和容错机制介绍
本文来自2018年8月11日在北京举行的 Flink Meetup会议,分享来自于施晓罡,目前在阿里大数据团队部从事Blink方面的研发,现在主要负责Blink状态管理和容错相关技术的研发
1441 0
下一篇
无影云桌面