MQTT的简单Demo

MQTT是物联网收发数据的一种协议

Elipse Paho是一个开源项目实现了MQTT的Clinent可以方便直接操纵数据的上传和下载

The Eclipse Paho project provides open-source client implementations of MQTT and MQTT-SN messaging protocols aimed at new, existing, and emerging applications for the Internet of Things (IoT).

发送数据：

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * 使用同步接口发布一个消息的例子
 * @author Administrator
 * 得到的输出
Connecting to broker: tcp://iot.eclipse.org:1883
Connected
Publishing message: Message from MqttPublishSample
Message published
Disconnected
 *
 */
public class MQTT_Client_send {

    public static void main(String[] args) {
        // TODO Auto-generated method stub

        String topic        = "MQTT Examples";
        String content      = "Message from MqttPublishSample";
        int qos             = 2;
        String broker       = "tcp://iot.eclipse.org:1883";
        String clientId     = "JavaSample";
        MemoryPersistence persistence = new MemoryPersistence();

        try {
            //Create an MqttClient that can be used to communicate with an MQTT server.就是创建client
            MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
            //Constructs a new MqttConnectOptions object using the default values.就是创建连接属性
            MqttConnectOptions connOpts = new MqttConnectOptions();
            //Sets whether the client and server should remember state across restarts and reconnects.就是说这个连接是无记忆的
            connOpts.setCleanSession(true);
            System.out.println("Connecting to broker: "+broker);
            //使用连接属性，建立连接
            sampleClient.connect(connOpts);
            System.out.println("Connected");
            System.out.println("Publishing message: "+content);
            //得到一个操作系统默认的编码格式的字节数组，Constructs a message with the specified byte array as a payload, and all other values set to defaults.,就是创建一个mqtt消息
            MqttMessage message = new MqttMessage(content.getBytes());
            //Sets the quality of service for this message.Quality of Service 2 - indicates that a message should be delivered once.
            message.setQos(qos);
            //调用发布方法，发布一个主题和一个消息
            sampleClient.publish(topic, message);
            System.out.println("Message published");
            //断开连接
            sampleClient.disconnect();
            System.out.println("Disconnected");
            System.exit(0);
        } catch(MqttException me) {
            System.out.println("reason "+me.getReasonCode());
            System.out.println("msg "+me.getMessage());
            System.out.println("loc "+me.getLocalizedMessage());
            System.out.println("cause "+me.getCause());
            System.out.println("excep "+me);
            me.printStackTrace();
        }
    }
    
    

}

接收数据：

import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
public class MQTT_Client_Receive {
    
    public static void Write2Files(String filePath,String text) throws IOException {
        FileWriter fileWriter=new FileWriter(filePath,true);
        fileWriter.write("");
        BufferedWriter bufferedWriter=new BufferedWriter(fileWriter);
        bufferedWriter.append(text+"
");
        bufferedWriter.close();
        fileWriter.close();
    }
    
    public static void Mqtt_Client_2_File(String Topic,String Path) throws MqttException, InterruptedException {
//        String broker="tcp://iotdevrd.chinacloudapp.cn:1889";
//        String clientID="5036cf062ade4a28bf74726e5bff895a";
//        int qos = 2;
//        String topic=Topic;
//        String userName="121";
//        String passWord="121";
        //内网测试数据
        String broker="tcp://3.1.2.244:1889";
        String clientID="5036cf062ade4a28bf74726e5bff895a";
        int qos = 2;
        String topic=Topic;
        String userName="121";
        String passWord="121";
        //内网决赛数据地址
//        String broker="tcp://3.1.2.244:1889";
//        String clientID="5036cf062ade4a28bf74726e5bff895a";
//        int qos = 2;
//        String topic=Topic;
//        String userName="121";
//        String passWord="121";
        
        MemoryPersistence persistence = new MemoryPersistence();
        
        MqttClient myClient=new MqttClient(broker, clientID, persistence);
        MqttConnectOptions connectOptions=new MqttConnectOptions();
        connectOptions.setUserName(userName);
        connectOptions.setPassword(passWord.toCharArray());
        connectOptions.setCleanSession(true);
        
        myClient.connect(connectOptions);
        MqttCallback callback=new MqttCallbackExtended() {
            @Override
            public void messageArrived(String arg0, MqttMessage arg1) throws Exception {
                //String filePath="C:\eclipse-workspace\Projects\IOTProjectPrepare\src\123.txt";
                String filePath=Path;
                Write2Files(filePath, arg1.toString());
                System.out.println(new String(arg1.toString()));
            }
            @Override
            public void deliveryComplete(IMqttDeliveryToken arg0) {
            }
            @Override
            public void connectionLost(Throwable arg0) {
            }
            @Override
            public void connectComplete(boolean arg0, String arg1) {
            }
        };
        myClient.setCallback(callback);
        myClient.subscribe(topic, qos);
        //监听10秒钟最后断开连接
        Thread.sleep(20000);
        if(myClient.isConnected())
            myClient.disconnect();
        //myClient.disconnect();
        //System.exit(0);
        
    }

    
    public static void Topics2Files(int index, String test_final,long time) throws MqttException, InterruptedException {
        
        //循环接收的逻辑
//        List<String> topicList=new ArrayList<>();
////        topicList.add("GPSLocation/test1/1");
////        topicList.add("GPSLocation/test1/2");
////        topicList.add("GPSLocation/test2/1");
////        topicList.add("GPSLocation/test2/2");
//        topicList.add("GPSLocation/test3/1");
//        topicList.add("GPSLocation/test3/2");
////        topicList.add("GPSLocation/test4/1");
////        topicList.add("GPSLocation/test4/2");
//        topicList.add("GPSLocation/test6/1");
//        topicList.add("GPSLocation/test6/2");
//        for (int i = 0; i < topicList.size(); ++i) {
//            String array_element = topicList.get(i);
//            String Path="D:\"+(index)+".json";
//            System.out.println(Path);
//            Mqtt_Client_2_File(array_element, Path);
//        }
        //正式比赛一个一个接收，使用下面的逻辑
        String topic="";
        if(test_final.equals("test")) {
            topic="GPSLocation/test"+index+"/2";
        }else if(test_final.equals("final")) {
            topic="GPULocation/"+index;
        }
        System.out.println("当前订阅路径是："+topic);
        String path="D:\"+(index)+"_"+time+".json";
        System.out.println("当前订阅的保存路径是："+path);
        Mqtt_Client_2_File(topic, path);
        
    }
    public static void main(String[] args) throws MqttException, InterruptedException  {
        //Topics2Files();
        long time=System.currentTimeMillis()/1000;
        Topics2Files(7, "test",time);

    }

}