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/