介紹

我們的使用者是經常在登陸，由於是涉及到裝置，產品希望每個使用者一登陸，後臺系統就能感知到其變化，即實時更新，
登陸資料量是很大的，大約一天有1500W左右的資料,且比較集中在晚上.高峰時1秒鐘要處理200多個登陸請求,
負責登陸的系統是業務的核心，架構上我們不想讓它受到更多的壓力，於是決定以 日誌方式，收集資料後進行使用者時間的更新。

我們採用flume+kafka的方式來收集
flume是1.7版本
kafka 是 0.10.1版本

flume及kafka的配置

啟動flume
./bin/flume-ng agent -n a1 -c conf -f conf/flume-conf.properties -Dflume.root.logger=INFO,console

配置檔案是要先寫好的

a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置 source
a1.sources.r1.type = org.apache.flume.source.taildir.TaildirSource
a1.sources.r1.positionFile = /xxxx/flume/taildir_position_klog.json
a1.sources.r1.filegroups = f1
a1.sources.r1.filegroups.f1 =/xxxx/log/business_login/business.tmp
a1.sources
 
.r1.batchSize = 100
a1.sources.r1.backoffSleepIncrement  = 1000
a1.sources.r1.maxBackoffSleep  = 5000
a1.sources.r1.recursiveDirectorySearch = true
a1.sources.r1.yarnApplicationHeader = true
a1.sources.r1.yarnContainerHeader = true
# 配置 sink
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k
 
1.brokerList = xxxx:9092
a1.sinks.k1.topic = klog
a1.sinks.k1.serializer.class = kafka.serializer.StringEncoder
# 定義 channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 用 channel 連線起來 source 和 sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

flume有source和sink ，source即來源，我們上面是指向一個日誌檔案的，sink則是資料流出，我們這裡是使用kafka的，flume有官方的kafka整合外掛，如果你想到sink到其它自定義的地方，也可以自己寫外掛

在上面的配置中，我們監聽 a1.sources.r1.filegroups.f1 ，在讀的過程 中，會把記錄存在 a1.sources.r1.positionFile，通過a1.channels.c1.type這個 channel, 把資料流向 a1.sinks.k1.type 中.

業務系統寫日誌是用log4j的，40M寫完一個檔案就切換，我們只需要監聽那個一直寫的檔案即可。

嚴重提醒!!!flume有一個極其蛋疼的問題，如果你監聽一個檔案，那個檔案追加的資料會被 flume 給獲取,但是！！如果你去主動追加/vi加上一行，那這個flume就會無限重複全檔案掃讀後扔給kafka。。可以自行測試下。

kafak不用作其它的配置，flume只是作為它的一個生產者.kafka會在硬盤裡存7天左右的資料，所以消費者要及時消費完。

使用狀態分析

我們有三臺伺服器會生成日誌，部署三個flume,都指向同一個kafka,我們的kafak現在還不是分佈的，但使用上也沒啥問題，其佔用記憶體挺少，我們沒有特別去配置kafka,運行了1個多月，其記憶體只佔用 32G中的4%左右, gc上fullgc也只是0次.
速度上也能保證。

kafka消費者的處理也是要多注意，相對生產者來講，消費者比較複雜，由於資料量較大，我們的消費者也分了三個服務，通過同一個group 來負載均衡的。