二、参数

1. 为了安全传输，http source也支持SSL，SSL支持的相关例子可以参见我的关于flume之Avro Source博客

2. Flume 事件使用一个可插拔的“handler”程序来实现转换，它必须实现的HTTPSourceHandler接口。此处理程序需要一个HttpServletRequest和返回一个flume 事件列表。默认是：JSONHandler。

   例如：xxx.handler=com.dxz.flume_demo.source.HTTPSourceXmlHandler

3. 自定义的handler如果想传入参数，可以使用handler.*配置

   如：xxx.handler.myparam=zhangsan

4. 如果配置中没有指定处理程序，HTTP SOURCE将使用与Flume绑定的处理程序，即:JSONHandler,它能处理JSON格式的事件。每个事件可以包含包装为数组的几个事件，尽管Source写入的管道可能有限制的事务能力。

   处理程序接受UTF-8,UTF-16,UTF-32编码的JSON格式的数据，并且将它转换成一个列表的事件。

   格式：

   [ { "headers":{"":"","":""                  },      "body":"the first event"    },    { "headers":{"":"","":""                  },      "body":"the second event"    }     ]

a1.sources=r1

a1.sinks=k1

a1.channels=c1  

a1.sources.r1.type=http

a1.sources.r1.bind=localhost

a1.sources.r1.port=50000

a1.sources.r1.channels=c1  

a1.sinks.k1.type=logger

a1.sinks.k1.channel=c1  

a1.channels.c1.type=memory

a1.channels.c1.capacity=1000

a1.channels.c1.transactionCapacity=100  

flume-ng.cmd agent -conf ../conf -conf-file ../conf/http_source.conf -name a1 -property flume.root.logger=INFO,console

$ curl -X POST -d'[{"headers":{"h1":"v1","h2":"v2"},"body":"hello body"}]'  http://192.168.1.102:50000

<events>

 <event>

     <headers><header1>value1</header1></headers>

     <body>test</body>

 </event>

 <event>

    <headers><header1>value1</header1></headers>

    <body>test2</body>

  </event>

 </events>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

  <modelVersion>4.0.0</modelVersion>

  <groupId>com.dxz</groupId>

  <artifactId>flume-demo</artifactId>

  <version>0.0.1-SNAPSHOT</version>

  <packaging>jar</packaging>

  <name>flume-demo</name>

  <url>http://maven.apache.org</url>

  <properties>

    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

  </properties>

  <dependencies>

    <dependency>

      <groupId>junit</groupId>

      <artifactId>junit</artifactId>

      <version>3.8.1</version>

      <scope>test</scope>

    </dependency>

    <dependency>

      <groupId>org.apache.flume</groupId>

      <artifactId>flume-ng-core</artifactId>

      <version>1.6.0</version>

      <scope>compile</scope>

    </dependency>

  </dependencies>

</project>

package com.dxz.flume_demo.source;

import com.google.common.base.Preconditions;

import org.apache.flume.Context;

import org.apache.flume.Event;

import org.apache.flume.event.EventBuilder;

import org.apache.flume.source.http.HTTPBadRequestException;

import org.apache.flume.source.http.HTTPSourceHandler;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.w3c.dom.Document;

import org.w3c.dom.Element;

import org.w3c.dom.Node;

import org.w3c.dom.NodeList;

import org.xml.sax.SAXException;

import javax.servlet.http.HttpServletRequest;

import javax.xml.parsers.DocumentBuilder;

import javax.xml.parsers.DocumentBuilderFactory;

import java.nio.charset.Charset;

import java.util.ArrayList;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

public class HTTPSourceXMLHandler implements HTTPSourceHandler {

    private final String ROOT = "events";

    private final String EVENT_TAG = "event";

    private final String HEADERS_TAG = "headers";

    private final String BODY_TAG = "body";

    private final String CONF_INSERT_TIMESTAMP = "insertTimestamp";

    private final String TIMESTAMP_HEADER = "timestamp";

