设为首页 收藏本站
查看: 1434|回复: 0

[经验分享] Flume-NG源码阅读之SinkGroups和SinkRunner

[复制链接]

尚未签到

发表于 2015-9-17 05:51:28 | 显示全部楼层 |阅读模式
  在AbstractConfigurationProvider类中loadSinks方法会调用loadSinkGroups方法将所有的sink和sinkgroup放到了Map<String, SinkRunner> sinkRunnerMap之中。
  SinkRunner可能对应一个sink也可能对应一个sinkgroup。因为如果配置文件中有sinkgroup则这个sinkgroup对应的sink会组成一个group然后封装为一个sinkRunner,然后不在sinkgroup中的sink会自己成为一个sinkRunner。每个SinkRunner的构造方法的参数是一个SinkProcessor是用来处理多个sink的。
  一、如果一个SinkRunner对应一个sink。SinkProcessor pr = new DefaultSinkProcessor()是默认的SinkProcessor。loadSinkGroups方法中的相关代码如下:



1           SinkProcessor pr = new DefaultSinkProcessor();
2           List<Sink> sinkMap = new ArrayList<Sink>();
3           sinkMap.add(entry.getValue());
4           pr.setSinks(sinkMap);
5           Configurables.configure(pr, new Context());
6           sinkRunnerMap.put(entry.getKey(),new SinkRunner(pr));   
  DefaultSinkProcessor.configure(Context context)是空方法,所有这句Configurables.configure(pr, new Context())没啥作用,重要的是DefaultSinkProcessor.process()方法,就一句代码就是return sink.process()直接调用sink的process。setSinks方法只会设置一个sink。DefaultSinkProcessor的start()方法也会直接调用sink.start()来启动sink。
  二、如果一个SinkRunner对应多个sink。则会构造一个SinkGroup group = new SinkGroup(groupSinks)然后获取SinkProcessor:group.getProcessor()。loadSinkGroups方法中的相关代码如下:



1           SinkGroup group = new SinkGroup(groupSinks);
2           Configurables.configure(group, groupConf);
3           sinkRunnerMap.put(comp.getComponentName(), new SinkRunner(group.getProcessor()));   
  Configurables.configure(group, groupConf)会调用SinkGroupConfiguration.configure(Context context),该方法会获取配置文件中关于"processor."的属性通过getKnownSinkProcessor方法获取SinkProcessorType(是FailoverSinkProcessor或者是LoadBalancingSinkProcessor),并执行该SinkProcessor.configure(processorContext)进行实例化和配置。
  1、如果SinkProcessor是LoadBalancingSinkProcessor,这是负载均衡的processor,会将channel中的发送到指定的所有sink。通过配置选择器selector来选择何种方式的负载均衡,1.3有两种:ROUND_ROBIN,轮询,就是轮流向channel发送数据;RANDOM,随机选择channel发送数据。只有这个SinkProcessor有选择器。
  (1)configure(Context context)方法。先获取选择器selector的类型,默认是ROUND_ROBIN,轮询;获取backoff(是否使用推迟算法,就是sink.process出问题后对这个sink设置惩罚时间,在此期间不再认为其可活动)的boolean值(默认false就是不启用);根据类型构造相应的选择器对象RoundRobinSinkSelector(实际上会构造一个RoundRobinOrderSelector)或者RandomOrderSinkSelector(实际上会构造一个RandomOrderSelector);然后实例化并设置sinks;最后对selector执行其configure(context)方法进行初始化。
  A、RoundRobinSinkSelector的实际操作者是RoundRobinOrderSelector extends OrderSelector,它实现了createIterator()方法,该方法用来选出所有的sink及其可活动sinkl的索引封装成一个SpecificOrderIterator<T>(indexOrder, getObjects())并返回,可以通过SpecificOrderIterator.hasNext()方法判断是否还有sink,用next()方法获取下一个sink。这样可以按照索引递增的顺序依次获取sink进行操作。SpecificOrderIterator主要是将两个:一个是索引数组,一个是sink列表。createIterator()方法代码如下:



1 @Override
2   public Iterator<T> createIterator() {
3     List<Integer> activeIndices = getIndexList();  //会获取最新的活动的sink的索引列表
4     int size = activeIndices.size();
5     // possible that the size has shrunk so gotta adjust nextHead for that
6     if (nextHead >= size) {    //可能会出现sink的总数调整,所以总得getIndexList()并调整nextHead
7       nextHead = 0;
8     }
9     int begin = nextHead++;    //注意++在后面说明是先赋值,在自加
10     if (nextHead == activeIndices.size()) {
11       nextHead = 0;
12     }
13
14     int[] indexOrder = new int[size];
15
16     for (int i = 0; i < size; i++) {
17       indexOrder = activeIndices.get((begin + i) % size);
18     }
19
20     return new SpecificOrderIterator<T>(indexOrder, getObjects());    //组成两个数组,大小都一样
21   }
  createIterator()方法中总会调用getIndexList()方法,因为可能有sink中断,或者sinkgroup再调整等情况,使得sinkgroup中实际活动的sink数产生变化。nextHead始终指向下一个可活动的sink索引,这样就可以保证轮询。indexOrder是新的活动sink的索引数组;getObjects()则返回所有sink的List,通过索引即可即可获取此List中对应的sink。
  B、RandomOrderSinkSelector的实际操作者是RandomOrderSelector extends OrderSelector,它实现了createIterator()方法:



