一略就、簡述
MQTT（Message Queuing Telemetry Transport，消息隊列遙測傳輸協(xié)議）藏澳，是一種基于發(fā)布/訂閱（publish/subscribe）模式的“輕量級”通訊協(xié)議，該協(xié)議構(gòu)建于TCP/IP協(xié)議上，由IBM在1999年發(fā)布颈嚼。MQTT最大優(yōu)點在于，可以以極少的代碼和有限的帶寬饭寺，為連接遠(yuǎn)程設(shè)備提供實時可靠的消息服務(wù)阻课。作為一種低開銷、低帶寬占用的即時通訊協(xié)議艰匙，使其在物聯(lián)網(wǎng)限煞、小型設(shè)備、移動應(yīng)用等方面有較廣泛的應(yīng)用员凝。
MQTT是一個基于客戶端-服務(wù)器的消息發(fā)布/訂閱傳輸協(xié)議署驻。MQTT協(xié)議是輕量、簡單健霹、開放和易于實現(xiàn)的旺上，這些特點使它適用范圍非常廣泛。在很多情況下糖埋，包括受限的環(huán)境中宣吱，如：機器與機器（M2M）通信和物聯(lián)網(wǎng)（IoT）。其在瞳别，通過衛(wèi)星鏈路通信傳感器征候、偶爾撥號的醫(yī)療設(shè)備、智能家居洒试、及一些小型化設(shè)備中已廣泛使用倍奢。
二、apache-Apollo 服務(wù)器搭建
下載地址：http://activemq.apache.org/apollo/download.html
-
1.將下載后的 壓縮包進(jìn)行解壓:
apollo.png -
2.打開 cmd 運行 bin/apollo.cmd
apollo.png -
3.創(chuàng)建一個服務(wù)器實例
image.png -
4.在服務(wù)中打開 apollo 服務(wù)
image.png -
5.啟動服務(wù)后垒棋，在瀏覽器上面輸入：https://127.0.0.1:61681/或http://127.0.0.1:61680/
apollo控制臺
初始賬號: admin 密碼:password
-
6.進(jìn)入apollo控制臺
image.png
進(jìn)入此頁面說明服務(wù)器成功搭建
三卒煞、服務(wù)端推送和客戶端監(jiān)聽
- SeverClass
package com.mqtt;
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.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
public class ServerMQTT {
//tcp://MQTT安裝的服務(wù)器地址:MQTT定義的端口號
public static final String HOST = "tcp://0.0.0.0:61613";
//定義一個主題
public static final String TOPIC = "message";
//定義MQTT的ID，可以在MQTT服務(wù)配置中指定
private static final String clientid = "server11";
private MqttClient client;
private MqttTopic topic11;
private String userName = "admin"; //非必須
private String passWord = "password"; //非必須
private MqttMessage message;
/**
* 構(gòu)造函數(shù)
* @throws MqttException
*/
public ServerMQTT() throws MqttException {
// MemoryPersistence設(shè)置clientid的保存形式叼架，默認(rèn)為以內(nèi)存保存
client = new MqttClient(HOST, clientid, new MemoryPersistence());
connect();
}
/**
* 用來連接服務(wù)器
*/
private void connect() {
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(false);
options.setUserName(userName);
options.setPassword(passWord.toCharArray());
// 設(shè)置超時時間
options.setConnectionTimeout(10);
// 設(shè)置會話心跳時間
options.setKeepAliveInterval(20);
try {
client.setCallback(new PushCallback());
client.connect(options);
topic11 = client.getTopic(TOPIC);
} catch (Exception e) {
e.printStackTrace();
}
}
/**
*
* @param topic
* @param message
* @throws MqttPersistenceException
* @throws MqttException
*/
public void publish(MqttTopic topic , MqttMessage message) throws MqttPersistenceException,
MqttException {
MqttDeliveryToken token = topic.publish(message);
token.waitForCompletion();
System.out.println("message is published completely! "
+ token.isComplete());
}
/**
* 啟動入口
* @param args
* @throws MqttException
*/
public static void main(String[] args) throws MqttException {
ServerMQTT server = new ServerMQTT();
server.message = new MqttMessage();
server.message.setQos(1); //保證消息能到達(dá)一次
server.message.setRetained(true);
server.message.setPayload("這是服務(wù)器發(fā)出的信息的內(nèi)容".getBytes());
server.publish(server.topic11 , server.message);
System.out.println(server.message.isRetained() + "------ratained狀態(tài)");
}
}
- ClientClass
package com.mqtt;
import java.util.concurrent.ScheduledExecutorService;
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.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
public class ClientMQTT {
public static final String HOST = "tcp://0.0.0.0:61613";
public static final String TOPIC1 = "message";
private static final String clientid = "client11";
private MqttClient client;
private MqttConnectOptions options;
private String userName = "admin"; //非必須
private String passWord = "password"; //非必須
@SuppressWarnings("unused")
private ScheduledExecutorService scheduler;
private void start() {
try {
// host為主機名畔裕，clientid即連接MQTT的客戶端ID，一般以唯一標(biāo)識符表示乖订，MemoryPersistence設(shè)置clientid的保存形式扮饶，默認(rèn)為以內(nèi)存保存
client = new MqttClient(HOST, clientid, new MemoryPersistence());
// MQTT的連接設(shè)置
options = new MqttConnectOptions();
// 設(shè)置是否清空session,這里如果設(shè)置為false表示服務(wù)器會保留客戶端的連接記錄，設(shè)置為true表示每次連接到服務(wù)器都以新的身份連接
options.setCleanSession(false);
// 設(shè)置連接的用戶名
options.setUserName(userName);
// 設(shè)置連接的密碼
options.setPassword(passWord.toCharArray());
// 設(shè)置超時時間 單位為秒
options.setConnectionTimeout(10);
// 設(shè)置會話心跳時間 單位為秒 服務(wù)器會每隔1.5*20秒的時間向客戶端發(fā)送個消息判斷客戶端是否在線乍构，但這個方法并沒有重連的機制
options.setKeepAliveInterval(20);
// 設(shè)置回調(diào)
client.setCallback(new PushCallback());
MqttTopic topic = client.getTopic(TOPIC1);
//setWill方法甜无，如果項目中需要知道客戶端是否掉線可以調(diào)用該方法。設(shè)置最終端口的通知消息
//options.setWill(topic, "close".getBytes(), 2, true);
client.connect(options);
//訂閱消息
//訂閱消息
client.subscribe(TOPIC1, 1);
} catch (Exception e) {
e.printStackTrace();
}
}
public static void main(String[] args) throws MqttException {
ClientMQTT client = new ClientMQTT();
client.start();
}
}
-
server
server -
client
client
