前言：接上一篇法竞，本篇文章實(shí)現(xiàn)一個(gè)自定義的攔截器撵渡。主要功能是在Event的body中添加IP地址雷则。因?yàn)闆](méi)有攔截器可以在Body中添加（host是在header中添加）郎楼，所以需要自定義万伤。掌握了這個(gè)，其他的情況根據(jù)自己的業(yè)務(wù)需求去寫(xiě)即可呜袁。

001.png

我又是前言: 說(shuō)一下學(xué)習(xí)自定義攔截器方法吧敌买，導(dǎo)入flume的源碼后，看flume-ng-core-xxx.jar下的org.apache.flume.interceptor
包阶界，里面有flume為我們寫(xiě)好的一些攔截器放妈，我們只需要仿照這些類(lèi)去寫(xiě)即可北救，同時(shí)可以查看接口的javadoc知道要重寫(xiě)的方法是什么作用。

一芜抒、流程

①搭建flume開(kāi)發(fā)環(huán)境（巧婦難為無(wú)米之炊珍策，你沒(méi)開(kāi)發(fā)環(huán)境怎么玩，程序都不知道你寫(xiě)的類(lèi)是個(gè)啥）
②新建一個(gè)類(lèi)宅倒，實(shí)現(xiàn)Interceptor接口攘宙，重寫(xiě)intercept(Event event)方法
③新建一個(gè)類(lèi)，實(shí)現(xiàn)Interceptor.Builder接口拐迁，重寫(xiě)configure(Context context)和build()方法
④打成jar包放到flume的lib目錄下
⑤編寫(xiě)相應(yīng)的flume.conf文件蹭劈，將type值使用類(lèi)的全限定名指定我們的攔截器。如果有自定義屬性线召，需要配置該自定義屬性铺韧。

二、搭建開(kāi)發(fā)環(huán)境

????新建一個(gè)maven工程缓淹，在pom.xml中添加如下依賴(lài)（我這里的版本號(hào)和我的flume版本號(hào)一致）：

<dependencies>
    <dependency>
        <groupId>org.apache.flume</groupId>
        <artifactId>flume-ng-sdk</artifactId>
        <version>1.8.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flume</groupId>
        <artifactId>flume-ng-core</artifactId>
        <version>1.8.0</version>
    </dependency>
</dependencies>

二哈打、編寫(xiě)自定義Interceptor和自定義Builder

????代碼不難，希望大家不要有恐懼心理讯壶，一看到代碼段就腦殼疼料仗。

package com.zhb.flume;

import java.util.List;

import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import com.google.common.base.Charsets;

public class MyInterceptor implements Interceptor {
    private String ipAddress = null;

    // 自定義屬性 serviceId
    public MyInterceptor(String ipAddress) {
        this.ipAddress = ipAddress;
    }

    public void initialize() {
        // TODO Auto-generated method stub

    }
       //攔截器的核心
    public Event intercept(Event event) {
        //獲得body的內(nèi)容
        String eventBody = new String(event.getBody(), Charsets.UTF_8);
        String fmt = "%s %s";
        // 添加ipAddress 到event的開(kāi)頭
        event.setBody(String.format(fmt, ipAddress, eventBody).getBytes());
        return event;
    }

    public List<Event> intercept(List<Event> events) {
        for (Event event : events) {
            intercept(event);
        }
        return events;
    }

    public void close() {
        // TODO Auto-generated method stub
    }

}

package com.zhb.flume;

import org.apache.flume.Context;
import org.apache.flume.interceptor.Interceptor;

public class AppendIPBuilder implements Interceptor.Builder {

    private String ipAddress = null;

    public void configure(Context context) {
        // set argument serviceId
        String configServiceId = context.getString("ipAddress");
        ipAddress = configServiceId;
    }

    public Interceptor build() {
        
        return new MyInterceptor(ipAddress);
    }

}

三、編寫(xiě)flume的配置文件appendIP.conf

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
 
# Describe/configure the source
a1.sources.r1.type = syslogtcp
a1.sources.r1.port = 50000
a1.sources.r1.host = 0.0.0.0
a1.sources.r1.channels = c1
 
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.zhb.flume.AppendIPBuilder
#這里配置的值和Builder里的變量名字要一樣
a1.sources.r1.interceptors.i1.ipAddress= 192.168.1.101

# Describe the sink
a1.sinks.k1.type = logger
a1.sinks.k1.channel = c1
 
# Use a channel which buffers events inmemory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

四伏蚊、運(yùn)行flume進(jìn)行測(cè)試

????首先進(jìn)入到flume的bin目錄下立轧，執(zhí)行./flume-ng agent -c ../conf -f ../conf/appendIP.conf -Dflume.root.logger=INFO,console -n a1
????成功啟動(dòng)后，新開(kāi)一個(gè)終端輸入echo "AppendIPAddress" | nc 127.0.0.1 50000
????這時(shí)躏吊，flume啟動(dòng)的終端的小齒輪轉(zhuǎn)了起來(lái)氛改，會(huì)心一笑，一定是成功了比伏。

002.jpg

003.png

004.jpg

五观挎、總結(jié)
????本文實(shí)現(xiàn)了自定義攔截器，在body中添加ip地址段化，實(shí)際工作中嘁捷，根據(jù)各種不同需求編寫(xiě)符合自己需求的攔截器即可。下一篇寫(xiě)自定義Sink显熏，解決剛才的問(wèn)題雄嚣。