1 public synchronized Iterator<T> createIterator() {
2     List<Integer> indexList = getIndexList();
3
4     int size = indexList.size();
5     int[] indexOrder = new int[size];
6     //indexList由于remove操作会动态变化,所以一直使用indexList.size()会获得实际大小
7     while (indexList.size() != 1) {
8       int pick = random.nextInt(indexList.size());
9       indexOrder[indexList.size() - 1] = indexList.remove(pick);    //取出pick位置的索引,这句总是使得indexOrder从后向前插入数据
10     }
11
12     indexOrder[0] = indexList.get(0);    //将最后一个索引放入indexOrder
13
14     return new SpecificOrderIterator<T>(indexOrder, getObjects());
15   }
  这个方法最终只是将"可活动"的sink的顺序按随机的方式打乱了而已。注意的一个是总是调用indexList.size()动态获取最新的大小;一个是indexOrder[indexList.size() - 1]始终是从后向前插入数据;一个是indexOrder[0] = indexList.get(0)将最后一个插入保证完整性。
  A和B都是OrderSelector抽象类的子类,都只实现了createIterator()方法,对于getIndexList()和sink.process()方法出现错误的时的selector.informSinkFailed(sink)都是一样的这两个方法决定了出现问题的sink的推迟时间,如果要修改推迟时间可以重写这两个方法。当sink.process运行出问题时informSinkFailed会更新对应sink的FailureState(就三个数,sequentialFails记录出错次数、restoreTime记录出错后惩罚恢复时间(在此期间不再认为是可活动的sink,通过getIndexList()来过滤)、lastFail记录上一次出错时间,三个初始化都是0),maxTimeout默认是3000。看informFailure代码:



public void informFailure(T failedObject) {
if (!shouldBackOff) {        //不允许推迟
return;
}
FailureState state = stateMap.get(failedObject);
long now = System.currentTimeMillis();    //获取现在系统时间
long delta = now - state.lastFail;        //获取和上次失败时间之间的时间间隔
long lastBackoffLength = Math.min(maxTimeout, 1000 * (1 << state.sequentialFails));    //获取上一次要推迟的时间增量
long allowableDiff = lastBackoffLength + CONSIDER_SEQUENTIAL_RANGE;        //CONSIDER_SEQUENTIAL_RANGE=2000
if (allowableDiff > delta) {
if (state.sequentialFails < EXP_BACKOFF_COUNTER_LIMIT) {        //说明是连续失败
state.sequentialFails++;
}
} else {            //说明期间曾重新正确process,重新计数
state.sequentialFails = 1;
}
state.lastFail = now;
state.restoreTime = now + Math.min(maxTimeout, 1000 * (1 << state.sequentialFails));
}
  这个informFailure方法有需要说明的地方:如何判断是连续失败?关键在于CONSIDER_SEQUENTIAL_RANGE这个变量是宽限期,等于2000,因为首先在惩罚时间内是“不被认可”的,是不被认为是可活动的,所以超过惩罚时间后自然会被重新认为是活动的,如果在“惩罚期+宽限期”=allowableDiff,(allowableDiff > delta), 内再次失败说明是连续失败,所以在失败次数不超过EXP_BACKOFF_COUNTER_LIMIT(等于16)时就增加state.sequentialFails,一旦超过16就不再增加就是16;当allowableDiff <=delta成立时认为process至少一次成功但这次失败,需要重置state.sequentialFails为1。
  getIndexList()方法用来过滤“不被认可”的sink的索引。代码如下:



