自定义一个数据源,让程序一边读一边写到新的文件里去.

import java.io.*;
import java.nio.charset.Charset;

/**
 *
 */
public class LogReaderAndWriter {

    static String readFileName = "saak";
    static String writeFileName;

    public static void main(String[] args){
        readFileName = args[0];
        writeFileName = args[1];
        readFileByLine(readFileName);
    }

    public static void readFileByLine(String readFileName){
        FileInputStream fis = null;
        InputStreamReader isr = null;
        BufferedReader in = null;
        String tmpStr = null;
        try {
            System.out.println("一行一行读取文件..............");
            fis = new FileInputStream(new File(readFileName));
            isr = new InputStreamReader(fis, "GBK");
            in = new BufferedReader(isr);
            int count = 0;
            while((tmpStr = in.readLine())!= null){
                count++;
                Thread.sleep(300);
//                String str = new String(tmpStr.getBytes("UTF-8"),"GBK");
                System.out.println("row:"+count+">>>>>>>>>>>>>>>>"+tmpStr);
                method(writeFileName,tmpStr);
            }
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            try {
                if(fis != null) {
                    fis.close();
                }else if(isr!=null){
                    isr.close();
                }else if(in != null){
                    in.close();
                }

            } catch (IOException e) {
                e.printStackTrace();
            }

        }
    }

    private static void method(String writeFileName, String tmpStr) {
        BufferedWriter out = null;
        try {
            out = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(writeFileName,true)));
            out.write("\n");
            out.write(tmpStr);
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            if (out != null) {
                try {
                    out.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

达成jar包放到flume节点中

2 编写启动的脚本

给出2个参数,一个是源文件一个是输出文件

image.png

3编写flume启动的脚本

image.png

#bin/bash
echo "flume-2 start ........................................"

#bin/flume-ng agent --conf conf -f conf/flume-conf.properties -n agent2 -Dflume.root.logger=DEBUG,console

bin/flume-ng agent --conf conf --conf-file conf/flume-conf.properties --name agent2 -Dflume.root.logger=INFO,co
nsole

4编写kafka的启动consumer脚本

image.png