如果你还没看过Flume-ng源码解析系列中的启动流程、Channel组件和Sink组件,可以点击下面链接:

Flume-ng源码解析之启动流程

Flume-ng源码解析之Channel组件

Flume-ng源码解析之Sink组件

在前面三篇文章中我们初步了解了Flume的启动流程、Channel组件和Sink组件,接下来我们一起来看看agent三大组件中Source组件。

1 Source

Source,作为agent中的消息来源组件,我们来看看它是如何将event传递给channel的和它的特性。

依然先看代码:

@InterfaceAudience.Public
@InterfaceStability.Stable
public interface Source extends LifecycleAware, NamedComponent {
  public void setChannelProcessor(ChannelProcessor channelProcessor);
  public ChannelProcessor getChannelProcessor();
}

我们可以看到它里面定义的两个需要实现方法是getChannelProcessor和setChannelProcessor,我们大概可以猜到,source就是通过ChannelProcessor将event传输给channel的。

这里先来了解一下Source的类型,Flume根据数据来源的特性将Source分成两类类,像Http、netcat和exec等就是属于事件驱动型(EventDrivenSource),而kafka和Jms等就是属于轮询拉取型(PollableSource)。

据我们在启动流程中了解到的,Application是先启动SourceRunner,再由SourceRunner来启动source,那么既然source有两种类型,那么Sourcerunner也分为EventDrivenSourceRunner和PollableSourceRunner,我们来看看它们的start():

EventDrivenSourceRunner

public class EventDrivenSourceRunner extends SourceRunner {
  …
  @Override
  public void start() {
    Source source = getSource();
    ChannelProcessor cp = source.getChannelProcessor();
    cp.initialize();
    source.start();
    lifecycleState = LifecycleState.START;
  }
  …
}

PollableSourceRunner

public class PollableSourceRunner extends SourceRunner {

  …
  @Override
  public void start() {
    PollableSource source = (PollableSource) getSource();
    ChannelProcessor cp = source.getChannelProcessor();
    cp.initialize();
    source.start();

    runner = new PollingRunner();

    runner.source = source;
    runner.counterGroup = counterGroup;
    runner.shouldStop = shouldStop;

    runnerThread = new Thread(runner);
    runnerThread.setName(getClass().getSimpleName() + "-" +
        source.getClass().getSimpleName() + "-" + source.getName());
    runnerThread.start();

    lifecycleState = LifecycleState.START;
  }

  …
  public static class PollingRunner implements Runnable {

    private PollableSource source;
    private AtomicBoolean shouldStop;
    private CounterGroup counterGroup;

    @Override
    public void run() {
      logger.debug("Polling runner starting. Source:{}", source);

      while (!shouldStop.get()) {
        counterGroup.incrementAndGet("runner.polls");

        try {
          if (source.process().equals(PollableSource.Status.BACKOFF)) {
            counterGroup.incrementAndGet("runner.backoffs");

            Thread.sleep(Math.min(
                counterGroup.incrementAndGet("runner.backoffs.consecutive")
                * source.getBackOffSleepIncrement(), source.getMaxBackOffSleepInterval()));
          } else {
            counterGroup.set("runner.backoffs.consecutive", 0L);
          }
        } catch (InterruptedException e) {
          logger.info("Source runner interrupted. Exiting");
          counterGroup.incrementAndGet("runner.interruptions");
        } catch (EventDeliveryException e) {
          logger.error("Unable to deliver event. Exception follows.", e);
          counterGroup.incrementAndGet("runner.deliveryErrors");
        } catch (Exception e) {
          counterGroup.incrementAndGet("runner.errors");
          logger.error("Unhandled exception, logging and sleeping for " +
              source.getMaxBackOffSleepInterval() + "ms", e);
          try {
            Thread.sleep(source.getMaxBackOffSleepInterval());
          } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
          }
        }
      }

      logger.debug("Polling runner exiting. Metrics:{}", counterGroup);
    }

  }

}

无论是PollableSourceRunner还是EventDrivenSourceRunner,都是调用它里面的source的start()。这个时候我们看到ChannelProcessor的存在,那么就会有疑惑,这ChannelProcessor哪来的?我们还是得看回AbstarctConfigurationProvider,查看里面的loadSources(),我们就会发现下面这段代码:

