由于要實時讀取redis的AOF文件，但是flume的taildir source在監(jiān)控文件的時候，如果文件的inode變化了，那么會出現(xiàn)重復(fù)讀取數(shù)據(jù)的情況，這里可以通過修改flume taildir源碼解決，只針對讀一個文件的情況。
- 去flume官網(wǎng)下載flume源碼下載
- 解壓后在idea中打開如下
配置好maven，到flume-ng-source中找到ReliableTaildirEventReader
- 找到updateTailFiles方法
/**
* Update tailFiles mapping if a new file is created or appends are detected
* to the existing file.
*/
public List<Long> updateTailFiles(boolean skipToEnd) throws IOException {
pass
...
...
for (TaildirMatcher taildir : taildirCache) {
long inode = getInode(f);
TailFile tf = tailFiles.get(inode);
//判斷是否是新文件，inode或者文件名不同就認(rèn)為是新文件
if (tf == null || !tf.getPath().equals(f.getAbsolutePath())) {
long startPos = skipToEnd ? f.length() : 0;
System.out.println(tf);
if (tf != null) {
inode = tf.getInode();
}
//找到這行，startPos是讀取文件的位置，當(dāng)有新文件時會從0開始讀
//tf = openFile(f, headers, inode, startPos);
//改成??，f.length()是此時讀到的位置
tf = openFile(f, headers, inode, f.length());
} else {
繼續(xù)找到TaildirSource類
private String toPosInfoJson() {
@SuppressWarnings("rawtypes")
List<Map> posInfos = Lists.newArrayList();
for (Long inode : existingInodes) {
TailFile tf = reader.getTailFiles().get(inode);
//這里會將新的inode寫到位置文件中
//posInfos.add(ImmutableMap.of("inode", inode, "pos", tf.getPos(), "file", tf.getPath()));
//改成??，inode=0
posInfos.add(ImmutableMap.of("inode", 0, "pos", tf.getPos(), "file", tf.getPath()));
}
return new Gson().toJson(posInfos);
}
右側(cè)找到package打包，或者進入到項目目錄用mvn package打包，復(fù)制target中的flume-taildir-source-1.8.0.jar到flume中的lib下即可，這樣即使文件的inode變化，也可以繼續(xù)讀
另外，我是把taildir這個包又復(fù)制了一份，這樣在flume的配置中a1.sources.r1.type = org.apache.flume.source.taildir2.TaildirSource
直接指定修改后的類名，這樣不會影響原來的TAILDIR
