Flume NG源码分析（三）使用Event接口表示数据流

标签：flume ng   event   

Flume NG有4个主要的组件：

Event表示在Flume各个Agent之间传递的数据流

Source表示从外部源接收Event数据流，然后传递给Channel

Channel表示对从Source传递的Event数据流的临时存储

Sink表示从Channel中接收存储的Event数据流，并传递给下游的Source或者终点仓库

这篇看一下Event接口表示的数据流。Source, Channel, Sink操作的数据流都是基于Event接口的封装。

public interface Event {

  /**
   * Returns a map of name-value pairs describing the data stored in the body.
   */
  public Map<String, String> getHeaders();

  /**
   * Set the event headers
   * @param headers Map of headers to replace the current headers.
   */
  public void setHeaders(Map<String, String> headers);

  /**
   * Returns the raw byte array of the data contained in this event.
   */
  public byte[] getBody();

  /**
   * Sets the raw byte array of the data contained in this event.
   * @param body The data.
   */
  public void setBody(byte[] body);

}

Event的类层次结构如下

来看一下常用的SimpleEvent的具体实现，ExecSource等组件都是使用它来封装本地日志数据。它的实现非常简单，就是设置了header和body两部分数据。

public class SimpleEvent implements Event {

  private Map<String, String> headers;
  private byte[] body;

  public SimpleEvent() {
    headers = new HashMap<String, String>();
    body = new byte[0];
  }

  @Override
  public Map<String, String> getHeaders() {
    return headers;
  }

  @Override
  public void setHeaders(Map<String, String> headers) {
    this.headers = headers;
  }

  @Override
  public byte[] getBody() {
    return body;
  }

  @Override
  public void setBody(byte[] body) {
    if(body == null){
      body = new byte[0];
    }
    this.body = body;
  }

  @Override
  public String toString() {
    Integer bodyLen = null;
    if (body != null) bodyLen = body.length;
    return "[Event headers = " + headers + ", body.length = " + bodyLen + " ]";
  }

}

public class EventBuilder {

  /**
   * Instantiate an Event instance based on the provided body and headers.
   * If <code>headers</code> is <code>null</code>, then it is ignored.
   * @param body
   * @param headers
   * @return
   */
  public static Event withBody(byte[] body, Map<String, String> headers) {
    Event event = new SimpleEvent();

    if(body == null) {
      body = new byte[0];
    }
    event.setBody(body);

    if (headers != null) {
      event.setHeaders(new HashMap<String, String>(headers));
    }

    return event;
  }
}

Flume NG源码分析（三）使用Event接口表示数据流

标签：flume ng   event   

原文地址：http://blog.csdn.net/iter_zc/article/details/46519119