package com.flume.flume;

import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;

public class MySink extends AbstractSink implements Configurable {

    private String msgPrefix;

    /**
     * 用來保存數(shù)據(jù),不斷調(diào)用次方法
     * @return
     * @throws EventDeliveryException
     */
    @Override
    public Status process() throws EventDeliveryException {
        //獲取sink對應(yīng)的channnel
        Channel channel = getChannel();
        Connection connection = null;
        PreparedStatement statement = null;
        //獲取事務(wù)對象
        Transaction transaction = channel.getTransaction();
        try{
            //開啟事務(wù)
            transaction.begin();
            //從channel中獲取數(shù)據(jù)
            Event event = channel.take();

            //切割數(shù)據(jù)
            String data = new String(event.getBody());
            String[] arr = data.split(",");
            String id = arr[0];
            String name = arr[1];
            int age = Integer.parseInt(arr[2]);

            //保存到mysql
            //1、獲取connect
            connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/test?useSSL=false","root","123321");
            statement = connection.prepareStatement("insert into test values(?,?,?)");
            saveToMysql(id,name,age,connection,statement);
            //模擬數(shù)據(jù)保存
            //System.out.println(msgPrefix+":"+new String(take.getBody()));
            //提交事務(wù)
            transaction.commit();

            return Status.READY;
        }catch (Exception e){
            transaction.rollback();
        }finally {
            //關(guān)閉事務(wù)
            transaction.close();
            if(statement!=null)
            //5、關(guān)閉
            {
                try {
                    statement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            if(connection!=null) {
                try {
                    connection.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }

        return Status.BACKOFF;
    }

    public void saveToMysql(String id,String name,int age,Connection connection,PreparedStatement statement) throws SQLException {

        //2、獲取statement對象
        //sql注入 【 select * from table where name='zhangsan' or 1=1】
        //connection.createStatement();

        //3、賦值
        statement.setString(1,id);
        statement.setString(2,name);
        statement.setInt(3,age);
        System.out.println(id+","+name+","+age);
        //4、保存
        statement.executeUpdate();


    }
    /**
     * 獲取sink的配置屬性
     * @param context
     */
    @Override
    public void configure(Context context) {

        msgPrefix = context.getString("msg.prefix");

    }
}

#定義agent
a1.sources = r1
a1.channels = c1
a1.sinks = k1

#定義source
a1.sources.r1.type = netcat
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 9999

#定義channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 1000

#定義sink
a1.sinks.k1.type = com.atguigu.flume.MySink
a1.sinks.k1.msg.prefix = message

#定義source、channel、sink之間的綁定關(guān)系
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

[hadoop@hadoop102 ~]$ cd /opt/module/flume/
[hadoop@hadoop102 flume]$ bin/flume-ng agent -c conf/ -n a1 -f job/mysik.config -Dflume.root.logger=INFO,console

[hadoop@hadoop102 ~]$ nc hadoop102 9999
1,ttt,8
OK