|
背景：银行日志生产方式一般有两种
1）按大小切分：xxx.logxxx.log1 xxx.log2，及最新日志写入.log，原来的.log
mv为.log1，.log1 mv为.log2，依次类推，每个日志固定大小（10M、50M之类）。
2）按天切分：xxx.log xxx.log-20171224(xxx.log-日期),最新日志写入.log，后面的按照日期备份，基本为每天一个日志。
目前缺陷：flume1.7增加了TAILDIR source，可以用于tail日志一个目录下的多个文件。但是
1）、出现数据重复（应用场景是用flume抓取应用日志，根据应用日志实时算交易量，所以会导致交易量增加），如果文件a.log达到定大小后，会被重命名为a.log.1，新建一个a.log,那么此时a.log和a.log.1的inode一样，文件名不一样，默认的flume
taildir会认为这是都是一个新文件，就会重新读一遍，所以导致数据重复；2）、不释放资源，如果a.log.11，a.log.12之前被flume监控过，现在把这两个文件删除了，flume不停止，资源不会被释放。
修改思路：文件如果更名后文件没有关闭，先关闭原文件的句柄，之后判断文件是否更新，若更新，重新打开文件；没有更新，只修改文件路径。
代码修改：taildir
包下
ReliableTaildirEventReader类的ReliableTaildirEventReader方法修改，
原代码为:
|
if(tf==null||!tf.getPath().equals(f.getAbsolutePath())){
long
startPos=skipToEndf.length():0;
tf=openFile(f,headers,inode,startPos);
}else{…}
|
修改为：
|
if (tf == null || !tf.getPath().equals(f.getAbsolutePath())) {
//自定义代码开始
long startPos;
if(tf!=null){
boolean updated = tf.getLastUpdated() < f.lastModified();
if(tf.getRaf()!=null){//得到文件句柄
tf.close();
}
startPos = tf.getPos();
if(updated){ //更新
tf = openFile(f, headers, inode, startPos);
}else{//没更新
tf.updateFilePath(f.getAbsolutePath());
}
}else{
startPos = skipToEnd f.length() : 0;
tf = openFile(f, headers, inode, startPos);
}
//自定义代码结束
//源代码
} else {
|
然后将新代码打jar包，放入flume lib文件里，并在配置文件里面调用该source，测试结果，上面两个问题都得到了解决。
|