如果你還沒看過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()通常是啟動線程去執行消費操作。