flume与Mosquitto的集成

文章来自：http://www.cnblogs.com/hark0623/p/4173714.html   转发请注明 

因业务需求，需要flume收集MQTT(Mosquitto)的数据。  方法就是flume自定义source，source中来订阅(subscribe)MQTT

flume source的java代码如下：

package com.yhx.sensor.flume.source;

import java.util.HashMap;
import java.util.Map;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.AbstractSource;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;

public class MQTTSource extends AbstractSource implements EventDrivenSource,
        Configurable {
    /**
     * The initialization method for the Source. The context contains all the
     * Flume configuration info, and can be used to retrieve any configuration
     * values necessary to set up the Source.
     */
    @Override
    public void configure(Context arg0) {
        // TODO Auto-generated method stub

    }

    SimpleMqttClient client = null;

    /**
     * Start any dependent systems and begin processing events.
     */
    @Override
    public void start() {
        // TODO Auto-generated method stub
        // super.start();
        client = new SimpleMqttClient();
        client.runClient();
    }

    /**
     * Stop processing events and shut any dependent systems down.
     */
    @Override
    public void stop() {
        // TODO Auto-generated method stub
        // super.stop();
        if (client != null) {
            client.closeConn();
        }
    }

    // public static void main(String[] args) {
    // SimpleMqttClient smc = new SimpleMqttClient();
    // smc.runClient();
    // }

    public class SimpleMqttClient implements MqttCallback {

        MqttClient myClient;
        MqttConnectOptions connOpt;

        String BROKER_URL = "tcp://192.168.116.128:1883";
        String M2MIO_DOMAIN = "192.168.116.128";
        String M2MIO_STUFF = "yhx";
        String M2MIO_THING = "yhx_flume";
        // String M2MIO_USERNAME = "<m2m.io username>";
        // String M2MIO_PASSWORD_MD5 =
        // "<m2m.io password (MD5 sum of password)>";

        Boolean subscriber = true;
        Boolean publisher = false;

        /**
         * 
         * connectionLost This callback is invoked upon losing the MQTT
         * connection.
         * 
         */
        @Override
        public void connectionLost(Throwable t) {
            System.out.println("Connection lost!");
            // code to reconnect to the broker would go here if desired
        }

        public void closeConn() {
            if (myClient != null) {
                if (myClient.isConnected()) {
                    try {
                        myClient.disconnect();
                    } catch (MqttException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            }
        }

        /**
         * 
         * deliveryComplete This callback is invoked when a message published by
         * this client is successfully received by the broker.
         * 
         */
        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {
            // System.out.println("Pub complete" + new
            // String(token.getMessage().getPayload()));
        }

        /**
         * 
         * messageArrived This callback is invoked when a message is received on
         * a subscribed topic.
         * 
         */
        @Override
        public void messageArrived(String topic, MqttMessage message)
                throws Exception {
            // System.out
            // .println("-------------------------------------------------");
            // // System.out.println("| Topic:" + topic.getName());
            // System.out.println("| Topic:" + topic);
            // System.out
            // .println("| Message: " + new String(message.getPayload()));
            // System.out
            // .println("-------------------------------------------------");


            Map<String, String> headers = new HashMap<String, String>();
            //headers.put("curDate", df.format(new Date()));

            Event flumeEvent = EventBuilder.withBody(message.getPayload(),
                    headers);
            try {
                getChannelProcessor().processEvent(flumeEvent);
            } catch (Exception e) {
                // TODO: handle exception
                e.printStackTrace();
            }

        }

        /**
         * 
         * runClient The main functionality of this simple example. Create a
         * MQTT client, connect to broker, pub/sub, disconnect.
         * 
         */
        public void runClient() {
            // setup MQTT Client
            String clientID = M2MIO_THING;
            connOpt = new MqttConnectOptions();

            connOpt.setCleanSession(true);
            connOpt.setKeepAliveInterval(3000);
            // connOpt.setUserName(M2MIO_USERNAME);
            // connOpt.setPassword(M2MIO_PASSWORD_MD5.toCharArray());

            // Connect to Broker
            try {
                myClient = new MqttClient(BROKER_URL, clientID);
                myClient.setCallback(this);
                myClient.connect(connOpt);
            } catch (MqttException e) {
                e.printStackTrace();
                System.exit(-1);
            }

            System.out.println("Connected to " + BROKER_URL);

            // setup topic
            // topics on m2m.io are in the form <domain>/<stuff>/<thing>
            String myTopic = M2MIO_DOMAIN + "/" + M2MIO_STUFF + "/"
                    + M2MIO_THING;
            System.out.println("myTopic:" + myTopic);
            MqttTopic topic = myClient.getTopic(myTopic);

            // subscribe to topic if subscriber
            if (subscriber) {
                try {
                    int subQoS = 0;
                    myClient.subscribe(myTopic, subQoS);

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }

            // publish messages if publisher
            if (publisher) {
                for (int i = 1; i <= 10; i++) {
                    String pubMsg = "{"pubmsg":" + i + "}";
                    int pubQoS = 0;
                    MqttMessage message = new MqttMessage(pubMsg.getBytes());
                    message.setQos(pubQoS);
                    message.setRetained(false);

                    // Publish the message
                    System.out.println("Publishing to topic "" + topic
                            + "" qos " + pubQoS);
                    MqttDeliveryToken token = null;
                    try {
                        // publish message to broker
                        token = topic.publish(message);
                        // Wait until the message has been delivered to the
                        // broker
                        token.waitForCompletion();
                        Thread.sleep(100);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }

            // disconnect
            try {
                // wait to ensure subscribed messages are delivered
                if (subscriber) {
                    while (true) {
                        Thread.sleep(5000);
                    }
                }
                // myClient.disconnect();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
            }
        }

    }

}

打JAR包注意要把Class-Path写上，如下：

Manifest-Version: 1.0
Class-Path: flume-ng-configuration-1.5.2.jar flume-ng-core-1.5.2.jar flume-ng-node-1.5.2.jar flume-ng-sdk-1.5.2.jar org.eclipse.paho.client.mqttv3-1.0.0.jar

将打好的JAR包放到flume的lib目录（注意，class-path说明的jar包在lib一定要有。 如果没有，则放上去）

接着修改一下flume的配置文件，如下(主要是sourceMqtt ，看这个。  因为我这块同时还监听了UDP)：

a1.sources = sourceMqtt sourceUdp
a1.sinks = sinkMqtt sinkUdp
a1.channels = channelMqtt channelUdp

# Describe/configure the source
a1.sources.sourceMqtt.type = com.yhx.sensor.flume.source.MQTTSource

# Describe the sink
a1.sinks.sinkMqtt.type = logger

# Use a channel which buffers events in memory
a1.channels.channelMqtt.type = memory
a1.channels.channelMqtt.capacity = 1000
a1.channels.channelMqtt.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.sourceMqtt.channels = channelMqtt
a1.sinks.sinkMqtt.channel = channelMqtt



# a2.sources = sourceUdp
# a2.sinks = sinkUdp
# a2.channels = channelUdp

# Describe/configure the source
a1.sources.sourceUdp.type = syslogudp
a1.sources.sourceUdp.host = 0.0.0.0
a1.sources.sourceUdp.port = 12459
a1.sources.sourceUdp.interceptors=interceptorUdp

a1.sources.sourceUdp.interceptors.interceptorUdp.type=com.yhx.sensor.flume.intercepter.UDPIntercepter$Builder

# Describe the sink
a1.sinks.sinkUdp.type = logger

# Use a channel which buffers events in memory
a1.channels.channelUdp.type = memory
a1.channels.channelUdp.capacity = 1000
a1.channels.channelUdp.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.sourceUdp.channels = channelUdp
a1.sinks.sinkUdp.channel = channelUdp

配置文件保存至flume目录下的conf,叫flume.conf

然后flume启动命令如下 

bin/flume-ng agent --conf conf --conf-file conf/flume.conf --name a1