flume 1.7.0推出了taildirSource元件。tail監控目錄下匹配上正則表示式的的所有檔案，實現斷點續傳。

但是我後來發現cdh版的flume-1.6.0也已經有這個元件了，而官方的Apache版的apache-flume-1.6.0-bin卻沒有這個元件。並且Apache版的flume1.7.0比cdh版的flume1.6.0的org.apache.flume.source.taildir包中多出TaildirMatcher.java，並且其他的程式碼也有所不同，應該Apache版的flume1.7.0比cdh版的flume1.6.0更加完善，兩者的差異在哪裡，有時間再研究研究吧。因為我的機器上已經裝有cdh版的apache-flume-1.6.0-cdh5.5.2-bin，所以就直接用這個而沒有重灌Apache版的flume1.7.0了，並且apache-flume-1.6.0-cdh5.5.2-bin目前已經夠用了。

[[email protected] conf]$ cat taildir.conf 

a1.sources = r1
a1.channels = c1
a1.sinks = k1

# Describe/configure the source
#source的型別為TAILDIR，這裡的型別大小寫都可以
a1.sources.r1.type = taildir
a1.sources.r1.channels = c1
#儲存tial最後一個位置儲存位置
a1.sources.r1.positionFile = /home/hadoop/hui/taildir_position.json
#設定tiail的組， 使用空格隔開
a1.sources.r1.filegroups = f1 f2
#設定每個分組的絕對路徑
a1.sources.r1.filegroups.f1 = /home/hadoop/hui/test1/hehe.txt
a1.sources.r1.filegroups.f2 = /home/hadoop/hui/test2/.*
#.匹配除換行符 \n 之外的任何單字元。*匹配前面的子表示式零次或多次。這裡也可以用messages.*
a1.sources.r1.fileHeader = true

# Describe the sink
a1.sinks.k1.type = file_roll
a1.sinks.k1.sink.directory = /home/hadoop/hui
a1.sinks.k1.sink.rollInterval = 0
 
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
 
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
 

[[email protected] hui]$ tree

.
|-- messages.1
|-- qiang
|   `-- hui.txt
|-- test1
|   |-- hehe.txt
|   `-- messages.2
`-- test2
    |-- messages.3
    `-- messages.4

3 directories, 6 files

[[email protected] hui]$ cat test1/hehe.txt 
hello world hehe
[[email protected] hui]$ cat test1/messages.2 
hello world 2
[

[email protected] hui]$ cat test2/messages.3 
hello world 3
[[email protected] hui]$ cat test2/messages.4
hello world 4

啟動flume程序：
[[email protected] apache-flume-1.6.0-cdh5.5.2-bin]$ bin/flume-ng agent -c . -f conf/taildir.conf -n a1 -Dflume.root.logger=INFO,console

[[email protected] hui]$ ls（在hui/目錄下生成了1489881718232-1和taildir_position.json檔案）
1489881718232-1  messages.1  qiang  taildir_position.json  test1  test2
[[email protected] hui]$ cat 1489881718232-1 
hello world hehe
hello world 3
hello world 4
[[email protected] hui]$ cat taildir_position.json 
[{"inode":1532721,"pos":17,"file":"/home/hadoop/hui/test1/hehe.txt"},{"inode":1532719,"pos":14,"file":"/home/hadoop/hui/test2/messages.3"},{"inode":1532720,"pos":14,"file":"/home/hadoop/hui/test2/messages.4"}]

往test1/hehe.txt中傳入資料：
[[email protected] hui]$ echo "ni hao world" >> test1/hehe.txt 

再觀察1489881718232-1和taildir_position.json檔案
[[email protected] hui]$ cat 1489881718232-1 
hello world hehe
hello world 3
hello world 4
ni hao world
[[email protected] hui]$ cat taildir_position.json 
[{"inode":1532721,"pos":30,"file":"/home/hadoop/hui/test1/hehe.txt"},{"inode":1532719,"pos":14,"file":"/home/hadoop/hui/test2/messages.3"},{"inode":1532720,"pos":14,"file":"/home/hadoop/hui/test2/messages.4"}]

限制：
1.可以配置一個監聽目錄，會監聽該目錄下所有的檔案，但是如果配置目錄下面嵌套了子目錄，則無法監聽。需要修改原始碼，我們可以遞迴地對配置目錄的所有子目錄的所有檔案進行監聽
2.taildirSource元件不支援檔案改名的。如果檔案改名會認為是新檔案，就會重新讀取，這就導致了日誌檔案重讀。（應用場景：Linux的系統日誌輪詢功能）

解決限制一：
網上已經有修改後的程式碼了，下載地址：https://github.com/qwurey/flume-source-taildir-recursive

將程式碼下載下來並匯入myeclipse中

解決限制二：
修改ReliableTaildirEventReader 類的 updateTailFiles 方法。
將其中的 tf.getPath().equals(f.getAbsolutePath()) 判斷條件去除。只用判斷檔案不為空即可，不用判斷檔案的名字。

