MQTT：
MQ 遙測傳輸 (MQTT) 是輕量級基于代理的發(fā)布/訂閱的消息傳輸協(xié)議兔院，設(shè)計思想是開放、簡單排吴、輕量秆乳、易于實現(xiàn)。這些特點使它適用于受限環(huán)境。該協(xié)議的特點有：
使用發(fā)布/訂閱消息模式屹堰，提供一對多的消息發(fā)布肛冶，解除應(yīng)用程序耦合。
一扯键、使用Apollo 搭建MQTT服務(wù)器
這里使用的Apollo 1.7.1睦袖。
官網(wǎng)：http://activemq.apache.org/index.html
下載地址：http://activemq.apache.org/apollo/download.html
快速開始教程：http://activemq.apache.org/apollo/documentation/getting-started.html
使用Apollo搭建MQTT服務(wù)器步驟：
下載Apollo服務(wù)器并解壓，在CMD環(huán)境運行其工作目錄下的...\bin\apollo.cmd荣刑，命令后面帶上參數(shù)「create mybroker」馅笙，創(chuàng)建服務(wù)器實例。這里需要Java環(huán)境厉亏，系統(tǒng)環(huán)境變量下要有JAVA_HOME董习。
創(chuàng)建實例之后會在bin目錄下生成mybroker文件夾，其中 ...\etc\apollo.xml文件下是配置服務(wù)器信息的文件爱只，...\etc\users.properties文件包含連接MQTT服務(wù)器時用到的用戶名和密碼皿淋，初始默認(rèn)帳號是admin，密碼password恬试；
進入...\mybroker\bin\ 目錄窝趣，在CMD輸入命令「apollo-broker.cmd run」，可以使用TAB鍵自動補全训柴，運行后輸出信息如下：
image.png
其中我們要留意的：
1.阿里云要在安全組配置tcp端口
2.windows服務(wù)器配置入站規(guī)則
3.MQTT服務(wù)器TCP連接端口：tcp://0.0.0.0:61613
4.后臺Web管理頁面：https://127.0.0.1:61681/或http://127.0.0.1:61680/
登錄服務(wù)器后哑舒，如果MQTT服務(wù)器有客戶端連接，后臺會顯示如下
image.png
發(fā)送：
public class ServerMQTT {
private static boolean finish=true;
private static Charset utf8 = Charset.forName("utf8");
/**
* 啟動入口
*/
public static void main(String[] args) throws MqttException {
final MqttConnect HFMqttConnect = new MqttConnect("tcp://ip:61613","*","*");
Timer timer = new Timer();
timer.schedule(new TimerTask() {
@Override
public void run() {
try {
setMessage(HFMqttConnect);
System.out.println("setMessage");
} catch (MqttException e) {
e.printStackTrace();
}
}
},0,10000);
}
private static void setMessage(MqttConnect mqttConnect1) throws MqttException {
mqttConnect1.message = new MqttMessage();
mqttConnect1.message.setQos(1);
mqttConnect1.message.setRetained(true);//禁用遺囑
mqttConnect1.publish(mqttConnect1.topic11, mqttConnect1.message);
}}
MqttCOnnect
class MqttConnect {
private MqttClient client;
MqttTopic topic11;
MqttMessage message;
/**
* 構(gòu)造函數(shù)
* @throws MqttException
*/
MqttConnect(String HOST, String clientid, String TOPIC) throws MqttException {
// MemoryPersistence設(shè)置clientid的保存形式幻馁，默認(rèn)為以內(nèi)存保存
client = new MqttClient(HOST, clientid, new MemoryPersistence());
connect(TOPIC);
}
MqttClient getClient() {
return client;
}
private void connect(String TOPIC) {
try {
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName("admin");
char mC[]={'p','a','s','s','w','o','r','d'};
options.setPassword(mC);
options.setCleanSession(false);
// 設(shè)置超時時間
options.setConnectionTimeout(10);
// 設(shè)置會話心跳時間
options.setKeepAliveInterval(20);
client.setCallback(new PushCallback(MqttConnect.this));
client.connect(options);
topic11 = client.getTopic(TOPIC);
} catch (Exception e) {
e.printStackTrace();
}
}
/**
*
* @param topic
* @param message
* @throws MqttPersistenceException
* @throws MqttException
*/
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());
}}
接收：
public class ClientMQTT {
public static final String HOST = "tcp://ip:61613";
public static final String TOPIC = "*";//主題
private static final String clientid = "*";
private MqttClient client;
private MqttConnectOptions options;
private String userName = "admin";
private String passWord = "password";
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(true);
// 設(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(TOPIC);
//setWill方法吏祸，如果項目中需要知道客戶端是否掉線可以調(diào)用該方法。設(shè)置最終端口的通知消息
options.setWill(topic, "close".getBytes(), 2, true);
client.connect(options);
//訂閱消息
int[] Qos = {1};
String[] topic1 = {TOPIC};
client.subscribe(topic1, Qos);
} catch (Exception e) {
e.printStackTrace();
}
}
public static void main(String[] args) throws MqttException {
ClientMQTT client = new ClientMQTT();
client.start();
}}
回調(diào)：
public class PushCallback implements MqttCallback {
private MqttConnect mqttConnect;
PushCallback(MqttConnect mqttConnect) {
this.mqttConnect = mqttConnect;
}
PushCallback() {
}
// MiddlewareMqttClient service;
@Override
public void connectionLost(Throwable cause) {
// 連接丟失后钩蚊，一般在這里面進行重連
System.out.println("連接斷開贡翘，可以做重連");
// for (int i = 0; i < 10; i++) {
try {
if (mqttConnect.getClient().isConnected()) {
return;
}
mqttConnect.getClient().reconnect();
} catch (MqttException e) {
e.printStackTrace();
}
// }
throw new RuntimeException("無法連接服務(wù)器");
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
System.out.println("deliveryComplete---------" + token.isComplete());
}
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
// subscribe后得到的消息會執(zhí)行到這里面
System.out.println("接收消息主題 : " + topic);
System.out.println("接收消息Qos : " + message.getQos());
System.out.println("接收消息內(nèi)容 : " + new String(message.getPayload()));
}}
image.png