ChannelSelectorConfiguration selectorConfig = config.getSelectorConfiguration();
ChannelSelector selector = ChannelSelectorFactory.create(sourceChannels, selectorConfig);
ChannelProcessor channelProcessor = new ChannelProcessor(selector);
Configurables.configure(channelProcessor, config);
source.setChannelProcessor(channelProcessor);

到这里我们基本已经了解了Source的启动流程,下面以AvroSource为例看看,source是在哪里调用ChannelProcessor的插入方法。

2 AvroSource

public class AvroSource extends AbstractSource implements EventDrivenSource,
    Configurable, AvroSourceProtocol {
  …
  @Override
  public Status append(AvroFlumeEvent avroEvent) {
    if (logger.isDebugEnabled()) {
      if (LogPrivacyUtil.allowLogRawData()) {
        logger.debug("Avro source {}: Received avro event: {}", getName(), avroEvent);
      } else {
        logger.debug("Avro source {}: Received avro event", getName());
      }
    }

    sourceCounter.incrementAppendReceivedCount();
    sourceCounter.incrementEventReceivedCount();

    Event event = EventBuilder.withBody(avroEvent.getBody().array(),
        toStringMap(avroEvent.getHeaders()));

    try {
      getChannelProcessor().processEvent(event);
    } catch (ChannelException ex) {
      logger.warn("Avro source " + getName() + ": Unable to process event. " +
          "Exception follows.", ex);
      return Status.FAILED;
    }

    sourceCounter.incrementAppendAcceptedCount();
    sourceCounter.incrementEventAcceptedCount();

    return Status.OK;
  }

  @Override
  public Status appendBatch(List<AvroFlumeEvent> events) {
    logger.debug("Avro source {}: Received avro event batch of {} events.",
        getName(), events.size());
    sourceCounter.incrementAppendBatchReceivedCount();
    sourceCounter.addToEventReceivedCount(events.size());

    List<Event> batch = new ArrayList<Event>();

    for (AvroFlumeEvent avroEvent : events) {
      Event event = EventBuilder.withBody(avroEvent.getBody().array(),
          toStringMap(avroEvent.getHeaders()));

      batch.add(event);
    }

    try {
      getChannelProcessor().processEventBatch(batch);
    } catch (Throwable t) {
      logger.error("Avro source " + getName() + ": Unable to process event " +
          "batch. Exception follows.", t);
      if (t instanceof Error) {
        throw (Error) t;
      }
      return Status.FAILED;
    }

    sourceCounter.incrementAppendBatchAcceptedCount();
    sourceCounter.addToEventAcceptedCount(events.size());

    return Status.OK;
  }
  …
}

在append方法中我们可以看到getChannelProcessor().processEvent(event);,所以不同的Source根据它的不同触发机制和拉取机制,在特定的时候调用ChannelProcessor来执行event的插入。 ·

到此为止,我们就完成了对Flume启动流程和三大组件的研究,鉴于能力,其中有些细节没办法深入研究,希望以后有时间能够继续深入分析下去。

