Flume-NG源码阅读:AvroSink

org.apache.flume.sink.AvroSink是用来通过网络来传输数据的,可以将event发送到RPC服务器(比如AvroSource),使用AvroSink和AvroSource可以组成分层结构。它继承自AbstractRpcSink  extends AbstractSink implements Configurable这跟其他的sink一样都得extends AbstractSink implements Configurable,所以重点也在confgure、start、process、stop这四个方法,实现了initializeRpcClient(Properties props)方法。

一、configure(Context context)方法,先获取配置文件中的主机hostname和端口port;设置clientProps的属性hosts=h1,hosts.h1=hostname:port;然后将配置信息中的所有信息放入clientProps中;获取cxnResetInterval表示重复建立连接的时间间隔,默认是0就是不重复建立连接。

二、start()方法是调用createConnection()建立连接,如果出现异常就调用destroyConnection()掐断连接,避免资源泄漏。createConnection()方法主要是初始化client = initializeRpcClient(clientProps)以及创建一个线程,并执行在给定延迟cxnResetInterval后执行一次销毁链接destroyConnection(),由于默认cxnResetInterval=0,所以是不会执行这个线程的。这点不是很明白,为什么要销毁???initializeRpcClient(clientProps)方法会根据配置文件中的信息进行构造相应的RpcClient:首先会获取"client.type"参数指定的类型可用的有四种(NettyAvroRpcClient(如果没有"client.type"则使用这个作为默认Client)、FailoverRpcClient、LoadBalancingRpcClient、ThriftRpcClient),实例化之后需要对其在进行必要的配置执行client.configure(properties)进行配置:

(1)NettyAvroRpcClient.configure(Properties properties)方法首先会获取锁,检查connState连接状态要保证是没有配置过的;其次获取"batch-size"设置batchSize,如果配置的小于1则使用默认值100;获取“hosts”,如果配置了多个hosts则只使用第一个;获取"hosts."前缀,如果有多个则使用第一个,再解析出hostname和port,构建一个InetSocketAddress的对象address;获取连接超时时间"connect-timeout",设置connectTimeout,如果配置的小于1000则使用默认值20000,单位是ms;获取相应时间"request-timeout",设置requestTimeout,如果配置的小于1000,则使用默认值20000,单位ms;获取压缩类型"compression-type",如果有配置压缩还需要获取压缩的等级compressionLevel;最后调用connect()链接RPC服务器。

实际的链接在connect(long timeout, TimeUnit tu)方法中,先构造一个线程池callTimeoutPool;然后根据是否有压缩构造相应的工厂类CompressionChannelFactory(有压缩配置)或者NioClientSocketChannelFactory(无压缩配置);构造一个

NettyTransceiver(this.address,socketChannelFactory,tu.toMillis(timeout))收发器对象transceiver;根据transceiver返回一个avroClient;最后设置链接状态为READY。

(2)FailoverRpcClient.configure(Properties properties)方法会调用configureHosts(Properties properties)方法,这个方法会获取配置文件中的host列表hosts;获取最大尝试次数"max-attempts",设置maxTries,默认是hosts的大小;获取批量大小

"batch-size",设置batchSize,如果配置的小于1则使用默认大小100;将此client置为活动的isActive=true。可以看出这个client可以使用多个host。

(3)LoadBalancingRpcClient.configure(Properties properties)会获取配置文件中的host列表hosts,且不允许少于两个,否则爆异常;获取主机选择器"host-selector",有两种内置的选择器:LoadBalancingRpcClient.RoundRobinHostSelector和LoadBalancingRpcClient.RandomOrderHostSelector,默认是ROUND_ROBIN(即RoundRobinHostSelector)轮询的方式(也可以自定义,要实现LoadBalancingRpcClient.HostSelector接口);获取"backoff",设置backoff(是否使用推迟算法,就是sink.process出问题后对这个sink设置惩罚时间,在此期间不再认为其可活动)的boolean值(默认false就是不启用);获取最大推迟时间"maxBackoff",设置maxBackoff;然后根据选择器是ROUND_ROBIN还是RANDOM选择对应的类并实例化selector,最后设置主机selector.setHosts(hosts)。

这两个内置选择器:RoundRobinHostSelector实际使用的是RoundRobinOrderSelector;RandomOrderHostSelector实际使用的是RandomOrderSelector,这两个都在Flume-NG源码阅读之SinkGroups和SinkRunner 这篇文章中有介绍,这里不再说明。

(4)ThriftRpcClient.configure(Properties properties)会获取状态锁stateLock.lock();获取配置文件中的host列表中的第一个,只需要一个;获取批量大小"batch-size",设置batchSize,如果配置的小于1则使用默认大小100;获取主机名hostname和端口port;获取响应时间requestTimeout,如果小于1000设置为默认的20000ms;获取连接池大小"maxConnections",设置connectionPoolSize,如果大小小于1则设置为默认的值5;创建连接池管理对象connectionManager= new ConnectionPoolManager(connectionPoolSize);设置连接状态为READY,connState = State.READY;最后状态锁解锁stateLock.unlock()。

这四个Client都是extends AbstractRpcClient implements RpcClient。

