前言：接上一篇，我们使用flume自带的logger有些bug，只能显示前16个字节。本文将介绍如何解决这个问题，并分析Logger Sink的源码进而得出自定义Sink的一般流程。
一、解决Logger Sink问题
1.1 尝试使用maxBytesToLog属性
之前我们说过，有问题就去官方文档看看，有关Logger Sink的内容如下：
可以看到这里面有个maxBytesToLog的属性，默认16，描述是Event的body从日志中输出的最大字节数。那我们可以直接设置这个的值试一试。
修改flume的conf目录下的appenIP.conf，添加logger sink的maxBytesToLog属性：a1.sinks.k1.maxBytesToLog =1024。然后重新启动flume。怀着信心，这次一定能成功。
1.2 自定义Logger Sink
结果还是不行，就很郁闷，然后准备开始看logger sink的源码来解决问题，在看源码之前我先把结果和大家说一下，我还是用上一篇的maven工程，新建了一个类MyLoggerSink，然后把LoggerSink的代码原封不动粘贴过去，只是在关键位置加了一些自己的log输出，用来方便查看运行状态，打包到flume的lib目录下。然后在flume的配置文件中进行修改，
#把sink换成我们自己的类。
a1.sinks.k1.type = com.zhb.flume.MyLoggerSink
#这个属性设置成1024字节
a1.sinks.k1.maxBytesToLog =1024
启动flume，进入到flume的bin目录下，执行./flume-ng agent -c ../conf -f ../conf/appendIP.conf -Dflume.root.logger=INFO,console -n a1,启动成功，有一些我们自己的状态信息，如下：
可以看到先进入到configure方法中，并且成功从配置文件中获得到了maxBytesToLog 的值1024。还有一些有趣的东西。在没有消息发送的情况下，process方法每隔几秒钟执行一次，如下：
简单看了下源码是SinkRunner这个类下的静态内部类PollingRunner的作用，这个类实现是Runnable接口，里面不断调用process方法并有Thread.sleep让线程休眠一段时间。（这部分源码我也没有特别研读，后面有机会再深入一下吧，又立flag。）。
然后，接下来测试一下吧，新开一个终端，输入很长的一串数据：echo "AppendIPAddressAppendIPAddressAppendIPAddressAppendIPAddressAppendIPAddressAppendIPAddress" | nc 127.0.0.1 50000 结果就可以了，有知道这是什么玄学的大佬请留言。
我们看下输出的内容，因为后面我们源码分析的时候可能会提到：
前面的类似00000010这个是偏移量，是16进制的，也就是每16个字节一组，中间的是数据的ASII码值，最后是真实数据，第一行没有偏移量，因为在源码中把00000000去掉了，说是什么数据太少，要偏移没意义。乖乖，数据多了你又不给我显示，你想气死我么？
二、Logger Sink源码解读
下面看下Logger Sink的源码，我会在代码中添加一些注释
//继承AbstractSink以能够处理Event，实现Configurable 则能够在配置文件中配置属性
public class LoggerSink extends AbstractSink implements Configurable {
private static final Logger logger = LoggerFactory
.getLogger(LoggerSink.class);
// Default Max bytes to dump
public static final int DEFAULT_MAX_BYTE_DUMP = 16;
// Max number of bytes to be dumped
private int maxBytesToLog = DEFAULT_MAX_BYTE_DUMP;
public static final String MAX_BYTES_DUMP_KEY = "maxBytesToLog";
@Override
public void configure(Context context) {
//从配置文件中读取maxBytesToLog属性的值
String strMaxBytes = context.getString(MAX_BYTES_DUMP_KEY);
if (!Strings.isNullOrEmpty(strMaxBytes)) {
try {
maxBytesToLog = Integer.parseInt(strMaxBytes);
} catch (NumberFormatException e) {
logger.warn(String.format(
"Unable to convert %s to integer, using default value(%d) for maxByteToDump",
strMaxBytes, DEFAULT_MAX_BYTE_DUMP));
maxBytesToLog = DEFAULT_MAX_BYTE_DUMP;
}
}
}
//这个函数是用来让sink消费从关联的channel中得到的数据，SinkRunner不断调用这个方法
@Override
public Status process() throws EventDeliveryException {
Status result = Status.READY;
Channel channel = getChannel();
Transaction transaction = channel.getTransaction();
Event event = null;
try {
transaction.begin();
//从channel中获得下一个event
event = channel.take();
if (event != null) {
//打印日志，dumpEvent方法是我们下一个要看的。
if (logger.isInfoEnabled()) {
logger.info("Event: " + EventHelper.dumpEvent(event, maxBytesToLog));
}
} else {
// No event found, request back-off semantics from the sink runner
result = Status.BACKOFF;
}
transaction.commit();
} catch (Exception ex) {
transaction.rollback();
throw new EventDeliveryException("Failed to log event: " + event, ex);
} finally {
transaction.close();
}
return result;
}
}
下面看dumpEvent(Event event, int maxBytes)这个方法：
public static String dumpEvent(Event event, int maxBytes) {
//其实就是完成字符串的拼接
StringBuilder buffer = new StringBuilder();
if (event == null || event.getBody() == null) {
buffer.append("null");
} else if (event.getBody().length == 0) {
// do nothing... in this case, HexDump.dump() will throw an exception
} else {
byte[] body = event.getBody();
byte[] data = Arrays.copyOf(body, Math.min(body.length, maxBytes));
ByteArrayOutputStream out = new ByteArrayOutputStream();
try {
//这条语句把data转为为这种形式：16进制偏移量 16进制ASII码 真实值
HexDump.dump(data, 0, out, 0);
String hexDump = new String(out.toByteArray());
// remove offset since it's not relevant for such a small dataset
//这里把第一次的偏移量去掉。也就是00000000去掉
if (hexDump.startsWith(HEXDUMP_OFFSET)) {
hexDump = hexDump.substring(HEXDUMP_OFFSET.length());
}
buffer.append(hexDump);
} catch (Exception e) {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Exception while dumping event", e);
}
buffer.append("...Exception while dumping: ").append(e.getMessage());
}
String result = buffer.toString();
//去掉换行符
if (result.endsWith(EOL) && buffer.length() > EOL.length()) {
buffer.delete(buffer.length() - EOL.length(), buffer.length()).toString();
}
}
//返回这种形式的日志内容。
return "{ headers:" + event.getHeaders() + " body:" + buffer + " }";
}
三、自定义Sink初探
通过阅读源码，我们可以大致总结出自定义Sink的流程：
① 搭建flume开发环境（我用maven然后引入flume的依赖）
② 新建一个类继承AbstractSink类，如果需要实现可在配置文件中配置，那么还需实现Configurable接口。
③添加自己需要的属性并重写process方法和configure方法（如果实现了Configurable接口）
四、总结
本文 “解决”了logger sink的一些问题（尴尬ing），并且分析了一下logger sink的源码，从源码中总结了一下自定义Sink的方法。下一篇准备实现自定义Sink到Mysql数据库，敬请期待。
京公网安备 11010502036488号