flume的部署及简单测试

冰轮封雪_飞

　　Flume 的一些核心概念：

组件功能Agent使用JVM 运行Flume。每台机器运行一个agent，但是可以在一个agent中包含多个sources和sinks。Client生产数据，运行在一个独立的线程。Source从Client收集数据，传递给Channel。Sink从Channel收集数据，运行在一个独立线程。Channel连接 sources 和 sinks ，这个有点像一个队列。Events可以是日志记录、 avro 对象等。配置flume
　　第一步：
　　下载最新的flume-ng
　　http://flume.apache.org/download.html
　　第二步：
　　配置conf
创建：flume-source-conf.properties
source：

　　#Describe  source
source.sources.src1.type = exec    //类型
source.sources.src1.command = tail  -n +0 -F /home/hduser/logs/debug.log //监控的日志文件

　　

　　#Describe channel      //配置channel
source.channels.ch1.type = file
source.channels.ch1.checkpointDir = tmp/channels/ch1/checkpoint
source.channels.ch1.dataDirs = tmp/channels/ch1/data

　　

　　#Describe sink     //配置sink
source.sinks.sink2.type = file_roll
source.sinks.sink2.sink.directory=store
source.sinks.sink2.sink.serializer = text
source.sinks.sink2.sink.serializer.appendNewline = true
source.sinks.sink2.sink.rollInterval = 43200

　　

　　//配置变量：
　　vim ~/.bash_profile
#export FLUME_HOME
export FLUME_HOME=/home/hduser/softsare/flume-1.6.0
export PATH=$PATH:$FLUME_HOME/sbin
　　

　　. ~/.bash_profile   //一定执行者一步让变量生效

　　

　　启动命令：flume strart source
　　停止命令：flume stop source
　　

　　//到此简单的flume配置完成，，可以用下面的日志生成程序产生日志。
　　package main;


import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;

import org.apache.log4j.Logger;
import org.apache.log4j.xml.DOMConfigurator;

import thread.DebugLogThread;
import thread.ErrorLogThread;
import thread.InfoLogThread;
import util.JFile;

public class LogFactory {

static Logger  logger = Logger.getLogger(LogFactory.class);
/**
*   绾跨▼瀹瑰櫒锛屼綔涓虹瓑寰呮墍鏈夌嚎绋嬮�鍑轰箣鐢ㄣ�
*/
private static ArrayList<Thread> threads = new ArrayList<Thread>();


/**
* 鍚&#57882;姩鏍稿績鍔熻兘鐩稿叧鐨勬墍鏈夋敞鍐岀嚎绋�
*/
public static void setpid() {

String processName = java.lang.management.ManagementFactory.getRuntimeMXBean().getName();

   String processID = processName.substring(0,processName.indexOf('@'));


   
   try {
JFile.write(&quot;pid.log&quot;, processID);
logger.debug(&quot;pid=&quot;&#43;processID);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}

}


/**
* 鍚&#57882;姩鏍稿績鍔熻兘鐩稿叧鐨勬墍鏈夋敞鍐岀嚎绋�
*/
public static void start() {
// 鍚&#57882;姩绾跨▼
for (int i = 0; i < 2; i&#43;&#43;)
{

DebugLogThread log = new DebugLogThread();
Thread thread = new Thread(log);
thread.start();
threads.add(thread);
}//for

// 鍚&#57882;姩绾跨▼
for (int i = 0; i < 2; i&#43;&#43;)
{

ErrorLogThread log = new ErrorLogThread();
Thread thread = new Thread(log);
thread.start();
threads.add(thread);
}//for

// 鍚&#57882;姩绾跨▼
for (int i = 0; i < 2; i&#43;&#43;)
{

InfoLogThread log = new InfoLogThread();
Thread thread = new Thread(log);
thread.start();
threads.add(thread);
}//for

/**
*   绛夊緟鎵�湁绾跨▼閫�嚭
*/
for (int i = 0; i < threads.size(); i&#43;&#43;)
{
try
{
// threads.get(i).join();
Thread.sleep(1);
} catch (Exception e) {
e.printStackTrace();
}
}

}

public static void main(String[] args) throws ClassNotFoundException {
/**
* 閰嶇疆log4j鐨勯厤缃&#57789;枃浠讹紝娉ㄦ剰杩欐槸xml鏍煎紡銆�
*/
DOMConfigurator.configure(&quot;etc/log4j.xml&quot; );

/**
* 鍙傛暟鍒ゆ柇
*/
if (args.length < 1) {
System.out.println(&quot;args error, args=&quot;&#43;Arrays.toString(args));
System.exit(-1);
}  
System.out.println(&quot;args=&quot;&#43;Arrays.toString(args));

/**
* 璁板綍杩涚▼鍙�
*/
setpid();

start();

logger.debug(&quot;process exit.&quot;);
}
}

　　

　　//使用tail -f XXXX（你监控的文件）  来跟踪日志生成情况；