【Hadoop】Flume-ng源碼解析之Sink元件

如果你還沒看過Flume-ng源碼解析系列中的啟動流程和Channel元件件，可以點選下面連結：

Flume-ng源碼解析之啟動流程

Flume-ng源碼解析之Channel元件

作為啟動流程中第二個啟動的元件，我們今天來看看Sink的細節

Sink

Sink在agent中扮演的角色是消費者，将event輸送到特定的位置

首先依然是看代碼，由代碼我們可以看出Sink是一個接口，裡面最主要的方法是process()，用來處理從Channel中擷取的資料。Sink的執行個體是由SinkFactory.create()生成的。

@InterfaceAudience.Public
@InterfaceStability.Stable
public interface Sink extends LifecycleAware, NamedComponent {
  public void setChannel(Channel channel);
  public Channel getChannel();
  /* 用來處理channel中取來的event*/
  public Status process() throws EventDeliveryException;
  public static enum Status {
    READY, BACKOFF
  }
}
           

在啟動流程中我們了解到Application中啟動的不是Sink，而是SinkRunner，由名字我們可以看出這是一個驅動類。我們來看看代碼，主要看它的start()

public class SinkRunner implements LifecycleAware {

  ...

  @Override
  public void start() {
    SinkProcessor policy = getPolicy();

    policy.start();

    runner = new PollingRunner();

    runner.policy = policy;
    runner.counterGroup = counterGroup;
    runner.shouldStop = new AtomicBoolean();

    runnerThread = new Thread(runner);
    runnerThread.setName("SinkRunner-PollingRunner-" +
        policy.getClass().getSimpleName());
    runnerThread.start();

    lifecycleState = LifecycleState.START;
  }
  ...

}
           

我們知道啟動SinkRunner實際上就是調用它的start()，而在start()中可以看到主要是啟動了一個SinkProcessor，而這個SinkProcessor在建立SinkRunnner的時候已經指定了，如果你想要了解配置檔案是如何處理的，可以要去看看conf包裡面的類，可以看看org.apache.flume.node.AbstractConfigurationProvider中的getConfiguration()。

我們接着看看SinkProcessor

public interface SinkProcessor extends LifecycleAware, Configurable {
  Status process() throws EventDeliveryException;
  void setSinks(List<Sink> sinks);
}
           

SinkProcesor是一個接口，他的實作類由SinkProcessorFactory的getProcessor()生成，在AbstractConfigurationProvider中的loadSinkGroup()調用SinkGroup中的configure()生成。

public class SinkGroup implements Configurable, ConfigurableComponent {
  List<Sink> sinks;
  SinkProcessor processor;
  SinkGroupConfiguration conf;

  public SinkGroup(List<Sink> groupSinks) {
    sinks = groupSinks;
  }

  public SinkProcessor getProcessor() {
    return processor;
  }

  @Override
  public void configure(ComponentConfiguration conf) {
    this.conf = (SinkGroupConfiguration) conf;
    processor =
        SinkProcessorFactory.getProcessor(this.conf.getProcessorContext(),
            sinks);
  }
}
           

那麼我們以DefalutSinkProcessor為例子看看

public class DefaultSinkProcessor implements SinkProcessor, ConfigurableComponent {
  private Sink sink;
  private LifecycleState lifecycleState;

  @Override
  public void start() {
    Preconditions.checkNotNull(sink, "DefaultSinkProcessor sink not set");
    sink.start();
    lifecycleState = LifecycleState.START;
  }

  @Override
  public void stop() {
    Preconditions.checkNotNull(sink, "DefaultSinkProcessor sink not set");
    sink.stop();
    lifecycleState = LifecycleState.STOP;
  }

  @Override
  public LifecycleState getLifecycleState() {
    return lifecycleState;
  }

  @Override
  public void configure(Context context) {
  }

  @Override
  public Status process() throws EventDeliveryException {
    return sink.process();
  }

  @Override
  public void setSinks(List<Sink> sinks) {
    Preconditions.checkNotNull(sinks);
    Preconditions.checkArgument(sinks.size() == , "DefaultSinkPolicy can "
        + "only handle one sink, "
        + "try using a policy that supports multiple sinks");
    sink = sinks.get();
  }

  @Override
  public void configure(ComponentConfiguration conf) {

  }

}
           

從上面的代碼中我們可以看到SinkProcessor執行的還是sink的start、stop和process方法，那麼SinkProcessor的作用是什麼，Flume提供leFailoverSinkProcessor和LoadBalancingSinkProcessor，顧名思義，一個是失效備援，一個是負載均衡，那麼SinkProcessor不同子類的存在就是為了實作不同的配置設定操作和政策。而sink的start()通常是啟動線程去執行消費操作。