flume接收http请求，并将数据写到kafka

flume接收http请求，并将数据写到kafka，spark消费kafka的数据。是数据采集的经典框架。

直接上flume的配置：

source : http

channel : file

sink : kafka

xx :~/software/flume1.8/conf$ cat http-file-kafka.conf 
# example.conf: A single-node Flume configuration
##########
# data example 
# use post request, select raw, data example : [{"body" : "{'xx':'xxxxx1'}"}]
# just use the office request demo

#[{
#  "headers" : {
#             "timestamp" : "434324343",
#             "host" : "random_host.example.com"
#             "topic" : "venn" # if headers contain topic, will replace the default topic
#             },
#  "body" : "random_body" # random_body is the message send to channel
#  }]

# Name the components on this agent1
agent1.sources = s1
agent1.sinks = k1
agent1.channels = c1
# Describe/configure the source
agent1.sources.s1.type = http
agent1.sources.s1.bind = spring  # localhost 只能接收本地请求
agent1.sources.s1.port = 8084  # http的端口
agent1.sources.s1.handler = org.apache.flume.source.http.JSONHandler  # 自带的接收http请求的handler
# Describe the sink
agent1.sinks.k1.type =  org.apache.flume.sink.kafka.KafkaSink  # kafkasink
agent1.sinks.k1.kafka.topic = mytopic  # topic
agent1.sinks.k1.kafka.bootstrap.servers = localhost:9092  # kafka host and port
agent1.sinks.k1.kafka.flumeBatchSize = 20
agent1.sinks.k1.kafka.producer.acks = 1
agent1.sinks.k1.kafka.producer.linger.ms = 1
agent1.sinks.k1.kafka.producer.compression.type = snappy  # 压缩
# Use a channel which buffers events in memory
agent1.channels.c1.type = file
#agent1.channels.c1.capacity = 1000  # 这两个参数要配置，需要配大一点，不然channel满了会报错，http返回503（通道已满）
#agent1.channels.c1.transactionCapacity = 100
agent1.channels.c1.checkpointDir = /opt/flume/checkpoint  
agent1.channels.c1.dataDirs = /opt/flume/channel
# Bind the source and sink to the channel
agent1.sources.s1.channels = c1
agent1.sinks.k1.channel = c1

有了flume的配置，下面启动flume：

./bin/flume-ng agent -n agent1 -c conf -f conf/http-to-kafka.properties -Dflume.root.logger=INFO,console

启动之后，就可以发http请求了。

http请求的格式如下：

[{
  "headers" : {
             "timestamp" : "434324343",
             "host" : "random_host.example.com",
             "topic" : "xxx"
             },
  "body" : "random_body"
  },
  {
  "headers" : {
             "namenode" : "namenode.example.com",
             "datanode" : "random_datanode.example.com"
             },
  "body" : "really_random_body"
  }]    

注： http请求的headers中又topic 会替代配置文件中的topic

　　flume官网文档说：1.8.0版本的flume只支持0.9.x的kafka，不支持0.8.x的kafka了（没测过）

然后就是发数的程序了（自己请求太麻烦了。）

package com.venn.http;

import com.venn.entity.User;
import java.io.BufferedReader;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.*;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.event.JSONEvent;
import com.google.gson.Gson;
import org.apache.flume.source.http.HTTPBadRequestException;
import org.apache.flume.source.http.HTTPSourceHandler;

import javax.servlet.http.HttpServletRequest;


/**
 * Created by venn on 19-1-17.
 */
public class HttpDemo  {

    private static String urlStr = "http://localhost:8084";
    private static Random random = new Random();
    public static void main(String[] args) throws InterruptedException {

        while (true){
            String message = new User().toString();
            send(message);

//            Thread.sleep(1);
        }
    }
    public static void send(String message){
        System.out.println("send  message : " + message);
        try{
            //创建连接
            URL url = new URL(urlStr);
            HttpURLConnection connection = (HttpURLConnection) url.openConnection();
            connection.setDoOutput(true);
            connection.setDoInput(true);
            connection.setRequestMethod("POST");
            connection.setUseCaches(false);
            connection.setInstanceFollowRedirects(true);
            connection.setRequestProperty("Content-Type",
                    "application/x-www-form-urlencoded");
            connection.connect();
            //POST请求
            DataOutputStream out = new DataOutputStream(
                    connection.getOutputStream());

            JSONEvent jsonEvent = new JSONEvent();
            Map header = new HashMap();
            header.put("timestamp", System.currentTimeMillis());
            header.put("host", "venn");
            header.put("topic","venn"+random.nextInt(4));
            jsonEvent.setBody(message.getBytes());
            jsonEvent.setHeaders(header);
            Gson gson = new Gson();
            List list = new ArrayList();
            list.add(jsonEvent);
            out.writeBytes(gson.toJson(list));
            out.flush();
            out.close();

            //读取响应
            BufferedReader reader = new BufferedReader(new InputStreamReader(
                    connection.getInputStream())); // 不会返回数据
            int code = connection.getResponseCode();
            String lines;
            StringBuffer sb = new StringBuffer("");
            while ((lines = reader.readLine()) != null) {
                lines = new String(lines.getBytes(), "utf-8");
                sb.append(lines);
            }
            System.out.println("code : " + code + ", message : " + sb);
            reader.close();
            // 断开连接
            connection.disconnect();
    } catch (MalformedURLException e) {
        e.printStackTrace();
    } catch (UnsupportedEncodingException e) {
        e.printStackTrace();
    } catch (IOException e) {
        e.printStackTrace();
    }
    }

}

搞定。。

发数：

 kafka接收到的数据：

注意： 由于在headers中加入了topic参数，实际接收到的数据是在不同的kafka topic中的