flume(사용자 정의 차단기)

1.pom에서.xml 파일에서 가져오기 의존

 
        <dependency>
            <groupId>org.apache.flumegroupId>
            <artifactId>flume-ng-coreartifactId>
            <version>1.7.0version>
        dependency>

2. 코드

import com.google.gson.Gson;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


public class MyInterceptor implements Interceptor {
    public static final String SELECTOER_HEADER ="logType";
    private Logger logger = LoggerFactory.getLogger(MyInterceptor.class);
    @Override  //   
    public void initialize() {

    }


    @Override  //       type    header       
    public Event intercept(Event event) {
        Gson gson = new Gson();
        String logString = new String(event.getBody());
        if(logString!=null&&logString.length()>0){
            HashMap<String,String> hashMap = gson.fromJson(logString, HashMap.class);
            String type = hashMap.get("type");
            //      
            if(type.equals("startup")){
                Map<String, String> headers = event.getHeaders();
                headers.put(SELECTOER_HEADER ,"startup"); //     
            }else if (type.equals("event")){
                Map<String, String> headers = event.getHeaders();
                headers.put(SELECTOER_HEADER,"event");
            }
        }
        return  event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {

        List<Event> intercepted = new ArrayList<>(events.size());
        for (Event event : events) {
            Event interceptedEvent = intercept(event);
            if (interceptedEvent != null) {
                intercepted.add(interceptedEvent);
            }
        }
        return intercepted;
    }

    @Override
    public void close() {

    }


    /**
     *                  flume  ，  Interceptor.Builder  ，        
     */
    public static class Builder implements Interceptor.Builder {
        /**
         *                      
         * @return
         */
        @Override
        public Interceptor build() {
            return new MyInterceptor();
        }

        @Override
        public void configure(Context context) {
                //    context   flume.conf       ，   Interceptor
        }
    }


    public static void main(String[] args) {
        Gson gson = new Gson();
        String logString="{\"area\":\"shanghai\",\"uid\":\"user_995\",\"os\":\"andriod\",\"appid\":\"app007\",\"mid\":\"88a147a9-ecc4-40be-885a-ae67bb9b7b2b\",\"type\":\"startup\",\"vs\":\"1.1.1\",\"ts\":1546780968866}";
        HashMap<String,String> hashMap = gson.fromJson(logString, HashMap.class);

        System.out.println(hashMap.get("type"));

    }

}

3. 작성한 코드를jar백으로 만들고 작성한jar백을flume/lib/아래로 복사