理解Storm的内部消息缓冲机制-低调大师

理解Storm的内部消息缓冲机制

2017-05-18 574

这篇文章是Apache Kafka的作者之一Michael G. Noll写的，他的博客地址在[这里]。

优化Storm计算拓扑性能的过程有助于我们理解Storm内部消息队列的配置和使用，在这篇简短的文章中，我将向大家解释并说明Storm（0.8或0.9版本）的一个工作进程（worker process）和与其相关的多个执行器线程是如何完成内部通信的。

Storm工作进程（Worker processes）的内部消息机制

在以下各章节中，我会交替地使用消息（message）和元组（tuple）两个关键字。

本文中当我提到“内部消息”时，它指的是发生在Storm的单个工作进程内部的消息通信，这类通信只在Storm集群的单台主机（节点）上展开。Storm使用由LMAX Disruptor提供的很多消息队列来完成通信，LMAX Disruptor是一个高性能的线程间消息通信库。

需要注意的是单个工作进程内多个线程之间的通信不同于Storm的多个工作进程之间的通信，工作进程间的通信通常都需要跨主机通过网络完成，对于工作进程间的通信Storm默认采用 ZeroMQ作为基础通信组件（在Storm 0.9版本中开始实验性地支持 Netty），也就是说，当一个工作进程中的任务（Task）想要发送数据到集群中另外一台主机的某个工作进程的某个任务时，Storm将使用ZeroMQ或Netty进行通信。

所以有如下结论供大家参考：

Storm工作进程内部的通信（同一个Storm工作节点（主机）的线程之间）：LMAX Disruptor
工作进程之间的通信（通过网络的工作节点（主机）之间）：ZeroMQ或者Netty
计算拓扑之间的通信：Storm并不原生支持，需要自己实现和维护，可以使用消息系统，比如：Kafka，RabbitMQ和数据库等等。

图片说明

在下一节讨论具体细节前，我们先来看看如下这幅图（参考原图）：

图1：Storm的单个工作进程的内部消息队列概览，与工作进程相关的队列都被标记为红色，与工作进程中的多个执行器线程相关的队列都被标记为绿色，为了看起来清晰，这里只展示了一个工作进程（一般在Storm集群的单个节点上运行着多个工作进程）和这个工作进程中的一个执行器中的所有线程（一个工作进程中一般也存在多个执行器）。

详细描述

现在我们对Storm的工作进程内部的消息机制有了一定了解，接下来可以深入讨论细节了。

工作进程（Worker processes）

为了管理自己的输入消息和输出消息，每个工作进程都有一个消息接收线程和一个消息发送线程，消息接收线程监听工作进程的TCP端口（通过supervisor.slots.ports参数进行配置），消息接收线程还会将收到的消息批量地发送到执行器线程的输入队列中（从recieve缓冲区中读取消息进行发送），topology.receiver.buffer.size参数用于指定接收线程的缓冲区大小，同样的，消息发送线程负责从所有执行器发送线程共享的transfer缓冲区中读取消息，并将消息发送给网络上的其他消息消费者，transfer缓冲区的大小通过topology.transfer.buffer.size参数控制。

“topology.receiver.buffer.size”参数是工作进程中的接收线程批量向执行线程的输入队列发送数据时的缓冲区内消息数的最大值（接收线程从网络读取消息），此参数如果设置过大可能会造成一些问题（心跳线程挂掉然后吞吐率直线下降），默认值是8条消息，设置的值必须是2的幂（N次方），这是为了兼容LMAX Disruptor组件。
// 示例: 通过Java API配置
Config conf = new Config();
conf.put(Config.TOPOLOGY_RECEIVER_BUFFER_SIZE, 16); // 默认值为8

topology.receiver.buffer.size参数不是配置LMAX Disruptor队列的大小，它是配置的一个ArrayList的长度，这个List用来作为输入消息的缓冲，它不需要被多个线程访问，它仅仅是工作进程所专有的，但是因为这个List的元素最终被用来填充基于Disruptor的队列（执行器输入对列），所以这个参数必须是2的幂，参考backtype.storm.messaging.loader.clj的launch-receive-thread!详细信息。

使用“topology.transfer.buffer.size”参数配置的transfer缓冲区中的每一个元素实际上是一个tuple的列表，多个Executor中的消息发送线程将批量地把消息从outgoing发送到多个Executor共享的transfer缓冲区（Executor中包含用户逻辑线程和消息发送线程），transfer缓冲区的大小默认是1024个元素。
// 示例: 通过JavaAPI配置
conf.put(Config.TOPOLOGY_TRANSFER_BUFFER_SIZE, 32); // 默认值为1024

执行器（Executors）

每个工作进程控制着一个或多个执行器线程，每个执行器都有一个输入队列（incoming queue）和一个输入队列（outcoming queue），如上所述，工作进程有它自己专有的消息接收线程，该线程将（从网络接收到的）消息发送到执行器线程的输入消息队列（在选择发送到哪个执行器线程的输入消息队列时，应该是会存在一定的算法进行选择）。同样的，每个执行器都有它自己专有的发送线程，该线程负责将本执行器的输出消息从输出消息队列发送到执行器所在的工作进程的transfer缓冲区，执行器的输入队列和输出队列的大小分别通过“topology.executor.receive.buffer.size”和“topology.executor.send.buffer.size”参数进行配置。