Flume-ng源码解析之Source组件的更多相关文章

  1. Flume-ng源码解析之Channel组件

    如果还没看过Flume-ng源码解析之启动流程,可以点击Flume-ng源码解析之启动流程 查看 1 接口介绍 组件的分析顺序是按照上一篇中启动顺序来分析的,首先是Channel,然后是Sink,最后 ...

  2. Flume-ng源码解析之Sink组件

    作为启动流程中第二个启动的组件,我们今天来看看Sink的细节 1 Sink Sink在agent中扮演的角色是消费者,将event输送到特定的位置 首先依然是看代码,由代码我们可以看出Sink是一个接 ...

  3. .Net Core缓存组件(Redis)源码解析

    上一篇文章已经介绍了MemoryCache,MemoryCache存储的数据类型是Object,也说了Redis支持五中数据类型的存储,但是微软的Redis缓存组件只实现了Hash类型的存储.在分析源 ...

  4. .Net Core缓存组件(MemoryCache)源码解析

    一.介绍 由于CPU从内存中读取数据的速度比从磁盘读取快几个数量级,并且存在内存中,减小了数据库访问的压力,所以缓存几乎每个项目都会用到.一般常用的有MemoryCache.Redis.MemoryC ...

  5. admin源码解析以及仿照admin设计stark组件

    ---恢复内容开始--- admin源码解析 一 启动:每个APP下的apps.py文件中. 首先执行每个APP下的admin.py 文件. def autodiscover(): autodisco ...

  6. admin源码解析及自定义stark组件

    admin源码解析 单例模式 单例模式(Singleton Pattern)是一种常用的软件设计模式,该模式的主要目的是确保某一个类只有一个实例存在.当你希望在整个系统中,某个类只能出现一个实例时,单 ...

  7. Django 之 admin组件使用&amp;源码解析

    admin组件使用 Django 提供了基于 web 的管理工具. Django 自动管理工具是 django.contrib 的一部分.可以在项目的 settings.py 中的 INSTALLED ...

  8. Java集合---Array类源码解析

    Java集合---Array类源码解析              ---转自:牛奶.不加糖 一.Arrays.sort()数组排序 Java Arrays中提供了对所有类型的排序.其中主要分为Prim ...

  9. 【vuejs深入二】vue源码解析之一,基础源码结构和htmlParse解析器

    写在前面 一个好的架构需要经过血与火的历练,一个好的工程师需要经过无数项目的摧残. vuejs是一个优秀的前端mvvm框架,它的易用性和渐进式的理念可以使每一个前端开发人员感到舒服,感到easy.它内 ...

随机推荐

  1. [我给Unity官方视频教程做中文字幕]beginner Graphics – Lessons系列之纹理Textures

    [我给Unity官方视频教程做中文字幕]beginner Graphics – Lessons系列之纹理Textures 本篇分享一下第6个已完工的视频,即<beginner Graphics ...

  2. javascript笔记---貌似大叔

    1.原型式继承和类式继承的区别 在基于类的面向对象方式中,对象(object)依靠类(class)来产生.而在基于原型的面向对象方式中,对象(object)则是依靠 构造器(constructor)利 ...

  3. 移动开发之fastclick 点击穿透

    穿透(点穿)是在mobile各种浏览器上发生的常见的bug.可能是由click事件的延迟(300ms)或者事件冒泡导致 现象:在A页面中有个 btn1<或a标签>,在B页面中有个 btn2 ...

  4. Android有效的治疗方法Bitmap,减少内存

    Android有效的治疗方法Bitmap,减少内存 照片可能有不同的大小. 在很多情况下,大小.比如,我们的Camera应用,我们所拍的照片的大小远大于屏幕显示的大小 假如你的应用被限制了内存使用,显 ...

  5. Spring使用注解进行事务的管理

    使用步骤: 步骤一.在spring配置文件中引入<tx:>命名空间 <beans xmlns="http://www.springframework.org/schema/ ...

  6. 探究ElasticSearch中的线程池实现

    探究ElasticSearch中的线程池实现 ElasticSearch里面各种操作都是基于线程池+回调实现的,所以这篇文章记录一下java.util.concurrent涉及线程池实现和Elasti ...

  7. jupyter-notebook后home页面空白问题

    jupyter-notebook后home页面空白问题 解决方案1   更换默认的浏览器,选择谷歌浏览器,很多360打不开的页面,更换谷歌后都能有效解决,并且确保是最新版本的google浏览器. 解决 ...

  8. pagehelper 使用

    MySQL对分页的支持 简单来说MySQL对分页的支持是通过limit子句.请看下面的例子. limit关键字的用法是 LIMIT [offset,] rows offset是相对于首行的偏移量(首行 ...

  9. Matlab 根号的输入

    二次根号: sqrt(a)或a^0.5 三次根号: x^(1/3)或者x.^(1/3) 根据x的数据结构类型 矩阵.数组需要.^

  10. Java 基础系列合集

    Java基础知识你知道多少? Java特性 Java三大特性:封装,继承,多态 Java 抽象类与接口 Java 浅拷贝和深拷贝 Java static和final Java 内部类.静态内部类.匿名 ...