2019独角兽企业重金招聘Python工程师标准>>> 
下面看一下源码中时间戳的拦截器
public class TimestampInterceptor implements Interceptor {
private final boolean preserveExisting;
/**
* Only {@link TimestampInterceptor.Builder} can build me
*/
private TimestampInterceptor(boolean preserveExisting) {
this.preserveExisting = preserveExisting;
}
@Override
public void initialize() {
// no-op
}
/**
* Modifies events in-place.
*这个是对单个事件的处理
*/
@Override
public Event intercept(Event event) {
//获得事件中的所有头信息，都是以key-value的形式存储
Map<String, String> headers = event.getHeaders();
if (preserveExisting && headers.containsKey(TIMESTAMP)) {
// we must preserve the existing timestamp
} else {
long now = System.currentTimeMillis();
//把时间戳放到头信息
headers.put(TIMESTAMP, Long.toString(now));
}
return event;
}
/**
* Delegates to {@link #intercept(Event)} in a loop.
* @param events
*这是对多个事件的处理
* @return
*/
@Override
public List<Event> intercept(List<Event> events) {
for (Event event : events) {
intercept(event);
}
return events;
}
@Override
public void close() {
// no-op
}
/**
* Builder which builds new instances of the TimestampInterceptor.
*这个方法是必须实现的，这个接口Interceptor中注释明确必须实现
*/
public static class Builder implements Interceptor.Builder {
private boolean preserveExisting = PRESERVE_DFLT;
@Override
public Interceptor build() {
return new TimestampInterceptor(preserveExisting);
}
@Override
public void configure(Context context) {
preserveExisting = context.getBoolean(PRESERVE, PRESERVE_DFLT);
}
}
public static class Constants {
public static String TIMESTAMP = "timestamp";
public static String PRESERVE = "preserveExisting";
public static boolean PRESERVE_DFLT = false;
}
}
大家可以写自己的过滤器，特别对于log4j的日志处理，例如如果你的日志每一个小时生成一个文件，但是我要把它按照天来存储，大家可以按照这个这种形式实现. 把生成的包放到/app/apache-flume-1.7.0-bin/plugins.d/interceptor/lib下，如果/plugins.d不存在，自己就建立这个目录就可以. 在配置文件中添加：xxx.sources.r1.interceptors.i1.type = com.hc360.interceptor.DateTimeSuffixInterceptor$Builder就可以。 现在想把自己做的一些东西记录一下，坚持，坚持。。。。。。。。。。。。。。