本文在【flume使用(二)：采集遠程日志數據到MySql數據庫】一文基礎之上進行測試操作。本文使用到的：

flume版本、jdk版本、mysql、數據庫表、javaBean、自定義的mysqlSink、以及多agent配置均同【flume使用(二)：采集遠程日志數據到MySql數據庫】

一、需求說明

采集遠程系統上log4j實時生成的日志到本機上進行數據庫持久化。

二、設計

數據流：

(1)log4j(配置文件配置flume輸出)

----[通過avro協議傳輸]------->(2)一個agent(source:avro; channel:memory; sink:avro)

----[通過avro協議傳輸]------>(3)一個agent(source:avro; channel:memory; sink:自定義mysqlSink)

---[mysqlSink進行jdbc持久化]----->(4)mysql數據庫

本測試為了方便一些：

(1)是在windows上直接eclipse執行；(可以打包讓運行在linux機器202上)

(2)是linux機器202；

(3)是linux機器201；

(4)數據庫在linux機器201上

三、實施

1.編寫實時輸出log4j日志程序

(1)新建maven測試項目：pom.xml

4.0.0

yichao.mym

flume-log4j

0.0.1-SNAPSHOT

log4j

log4j

1.2.16

org.apache.flume

flume-ng-core

1.7.0

org.apache.flume.flume-ng-clients

flume-ng-log4jappender

1.7.0

(2)編寫log4j.properties

log4j.rootLogger=INFO, stdout, flume

log4j.logger.per.flume=INFO

### flume ###

log4j.appender.flume=org.apache.flume.clients.log4jappender.Log4jAppender

log4j.appender.flume.layout=org.apache.log4j.PatternLayout

log4j.appender.flume.Hostname=192.168.216.202

log4j.appender.flume.Port=44444

### stdout ###

log4j.appender.stdout=org.apache.log4j.ConsoleAppender

log4j.appender.stdout.Threshold=INFO

log4j.appender.stdout.Target=System.out

log4j.appender.stdout.layout=org.apache.log4j.PatternLayout

log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %c{1} [%p] %m%n

(3)編寫日志打印程序

package yichao.mym.base.bean;

import org.apache.log4j.Logger;

public class WriteLog {

protected static final Logger logger = Logger.getLogger(WriteLog.class);

public static void main(String[] args) throws Exception {

int name = 0;

while (true) {

logger.info("log4j-flume-name-"+name+","+(name++));

Thread.sleep(1000);

}

}

}

注意這里打印的數據必須要能讓mysqlSink.java能夠解析。由于log4j輸出數據格式問題，導致原本根據“,”拆分的字符不能正確解析，原因是輸出的數據如： [?log4j-flume-name-1668,1668?? ] 。這條數據前后都有間隔，導致第二個字段"1688? "不能被轉換成整型。

解決方式：修改解析方式：mysqlSink.java中在string轉int類型時，先對string進行去除空格處理

處理前：person.setAge(Integer.parseInt(content.substring(content.indexOf(",")+1)));

處理后：person.setAge(Integer.parseInt(content.substring(content.indexOf(",")+1).trim()));

目前文章flume使用(二)已經做了修正

2.編寫采集log4j實時傳輸過來數據的flume的配置文件

此配置文件放到linux機器202中，而原linux機器201的conf配置文件不變仍然使用：avro-mysql.conf

文件名：log4jAvro-avro.conf

agent1.sources=source1

agent1.channels=channel1

agent1.sinks=mysqlSink

# describe/configure source1

agent1.sources.source1.type=avro

agent1.sources.source1.bind=192.168.216.202

agent1.sources.source1.port=44444

agent1.sources.source1.channels=channel1

# use a channel which buffers events in memory

# type:memory or file is to temporary to save buffer data which is sink using

agent1.channels.channel1.type=memory

agent1.channels.channel1.capacity=5000

agent1.channels.channel1.transactionCapacity=1000

agent1.sinks.mysqlSink.type=avro

agent1.sinks.mysqlSink.channel=channel1

agent1.sinks.mysqlSink.hostname=192.168.216.201

agent1.sinks.mysqlSink.port=4545

配置文件同樣放在linux機器202的flume目錄中的bin中

四、測試都進入flume中的bin目錄下執行命令

1.啟動linux機器201

./flume-ng agent -c ../conf -f ../conf/avro-mysql.conf -n agent1 -Dflume.root.logger=INFO,console

2.啟動linux機器202(注意啟動的配置文件)

./flume-ng agent -c ../conf -f ../conf/log4jAvro-avro.conf -n agent1 -Dflume.root.logger=INFO,console

3.在windows上執行編寫好的實時輸出log4j的程序

此時刷新mysql數據庫表，即可看到數據正源源不斷的插入進來：

總結

以上是生活随笔為你收集整理的flume可以实时监控mysql嘛_flume使用（三）：实时log4j日志通过flume输出到MySql数据库...的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。