物聯網消息隊列協議MQTT

簡介
Mqtt是一個物聯網消息傳輸協議

mosquitto是mqtt協議的一個開源實現，http://mosquitto.org/

paho是mqtt協議的客戶端實現，這裏主要用paho的mqtt java客戶端jar包。

http://wiki.eclipse.org/Paho

只須要在pom.xml中增長：

<dependency>
　　<groupId>org.eclipse.paho</groupId>
　　<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
　　<version>1.0.2</version>
</dependency>

 

mosquitto能夠理解爲mutt協議的server端，可部署在linux及windows上

CentOS的安裝過程：

參照官網文檔，在yum的repo配置中增長：

[home_oojah_mqtt]
name=mqtt (CentOS_CentOS-6)
type=rpm-md
baseurl=http://download.opensuse.org/repositories/home:/oojah:/mqtt/CentOS_CentOS-6/
gpgcheck=1
gpgkey=http://download.opensuse.org/repositories/home:/oojah:/mqtt/CentOS_CentOS-6//repodata/repomd.xml.key
enabled=1

 

mosquitto的配置文件項可參考：http://blog.csdn.net/shagoo/article/details/7910598

集羣部署 參照http://www.cnblogs.com/yinyi521/p/6087215.html

而後執行：
yum install mosquitto
yum install mosquitto-client

啓動mosquitto服務：

mosquitto -v

 

 

mqtt java客戶端實例：

package com.haibran.mqtt_demo;

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 MqttPublicSample {

public static void main(String[] args) {
// TODO Auto-generated method stub
String topic = "sensor";
String content = "Message from MqttPublishSample";
int qos = 2;
String broker = "tcp://192.168.0.20:1883";
String clientId = "JavaSample";
MemoryPersistence persistence = new MemoryPersistence();

try {
MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
MqttConnectOptions connOpts = new MqttConnectOptions();
connOpts.setCleanSession(true);
System.out.println("Connecting to broker: "+broker);
sampleClient.connect(connOpts);
System.out.println("Connected");
System.out.println("Publishing message: "+content);
MqttMessage message = new MqttMessage(content.getBytes());
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();
}
}

}