本栏目更多精彩内容:http://www.bianceng.cnhttp://www.bianceng.cn/webkf/tools/

时间: 2024-10-27 14:46:16

Flume-NG源码阅读:AvroSink的相关文章

Flume-NG源码阅读:HBaseSink

关于HBase的sink的所有内容均在org.apache.flume.sink.hbase包下. 每个sink包括自己定制的,都extends AbstractSink implements Configurable. 一.首先是configure(Context context)方法.该方法是对HBaseSink的参数初始化.主要包括以下几个: tableName:要写入的HBase数据表名,不能为空: columnFamily:数据表对应的列簇名,这个sink目前只支持一个列簇,不能为空:

淘宝数据库OceanBase SQL编译器部分 源码阅读--Schema模式

淘宝数据库OceanBase SQL编译器部分 源码阅读--Schema模式 什么是Database,什么是Schema,什么是Table,什么是列,什么是行,什么是User?我们可以可以把Database看作是一个大仓库,仓库分了很多很多的房间,Schema就是其中的房间,一个Schema代表一个房间,Table可以看作是每个Schema中的柜子,行和列就是柜子中的格子.User就是房间的主人.简单来说,Schema是包括表,列,索引,视图等数据库对象的集合. OceanBase中的强Sche

淘宝数据库OceanBase SQL编译器部分 源码阅读--生成物理查询计划

SQL编译解析三部曲分为:构建语法树,制定逻辑计划,生成物理执行计划.前两个步骤请参见我的博客<<淘宝数据库OceanBase SQL编译器部分 源码阅读--解析SQL语法树>>和<<淘宝数据库OceanBase SQL编译器部分 源码阅读--生成逻辑计划>>.这篇博客主要研究第三步,生成物理查询计划. 一. 什么是物理查询计划 与之前的阅读方法一致,这篇博客的两个主要问题是what 和how.那么什么是物理查询计划?物理查询计划能够直接执行并返回数据结果数

CI框架源码阅读笔记2 一切的入口 index.php

上一节(CI框架源码阅读笔记1 - 环境准备.基本术语和框架流程)中,我们提到了CI框架的基本流程,这里再次贴出流程图,以备参考: 作为CI框架的入口文件,源码阅读,自然由此开始.在源码阅读的过程中,我们并不会逐行进行解释,而只解释核心的功能和实现. 1. 设置应用程序环境 define("ENVIRONMENT", "development"); 这里的development可以是任何你喜欢的环境名称(比如dev,再如test),相对应的,你要在下面的switch

Flume-NG源码阅读:SourceRunner及选择器selector和拦截器interceptor的执行

在AbstractConfigurationProvider类中loadSources方法会将所有的source进行封装成SourceRunner放到了Map<String, SourceRunner> sourceRunnerMap之中.相关代码如下: Map<String, String> selectorConfig = context.getSubProperties( BasicConfigurationConstants.CONFIG_SOURCE_CHANNELSEL

java源码阅读方法以及经验

问题描述 java源码阅读方法以及经验 如何更好的阅读java源码,更注重阅读哪些包里面的源码,当然连好的阅读源码的工具也说明一下更好了 解决方案 我在这里假设你在问怎么阅读jdk的源码,java源码这个名字有点奇怪. 你可以build 一个fast debug版本,然后使用debugger去调试你的程序,这样对程序是怎么调用的有很直观的视图. 其次,可以看看jdk里面的regression tests,里面有很多例子. 其次,openjdk提供了netbean的jdk project,你可以很

淘宝数据库OceanBase SQL编译器部分 源码阅读--生成逻辑计划

淘宝数据库OceanBase SQL编译器部分 源码阅读--生成逻辑计划 SQL编译解析三部曲分为:构建语法树,生成逻辑计划,指定物理执行计划.第一步骤,在我的上一篇博客淘宝数据库OceanBase SQL编译器部分 源码阅读--解析SQL语法树里做了介绍,这篇博客主要研究第二步,生成逻辑计划. 一. 什么是逻辑计划? 我们已经知道,语法树就是一个树状的结构组织,每个节点代表一种类型的语法含义.如 update student set sex="M" where name ="

java 源码-Java项目源码阅读技巧

问题描述 Java项目源码阅读技巧 拿到一个项目的源代码,没有项目文档,注释很少,项目用的struts,hibernate,spring框架,该如何快速掌握整个项目的脉络,一点头绪都没有不知从哪下手!求大神指点 解决方案 把项目先运行起来,看看页面大致显示的什么内容. 然后再根据需求去熟悉对于的源码.

Apache Storm源码阅读笔记&amp;OLAP在大数据时代的挑战

 <一>Apache Storm源码阅读笔记 楔子 自从建了Spark交流的QQ群之后,热情加入的同学不少,大家不仅对Spark很热衷对于Storm也是充满好奇.大家都提到一个问题就是有关storm内部实现机理的资料比较少,理解起来非常费劲. 尽管自己也陆续对storm的源码走读发表了一些博文,当时写的时候比较匆忙,有时候衔接的不是太好,此番做了一些整理,主要是针对TridentTopology部分,修改过的内容采用pdf格式发布,方便打印. 文章中有些内容的理解得益于徐明明和fxjwind两