《Storm企业级应用:实战、运维和调优》——3.2 Spout数据源

简介:

本节书摘来自华章计算机《Storm企业级应用:实战、运维和调优》一书中的第3章,第3.2节,作者:马延辉 陈书美 雷葆华著, 更多章节内容可以访问云栖社区“华章计算机”公众号查看。

3.2 Spout数据源

消息源Spout是Storm的Topology中的消息生产者(即Tuple的创造者)。
3.2.1 Spout介绍
1.?Spout的结构
Spout是Storm的核心组件之一,最源头的接口是IComponent,如图3-2所示,几个Spout接口都继承自IComponent。

image

2.?Spout发出的消息
Spout从外部获取数据后,向Topology中发出的Tuple可以是可靠的,也可以是不可靠的。
一个可靠的消息源可以重新发射一个Tuple(如果该Tuple没有被Storm成功处理),但是一个不可靠的消息源Spout一旦发出,一个Tuple就把它彻底“遗忘”,也就不可能再发了。
3.?Spout发射的流
Spout可以发射多个流。要达到这样的效果,使用OutputFieldsDeclarer.declareStream来定义多个流(即定义多个Stream),然后使用SpoutOutputCollector来发射指定的流。
4.?Spout的重要方法
Spout的重要方法是nextTuple。nextTuple方法发射一个新的元组到Topology,如果没有新元组发射,则直接返回。注意任务Spout的nextTuple方法都不要实现成阻塞的,因为Storm是在相同的线程中调用Spout的方法。Spout的另外两个重要方法是ack和fail方法。当Spout发射的元组被拓扑成功处理时,调用ack方法;当处理失败时,调用fail方法。ack和fail方法仅被可靠的Spout调用。
5.?Spout的组件
Spout的最顶层抽象是ISpout接口。在通常情况下(Shell和事务型的除外),实现一个Spout,可以直接实现接口IRichSpout,如果不想写多余的代码,可以直接继承BaseRichSpout。
3.2.2 Spout实例
下面通过创建一个实例RandomSpout来介绍Spout,图3-3为RandomSpout继承自BasicRichSpout及其实现的原理图。
图3-4列出了实例RandomSpout继承自BaseRichSpout中的几个重要方法。
下面对图3-4中的方法进行详细介绍。
(1)open方法
当一个Task被初始化时会调用此open方法。一般都会在此方法中初始化发送Tuple的对象SpoutOutputCollector和配置对象TopologyContext。
代码示例如下:

public void open(Map conf, TopologyContext context,   SpoutOutputCollector collector) {
      this.collector = collector;
      random = new Random();   
}

image

(2)declareOutputFields方法
此方法用于声明当前Spout的Tuple发送流。流的定义是通过OutputFieldsDeclare.declareStream方法完成的,其中的参数包括了发送的域Fields。
示例代码如下:

public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("randomInt"));
}

(3)nextTuple方法
这是Spout类中最重要的一个方法。发射一个Tuple到Topology都是通过该方法来实现的。
示例代码如下:

public void nextTuple() {
      while(true){
          Values val = new Values(random.nextInt(100));
          collector.emit(val);
          try {
              Thread.sleep(500);
          } catch (InterruptedException e) {
              e.printStackTrace();
          }
      }      
  }

以上代码从100以内的整数中随机生成一个数作为Tuple的值,然后通过_collector发送到Topology。
另外,除了上述几个方法之外,还有getComponentConf?iguration、ack、fail和close方法等。getComponentConf?iguration方法用于配置当前组件的参数,Storm监测到一个Tuple被成功处理时调用ack方法,处理失败时调用fail方法,这两个方法在BaseRichSpout类中已经被隐式-实现了。

