1.1．自定义Source说明

Source是负责接收数据到Flume Agent的组件。官方提供的source类型已经很多，但是有时候并不能满足实际开发当中的需求，此时就需要根据实际需求自定义某些source。

官方提供了自定义source的接口说明：

<u>https://flume.apache.org/FlumeDeveloperGuide.html#source</u>

自定义source中有一种类型叫做PollableSource。底层是通过线程不断去调用process方法，主动拉取消息。

1.2．自定义Source原理

根据官方说明自定义source组件需要继承AbstractSource类并实现Configurable和PollableSource接口，并实现相应方法：

image.png

1.3．示例说明

public class MySource extends AbstractSource implements Configurable, PollableSource {
  private String myProp;

  @Override
  public void configure(Context context) {
    String myProp = context.getString("myProp", "defaultValue");

    // 获取处理自定义 conf 文件中的属性值（验证，或类型转换......）
    // preText = context.getString("preText");

    // 将 myProp 放在一个变量中 后续 process() 方法需要用
    this.myProp = myProp;
  }

  @Override
  public void start() {
    // 初始化与外部客户端的连接
  }

  @Override
  public void stop () {
    // 释放资源或清空字段等
  }

  @Override
  public Status process() throws EventDeliveryException {
    Status status = null;

    try {
      // 自定义处理

      // 接收新数据
      Event e = getSomeData();

      // 将 Event  存储到此 Source 相关Channel(s) 中
      getChannelProcessor().processEvent(e);

      status = Status.READY;
    } catch (Throwable t) {
      // 异常记录, 处理

      status = Status.BACKOFF;

      // 异常抛出
      if (t instanceof Error) {
        throw (Error)t;
      }
    } finally {
      txn.close();
    }
    return status;
  }
}