每个执行器都有一个单独的线程用于处理Spout或Bolt中的用户逻辑，另外还有一个发送线程将消息从执行器的消息输出队列批量发送到工作进程的transfer缓冲区。

topology.executor.receive.buffer.size，该参数是执行器的输入消息队列的配置参数，队列的每个元素是一个tuple列表，tuple被批量地加入到队列元素中，此参数的默认配置是1024个元素，修改配置时的值必须是2的幂（适配于LMAX Disruptor）。
// 示例: 通过Java API配置
conf.put(Config.TOPOLOGY_EXECUTOR_RECEIVE_BUFFER_SIZE, 16384); // 批量加入tuple; 默认值是1024
topology.executor.send.buffer.size，该参数是执行器输入消息队列的配置参数，这个队列的每个元素是一个单独的tuple，默认值为1024，修改配置时的值必须是2的幂（适配于LMAX Disruptor）。
// 示例: 通过Java API配置
conf.put(Config.TOPOLOGY_EXECUTOR_SEND_BUFFER_SIZE, 16384); // 单独的tuple; 默认值是1024

更进一步

如何配置Storm的内部消息缓冲

上面提到的各种默认参数配置都可以在 conf/defaults.yaml中找到，你可以通过配置Storm集群的conf/storm.yaml文件全部覆盖这些默认配置信息，你也可以在创建一个Topology时通过Java API中的 backtype.storm.Config 类单独地配置这些参数。

如何配置Storm的parallelism（并行）参数（实际上就是配置集群中工作进程数，各个Spout和Bolt的实例数和线程数）

正确配置Storm集群的消息缓冲与集群的工作负载模式以及Storm集群的并行配置紧密相关，关于Storm集群的并行配置可以看看这篇文章： Understanding the Parallelism of a Storm Topology

搞明白Storm的拓扑内部在做什么

Storm UI是你观察运行中集群的各项关键指标的一个好的入口，比如：它向你说明了一个Spout或Bolt所谓的“容量”，这各项运行指标可以帮助你决定是否需要修改本文提到的各项与缓冲相关的参数的配置，这些参数的变动将会影响集群的计算性能，这篇文章可以看看： Running a Multi-Node Storm Cluster

除了这些，你还可以使用像Graphite这样的工具生成各项运行指标并监控这些指标，可以看看下面两篇我写的文章：

Sending Metrics From Storm to Graphite
Installing and Running Graphite via RPM and Supervisord

ooyala放在Github上的metrics_storm项目也是值得参考的（但我还没有用过这个工具）。

性能优化的建议

可以看看Nathan Marz（Storm作者）的演讲： Tuning and Productionization of Storm.

最开始，可以试试如下参数配置，看看是否能够提升Storm集群的性能。

 
       1 
       conf.put(Config.TOPOLOGY_RECEIVER_BUFFER_SIZE, 8); 
      
       2 
       conf.put(Config.TOPOLOGY_TRANSFER_BUFFER_SIZE, 32); 
      
       3 
       conf.put(Config.TOPOLOGY_EXECUTOR_RECEIVE_BUFFER_SIZE, 16384); 
      
       4 
       conf.put(Config.TOPOLOGY_EXECUTOR_SEND_BUFFER_SIZE, 16384);

转载自并发编程网 - ifeve.com

微信关注我们

原文链接：https://yq.aliyun.com/articles/86119

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

Cloudera将于4月底登陆纽交所，基于Hadoop的商业公司前景何在？

虽然名字里有一个“云”，Cloudera并不是一个真正的云公司。它是一个建立在开放源代码Hadoop项目之上的商业产品，hadoop的公司吉祥物是一只可爱的大象。在数据量越来越大的时代（业界常称为“大数据”），Hadoop把数据处理变得越来越方便。正如我们所看到的，今年第一季度，高科技企业IPO突然风生水起，选择在这样的时机上市，也算顺理成章。而在2015年英特尔首脑会议上，当Cloudera CEO Tom Reilly被问到公司上市时间表时，他还称，他愿意等到“我们所认为的最好时机”。其实Cloudera当时的体量足够上市，但刚刚筹集了超过10亿美元，拥有大量现金，因此他并不急于上市。基于Hadoop打包产品商业公司 Cloudera与Hortonworks的区别 ▼ 尽管Hadoop可能是处理大数据的有力方式且它可以开源

2017-05-19

671

机器学习库（MLlib）指南 MLlib是Spark的机器学习（ML）库。旨在简化机器学习的工程实践工作，并方便扩展到更大规模。MLlib由一些通用的学习算法和工具组成，包括分类、回归、聚类、协同过滤、降维等，同时还包括底层的优化原语和高层的管道API。 MLllib目前分为两个代码包： spark.mllib包含基于RDD的原始算法API。 spark.ml则提供了基于DataFrames高层次的API，可以用来构建机器学习管道。我们推荐您使用spark.ml，因为基于DataFrames的API更加的通用而且灵活。不过我们也会继续支持spark.mllib包。用户可以放心使用，spark.mllib还会持续地增加新的功能。不过开发者需要注意，如果新的算法能够适用于机器学习管道的概念，就应该将其放到spark.ml包中，如：特征提取器和转换器。下面的列表列出了两个包的主要功能。 spark.mllib: 数据类型，算法以及工具 Data types（数据类型） Basic statistics（基础统计） summary statistics（摘要统计） correlations...

2017-05-19

591

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。