【Java】【Flume】Flume-NG阅读源代码AvroSink

　　org.apache.flume.sink.AvroSink是用来通过网络来数据传输的。能够将event发送到RPCserver（比方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()链接RPCserver。

　　实际的链接在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。

　　三、process()方法。代码例如以下：

1   public Status process() throws EventDeliveryException {
2     Status status = Status.READY;
3     Channel channel = getChannel();    //获得channel
4     Transaction transaction = channel.getTransaction();    //创建事务
5
6     try {
7       transaction.begin();    //事务開始
8
9       verifyConnection();    //确保存在链接且处于活动状态，假设链接处于非活动状态销毁并重建链接
10
11       List<Event> batch = Lists.newLinkedList();
12
13       for (int i = 0; i < client.getBatchSize(); i++) {    //保证这批次的event数量不可能超过客户端批量处理的最大处理数量
14         Event event = channel.take();
15
16         if (event == null) {        //表示channel中没有数据了
17           break;
18         }
19
20         batch.add(event);    //增加event列表
21       }
22
23       int size = batch.size();    //获取这批次取得的event的数量
24       int batchSize = client.getBatchSize();        //获取客户端能够批量处理的大小
25
26       if (size == 0) {
27         sinkCounter.incrementBatchEmptyCount();
28         status = Status.BACKOFF;
29       } else {
30         if (size < batchSize) {
31           sinkCounter.incrementBatchUnderflowCount();
32         } else {
33           sinkCounter.incrementBatchCompleteCount();
34         }
35         sinkCounter.addToEventDrainAttemptCount(size);
36         client.appendBatch(batch);        //批量处理event
37       }
38
39       transaction.commit();        //事务提交
40       sinkCounter.addToEventDrainSuccessCount(size);
41
42     } catch (Throwable t) {
43       transaction.rollback();    //事务回滚
44       if (t instanceof Error) {
45         throw (Error) t;
46       } else if (t instanceof ChannelException) {
47         logger.error("Rpc Sink " + getName() + ": Unable to get event from" +
48             " channel " + channel.getName() + ". Exception follows.", t);
49         status = Status.BACKOFF;
50       } else {
51         destroyConnection();        //销毁链接
52         throw new EventDeliveryException("Failed to send events", t);
53       }
54     } finally {
55       transaction.close();    //事务关闭
56     }
57
58     return status;
59   }

　　即使本批次event的数量达不到client.getBatchSize()(channel中没数据了)也会马上发送到RPCserver。verifyConnection()方法是确保存在链接且处于活动状态。假设链接处于非活动状态销毁并重建链接。

假设本批次没有event，则不会想RPC发送不论什么数据。client.appendBatch(batch)方法是批量发送event。

　　(1)NettyAvroRpcClient.appendBatch(batch)方法会调用appendBatch(events, requestTimeout, TimeUnit.MILLISECONDS)方法，该方法会首先确认链接处于READY状态，否则报错；然后将每一个event又一次封装成AvroFlumeEvent，放入avroEvents列表中；然后构造一个CallFuture和avroEvents一同封装成一个Callable放入线程池 handshake = callTimeoutPool.submit(callable)中去运行，其call方法内容是avroClient.appendBatch(avroEvents,
callFuture)就是在此批量提交到RPCserver。然后handshake.get(connectTimeout, TimeUnit.MILLISECONDS)在规定时间等待运行的返回结果以及等待append的完毕waitForStatusOK(callFuture, timeout, tu)。具体的可看这里Flume的Avro
Sink和Avro Source研究之二 ： Avro Sink 。有对于这两个future更深入的分析。

一个批次传输的event的数量是min(batchSize,events.size())

　　(2)FailoverRpcClient.appendBatch(batch)方法会做最多maxTries次尝试直到获取到能够正确发送events的Client，通过localClient=getClient()--》getNextClient()来获取client，这种方法每次会获取hosts中的下一个HostInfo，并使用NettyAvroRpcClient来作为RPC Client，这就又回到了(1)中。这种方法另一个要注意的就是会先从当前的lastCheckedhost+1位置向后找能够使用的Client，假设不行会再从開始到到lastCheckedhost再找，再找不到就报错。使用localClient.appendBatch(events)来处理events。可參考(1)。

　　(3)LoadBalancingRpcClient.appendBatch(batch)方法，首先会获取能够发送到的RPCserver的迭代器Iterator<HostInfo> it = selector.createHostIterator()。然后取一个HostInfo,RpcClient client = getClient(host)这个Client和(2)一样都是NettyAvroRpcClient。可是getClient方法会设置一个保存名字和client映射的clientMap；client.appendBatch(events)运行之后就会跳出循环，下一次appendBatch会选择下一个client运行。

　　(4)ThriftRpcClient.appendBatch(batch)方法，从connectionManager.checkout()获取一个client。ConnectionPoolManager类主要维护俩对象availableClients用来存放可用的client(是一个ClientWrapper。维护一个ThriftSourceProtocol.Client client 是用来批量处理event的)、checkedOutClients用来存储从availableClients中拿出的Client表示正在使用的Client；ConnectionPoolManager.checkout()用于从availableClients中remove出client并放入checkedOutClients中，返回这个client。ConnectionPoolManager.checkIn(ClientWrapper
client)方法用于将指定的Client从checkedOutClient中remove出并放入availableClients中；ConnectionPoolManager.destroy(ClientWrapper client)用于将checkedOutClients中的指定Client remove并close。appendBatch方法中获得client后，会每次封装min(batchSize,events.size())个event，把他们封装成ThriftFlumeEvent增加thriftFlumeEvents列表，然后假设thriftFlumeEvents>0则运行doAppendBatch(client,
thriftFlumeEvents).get(requestTimeout,TimeUnit.MILLISECONDS)堵塞等待传输完成。

doAppendBatch方法会构建一个Callable其call方法运行client.client.appendBatch(e)。将这个Callable放入线程池callTimeoutPool中运行并返回运行结果Future。

　　以上四种RpcClient的append(Event event)方法也比較easy理解，不再讲述。

　　四、stop()方法主要是销毁链接，关闭cxnResetExecutor。

　　

　　事实上flume支持avro和thrift两种(眼下)传输，上面的(2)和(3)仅仅只是是对(1)的上层业务做了一次封装而已，本质上还它们是相同的avro(基于netty)。还记得在同一时间avrosink它支持压缩。