//        if (tf == null || !tf.getPath().equals(f.getAbsolutePath())) {
                if (tf == null) {//檔案不存在 position 中則全讀。

修改TailFile 類的 updatePos 方法
此處同樣的原因，inode 已經能夠確定唯一的 檔案了，所以不用加 path 作為判定條件了。所以去掉該條件就支援了檔案重新命名情況。

//     if (this.inode == inode && this.path.equals(path)) {
          if (this.inode == inode) {

修改這兩個地方就支援了檔案重新命名的問題，實現了目錄下多檔案監控，斷點續傳。

新增自定義source入口，也就是將原始碼拷貝過來，然後將修改過的程式碼打包為自定義taildir的jar包執行flume。將taidir.jar匯入到flume的lib目錄下。

[[email protected] conf]$ cat taildir.conf

a1.sources = r1
a1.channels = c1
a1.sinks = k1
  
# Describe/configure the source
a1.sources.r1.type = com.urey.flume.source.taildir.TaildirSource
a1.sources.r1.channels = c1
a1.sources.r1.positionFile = /home/hadoop/q1/taildir_position.json
a1.sources.r1.filegroups = f1
a1.sources.r1.filegroups.f1 = /home/hadoop/hui/.*
a1.sources.r1.batchSize = 100
a1.sources.r1.backoffSleepIncrement  = 1000
a1.sources.r1.maxBackoffSleep  = 5000

# Describe the sink
a1.sinks.k1.type = file_roll
a1.sinks.k1.sink.directory = /home/hadoop/q1
a1.sinks.k1.sink.rollInterval = 0

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

[[email protected] ~]$ mkdir q1
[[email protected] hui]$ tree

.
|-- messages.1
`-- test2
    |-- messages.3
    |-- messages.4
    `-- test1
        |-- hehe.txt
        `-- messages.2

2 directories, 5 files

[[email protected] hui]$ cat messages.1 
hello world 1
[[email protected] hui]$ cat test2/messages.3
hello world 3
[[email protected] hui]$ cat test2/messages.4 
hello world 4
[[email protected] hui]$ cat test2/test1/hehe.txt 
hello world hehe
[[email protected] hui]$ cat test2/test1/messages.2 
hello world 2

啟動flume：
[[email protected] apache-flume-1.6.0-cdh5.5.2-bin]$ bin/flume-ng agent -c . -f conf/taildir.conf -n a1 -Dflume.root.logger=INFO,console

[[email protected] q1]$ ls
1489910670584-1  taildir_position.json
[[email protected] q1]$ cat 1489910670584-1 
hello world 2
hello world 4
hello world 1
hello world hehe
hello world 3
[[email protected] q1]$ cat taildir_position.json 
[{"inode":1532727,"pos":14,"file":"/home/hadoop/hui/test2/test1/messages.2"},{"inode":1532720,"pos":14,"file":"/home/hadoop/hui/test2/messages.4"},{"inode":1532722,"pos":14,"file":"/home/hadoop/hui/messages.1"},{"inode":1532718,"pos":17,"file":"/home/hadoop/hui/test2/test1/hehe.txt"},{"inode":1532712,"pos":14,"file":"/home/hadoop/hui/test2/messages.3"}]

[[email protected] hui]$ mv test2/test1/hehe.txt test2/haha.txt
[[email protected] hui]$ cat ../q1/1489910670584-1 
hello world 2
hello world 4
hello world 1
hello world hehe
hello world 3
[[email protected] hui]$ cat ../q1/taildir_position.json 
[{"inode":1532727,"pos":14,"file":"/home/hadoop/hui/test2/test1/messages.2"},{"inode":1532720,"pos":14,"file":"/home/hadoop/hui/test2/messages.4"},{"inode":1532722,"pos":14,"file":"/home/hadoop/hui/messages.1"},{"inode":1532718,"pos":17,"file":"/home/hadoop/hui/test2/test1/hehe.txt"},{"inode":1532712,"pos":14,"file":"/home/hadoop/hui/test2/messages.3"}]

[[email protected] hui]$ echo "hello world haha" >> test2/haha.txt 
[[email protected] hui]$ cat ../q1/1489910670584-1 
hello world 2
hello world 4
hello world 1
hello world hehe
hello world 3
hello world haha
[[email protected] hui]$ cat ../q1/taildir_position.json 
[{"inode":1532727,"pos":14,"file":"/home/hadoop/hui/test2/test1/messages.2"},{"inode":1532720,"pos":14,"file":"/home/hadoop/hui/test2/messages.4"},{"inode":1532722,"pos":14,"file":"/home/hadoop/hui/messages.1"},{"inode":1532712,"pos":14,"file":"/home/hadoop/hui/test2/messages.3"},{"inode":1532718,"pos":34,"file":"/home/hadoop/hui/test2/haha.txt"}]

[[email protected] hui]$ echo "hello china" >> test2/test1/hehe.txt
[[email protected] hui]$ cat ../q1/1489910670584-1 
hello world 2
hello world 4
hello world 1
hello world hehe
hello world 3
hello world haha
hello china
[[email protected] hui]$ cat ../q1/taildir_position.json 
[{"inode":1532727,"pos":14,"file":"/home/hadoop/hui/test2/test1/messages.2"},{"inode":1532720,"pos":14,"file":"/home/hadoop/hui/test2/messages.4"},{"inode":1532722,"pos":14,"file":"/home/hadoop/hui/messages.1"},{"inode":1532712,"pos":14,"file":"/home/hadoop/hui/test2/messages.3"},{"inode":1532718,"pos":34,"file":"/home/hadoop/hui/test2/haha.txt"},{"inode":1532714,"pos":12,"file":"/home/hadoop/hui/test2/test1/hehe.txt"}]

參考：
http://blog.csdn.net/u012373815/article/details/62241528
http://www.2cto.com/net/201703/616085.html
http://blog.csdn.net/yeruby/article/details/51812759