1 protected List<Integer> getIndexList() {
2     long now = System.currentTimeMillis();
3
4     List<Integer> indexList = new ArrayList<Integer>();
5
6     int i = 0;
7     for (T obj : stateMap.keySet()) {        //是sink的集合
8       if (!isShouldBackOff() || stateMap.get(obj).restoreTime < now) {    //
9         indexList.add(i);        //将索引存储
10       }
11       i++;
12     }
13     return indexList;
14   }
  如果shouldBackOff=true则会返回的列表将是所有的sink的索引。stateMap.get(obj).restoreTime < now这句会过滤掉当前还处在惩罚时间内不被认可的sink的索引。
  (2)start()方法会先启动AbstractSinkProcessor.start()方法将所有的sink启动(start()),然后启动选择器 selector.start()。
  (3)process()方法遍历sinkIterator一次获取可活动的sink,执行sink.process()方法,如果有异常就跳出循环并执行失败处理informSinkFailed,如果如果正常执行完毕就退出循环,循环就是找到第一个可以正常处理完毕的sink后退出。
  2、如果SinkProcessor是FailoverSinkProcessor,这是容错的processor,一旦有一个sink中断可以使用其他的代替。
  (1)setSinks(List<Sink> sinks)方法会将sinks列表中的所有sink,先调用父类的setSinks方法为的是可以执行父类的start和stop方法(子类中没有实现这俩方法),然后放入Map<String, Sink> sinks中。
  (2)configure(Context context)方法,会先获取中断时间的上限maxPenalty,然后将所有的sink及其对应的优先级放入liveSinks(这是一个TreeMap,默认根据键值的自然顺序排序存储),最后activeSink = liveSinks.get(liveSinks.lastKey())获取优先级最高的sink作为活动sink。failedSinks = new PriorityQueue<FailedSink>()是一个保存中断sink的一个优先级队列。
  (3)process()方法。循环执行如果failedSinks不为空并且记录惩罚时间小于当前系统时间,则取出failedSinks的head然后尝试执行getSink().process()如果能获取到Rady状态说明这个节点又重新建立了链接,则将其加入liveSinks,并重新获取优先级最高的sink作为activeSink,如果获取的是backOff状态则重新将其加入failedSinks,返回状态,如果出现异常则cur.incFails()重新记录惩罚时间并加入failedSinks,惩罚时间会动态变化,会根据失败的次数增加(会和设置的比较取较大者)。如果failedSinks为空或者当前系统小于惩罚时间则使用当前活动的sink:activeSink.process()。
  注:惩罚时间是动态变化的,会随着链接失败的次数而变化,失败次数越多到下次使用它的间隔越长。
  返回顶端sinkRunnerMap会在Application.startAllComponents方法中调用,放放到LifecycleSupervisor.supervise方法中去执行,最终会执行SinkRunner.start()方法来启动组件。



1 public void start() {
2     SinkProcessor policy = getPolicy();
3
4     policy.start();
5
6     runner = new PollingRunner();
7
8     runner.policy = policy;
9     runner.counterGroup = counterGroup;
10     runner.shouldStop = new AtomicBoolean();
11
12     runnerThread = new Thread(runner);
13     runnerThread.setName("SinkRunner-PollingRunner-" +
14         policy.getClass().getSimpleName());
15     runnerThread.start();
16
17     lifecycleState = LifecycleState.START;
18   }
  上述代码中的policy其实就是SinkProcessor,可能是LoadBalancingSinkProcessor、FailoverSinkProcessor、DefaultSinkProcessor三者中其中之一。policy.start()会启动SinkProcessor。然后会启动一个线程PollingRunner,该线程会始终执行policy.process()方法根据返回的状态做一些统计。这就是我们自定义也要实现process方法的所在,及其需要返回Status的原因。
  至此SinkGroup的介绍完结。

运维网声明 1、欢迎大家加入本站运维交流群:群②:261659950 群⑤:202807635 群⑦870801961 群⑧679858003
2、本站所有主题由该帖子作者发表,该帖子作者与运维网享有帖子相关版权
3、所有作品的著作权均归原作者享有,请您和我们一样尊重他人的著作权等合法权益。如果您对作品感到满意,请购买正版
4、禁止制作、复制、发布和传播具有反动、淫秽、色情、暴力、凶杀等内容的信息,一经发现立即删除。若您因此触犯法律,一切后果自负,我们对此不承担任何责任
5、所有资源均系网友上传或者通过网络收集,我们仅提供一个展示、介绍、观摩学习的平台,我们不对其内容的准确性、可靠性、正当性、安全性、合法性等负责,亦不承担任何法律责任
6、所有作品仅供您个人学习、研究或欣赏,不得用于商业或者其他用途,否则,一切后果均由您自己承担,我们对此不承担任何法律责任
7、如涉及侵犯版权等问题,请您及时通知我们,我们将立即采取措施予以解决
8、联系人Email:admin@iyunv.com 网址:www.yunweiku.com

所有资源均系网友上传或者通过网络收集,我们仅提供一个展示、介绍、观摩学习的平台,我们不对其承担任何法律责任,如涉及侵犯版权等问题,请您及时通知我们,我们将立即处理,联系人Email:kefu@iyunv.com,QQ:1061981298 本贴地址:https://www.yunweiku.com/thread-114550-1-1.html 上篇帖子: flume 转 下篇帖子: C++ Thrift Client 与 Flume Thrift Source 对接
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

扫码加入运维网微信交流群X

扫码加入运维网微信交流群

扫描二维码加入运维网微信交流群,最新一手资源尽在官方微信交流群!快快加入我们吧...

扫描微信二维码查看详情

客服E-mail:kefu@iyunv.com 客服QQ:1061981298


QQ群⑦:运维网交流群⑦ QQ群⑧:运维网交流群⑧ k8s群:运维网kubernetes交流群


提醒:禁止发布任何违反国家法律、法规的言论与图片等内容;本站内容均来自个人观点与网络等信息,非本站认同之观点.


本站大部分资源是网友从网上搜集分享而来,其版权均归原作者及其网站所有,我们尊重他人的合法权益,如有内容侵犯您的合法权益,请及时与我们联系进行核实删除!



合作伙伴: 青云cloud

快速回复 返回顶部 返回列表