    private final DocumentBuilderFactory documentBuilderFactory = DocumentBuilderFactory.newInstance();

    // Document builders are not thread-safe.

    // So make sure we have one for each thread.

    private final ThreadLocal<DocumentBuilder> docBuilder = new ThreadLocal<DocumentBuilder>();

    private boolean insertTimestamp;

    private static final Logger LOG = LoggerFactory.getLogger(HTTPSourceXMLHandler.class);

    public List<Event> getEvents(HttpServletRequest httpServletRequest) throws HTTPBadRequestException, Exception {

        if (docBuilder.get() == null) {

            docBuilder.set(documentBuilderFactory.newDocumentBuilder());

        }

        Document doc;

        final List<Event> events;

        try {

            doc = docBuilder.get().parse(httpServletRequest.getInputStream());

            Element root = doc.getDocumentElement();

            root.normalize();

            // Verify that the root element is "events"

            Preconditions.checkState(ROOT.equalsIgnoreCase(root.getTagName()));

            NodeList nodes = root.getElementsByTagName(EVENT_TAG);

            LOG.info("get nodes={}", nodes);

            int eventCount = nodes.getLength();

            events = new ArrayList<Event>(eventCount);

            for (int i = 0; i < eventCount; i++) {

                Element event = (Element) nodes.item(i);

                // Get all headers. If there are multiple header sections,

                // combine them.

                NodeList headerNodes = event.getElementsByTagName(HEADERS_TAG);

                Map<String, String> eventHeaders = new HashMap<String, String>();

                for (int j = 0; j < headerNodes.getLength(); j++) {

                    Node headerNode = headerNodes.item(j);

                    NodeList headers = headerNode.getChildNodes();

                    for (int k = 0; k < headers.getLength(); k++) {

                        Node header = headers.item(k);

                        // Read only element nodes

                        if (header.getNodeType() != Node.ELEMENT_NODE) {

                            continue;

                        }

                        // Make sure a header is inserted only once,

                        // else the event is malformed

                        Preconditions.checkState(!eventHeaders.containsKey(header.getNodeName()),

                                "Header expected only once " + header.getNodeName());

                        eventHeaders.put(header.getNodeName(), header.getTextContent());

                    }

                }

                Node body = event.getElementsByTagName(BODY_TAG).item(0);

                if (insertTimestamp) {

                    eventHeaders.put(TIMESTAMP_HEADER, String.valueOf(System.currentTimeMillis()));

                }

                System.out.println("httpServletRequest.getCharacterEncoding()="+httpServletRequest.getCharacterEncoding());

                System.out.println("body.getTextContent()=" + body.getTextContent());

                events.add(EventBuilder.withBody(

                        body.getTextContent().getBytes(Charset.defaultCharset()), eventHeaders));

            }

        } catch (SAXException ex) {

            throw new HTTPBadRequestException("Request could not be parsed into valid XML", ex);

        } catch (Exception ex) {

            throw new HTTPBadRequestException(

                    "Request is not in expected format. " + "Please refer documentation for expected format.", ex);

        }

        return events;

    }

    public void configure(Context context) {

        insertTimestamp = context.getBoolean(CONF_INSERT_TIMESTAMP, false);

    }

}

a1.sources=r1

a1.sinks=k1

a1.channels=c1  

a1.sources.r1.type=http

a1.sources.r1.bind=localhost

a1.sources.r1.port=50000

a1.sources.r1.channels=c1

a1.sources.r1.handler=com.dxz.flume_demo.source.HTTPSourceXMLHandler

a1.sources.r1.insertTimestamp=true

a1.sinks.k1.type=logger

a1.sinks.k1.channel=c1  

a1.channels.c1.type=memory

a1.channels.c1.capacity=1000

a1.channels.c1.transactionCapacity=100

flume-ng.cmd agent -conf ../conf -conf-file ../conf/http_source_xml.conf -name a1 -property flume.root.logger=INFO,console