相关实践学习
深入解析Docker容器化技术
Docker是一个开源的应用容器引擎,让开发者可以打包他们的应用以及依赖包到一个可移植的容器中,然后发布到任何流行的Linux机器上,也可以实现虚拟化,容器是完全使用沙箱机制,相互之间不会有任何接口。Docker是世界领先的软件容器平台。开发人员利用Docker可以消除协作编码时“在我的机器上可正常工作”的问题。运维人员利用Docker可以在隔离容器中并行运行和管理应用,获得更好的计算密度。企业利用Docker可以构建敏捷的软件交付管道,以更快的速度、更高的安全性和可靠的信誉为Linux和Windows Server应用发布新功能。 在本套课程中,我们将全面的讲解Docker技术栈,从环境安装到容器、镜像操作以及生产环境如何部署开发的微服务应用。本课程由黑马程序员提供。     相关的阿里云产品:容器服务 ACK 容器服务 Kubernetes 版(简称 ACK)提供高性能可伸缩的容器应用管理能力,支持企业级容器化应用的全生命周期管理。整合阿里云虚拟化、存储、网络和安全能力,打造云端最佳容器化应用运行环境。 了解产品详情: https://www.aliyun.com/product/kubernetes
相关文章
|
存储 弹性计算 运维
可观测性体系问题之Event Center对异构数据源的支持如何解决
可观测性体系问题之Event Center对异构数据源的支持如何解决
245 0
|
SQL 运维 监控
【运维知识进阶篇】Zabbix5.0稳定版详解11(在Grafana中使用Zabbix插件:安装Grafana+安装Zabbix插件+添加数据源+Grafana直连MySQL数据库取值)(下)
【运维知识进阶篇】Zabbix5.0稳定版详解11(在Grafana中使用Zabbix插件:安装Grafana+安装Zabbix插件+添加数据源+Grafana直连MySQL数据库取值)(下)
629 1
|
运维 监控 数据可视化
【运维知识进阶篇】Zabbix5.0稳定版详解11(在Grafana中使用Zabbix插件:安装Grafana+安装Zabbix插件+添加数据源+Grafana直连MySQL数据库取值)(上)
【运维知识进阶篇】Zabbix5.0稳定版详解11(在Grafana中使用Zabbix插件:安装Grafana+安装Zabbix插件+添加数据源+Grafana直连MySQL数据库取值)
1493 0
|
数据采集 机器学习/深度学习 人工智能
运维人的“福音”?AI 驱动的自动化网络监控到底香不香!
运维人的“福音”?AI 驱动的自动化网络监控到底香不香!
1451 0
|
人工智能 运维 监控
运维安全还能靠“人盯人”?别闹了,聊聊自动化处理的真功夫
运维安全还能靠“人盯人”?别闹了,聊聊自动化处理的真功夫
432 17
|
机器学习/深度学习 人工智能 运维
运维不背锅,从“自动修锅”开始:AI自动化运维是怎么回事?
运维不背锅,从“自动修锅”开始:AI自动化运维是怎么回事?
884 49
|
人工智能 运维 安全
运维老哥的救星?AI 驱动的自动化配置管理新趋势
运维老哥的救星?AI 驱动的自动化配置管理新趋势
569 11
|
运维 Prometheus 监控
系统崩了怪运维?别闹了,你该问问有没有自动化!
系统崩了怪运维?别闹了,你该问问有没有自动化!
427 9
|
机器学习/深度学习 运维 监控
智能运维Agent:自动化运维的新范式
在数字化转型浪潮中,智能运维Agent正重塑运维模式。它融合人工智能与自动化技术,实现从被动响应到主动预防的转变。本文详解其四大核心功能:系统监控、故障诊断、容量规划与安全响应,探讨如何构建高效、可靠的自动化运维体系,助力企业实现7×24小时无人值守运维,推动运维效率与智能化水平全面提升。
3020 0
|
运维 监控 应用服务中间件
运维打铁: Ruby 脚本在运维自动化中的应用探索
Ruby 是一种简洁、动态类型的编程语言,适合运维自动化任务。本文介绍了其在服务器配置管理、定时任务执行和日志分析处理中的应用,并提供了代码示例,展示了 Ruby 在运维自动化中的实际价值。
695 2

热门文章

最新文章