MQTT協議的簡單介紹和伺服器的安裝
阿新 • • 發佈：2019-02-16
最近公司做的專案中有用到訊息推送，經過多方面的篩選之後確定了使用MQTT協議，相對於XMPP，MQTT更加輕量級，並且佔用使用者很少的頻寬。
MQTT是IBM推出的一種針對移動終端裝置的基於TCP/IP的釋出/預訂協議，可以連線大量的遠端感測器和控制裝置。
下面以伺服器Apollo 1.6為例，之前嘗試過使用ActiveMQ，效果很不理想，只能實現伺服器和客戶端一對一的通訊，從官網上了解到Apollo屬於activemq的一個子工程。先不管這些了，言歸正傳，以下在windows環境下。
1、在這裡下載Apollo伺服器，下載後解壓，然後執行apache-apollo-1.6\bin\apollo.cmd，輸入create mybroker（名字任意取，這裡是根據 官網介紹的來取的）建立伺服器例項，伺服器例項包含了所有的配置，執行時資料等，並且和一個伺服器程序關聯。
2、create mybroker之後會在bin目錄下生成mybroker資料夾，裡面包含有很多資訊，其中etc\apollo.xml檔案下是配置伺服器資訊的檔案，etc\users.properties檔案包含連線MQTT伺服器時用到的使用者名稱和密碼，後面會介紹，可以修改原始的admin=password，可以接著換行新增新的使用者名稱密碼。
3、開啟cmd，執行…apache-apollo-1.6\bin\mybroker\bin\apollo-broker.cmd run 開啟伺服器，可以在瀏覽器中輸入 http://127.0.0.1:61680/檢視是否安裝成功，該介面展示了topic，連線數等很多資訊。
經過上面的簡單步驟，伺服器基本上就已經完成，下一篇將介紹Android客戶端的編寫和注意事項。
客戶端使用的API，開始我使用的是mqtt-client，使用過後發現問題百出，不能很好的滿足要求，後來使用了官方推薦的Eclipse Paho，下面開始客戶端程式碼的編寫，為了方便測試這裡有android和j2se兩個工程：
1、新建android工程MQTTClient
2、MainActivity程式碼如下：
package ldw.mqttclient; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; 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.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; import android.app.Activity; import android.os.Bundle; import android.os.Handler; import android.os.Message; import android.view.KeyEvent; import android.widget.TextView; import android.widget.Toast; public class MainActivity extends Activity { private TextView resultTv; private String host = "tcp://127.0.0.1:1883"; private String userName = "admin"; private String passWord = "password"; private Handler handler; private MqttClient client; private String myTopic = "test/topic"; private MqttConnectOptions options; private ScheduledExecutorService scheduler; @Override protected void onCreate(Bundle savedInstanceState) { super.onCreate(savedInstanceState); setContentView(R.layout.main); resultTv = (TextView) findViewById(R.id.result); init(); handler = new Handler() { @Override public void handleMessage(Message msg) { super.handleMessage(msg); if(msg.what == 1) { Toast.makeText(MainActivity.this, (String) msg.obj, Toast.LENGTH_SHORT).show(); System.out.println("-----------------------------"); } else if(msg.what == 2) { Toast.makeText(MainActivity.this, "連線成功", Toast.LENGTH_SHORT).show(); try { client.subscribe(myTopic, 1); } catch (Exception e) { e.printStackTrace(); } } else if(msg.what == 3) { Toast.makeText(MainActivity.this, "連線失敗，系統正在重連", Toast.LENGTH_SHORT).show(); } } }; startReconnect(); } private void startReconnect() { scheduler = Executors.newSingleThreadScheduledExecutor(); scheduler.scheduleAtFixedRate(new Runnable() { @Override public void run() { if(!client.isConnected()) { connect(); } } }, 0 * 1000, 10 * 1000, TimeUnit.MILLISECONDS); } private void init() { try { //host為主機名，test為clientid即連線MQTT的客戶端ID，一般以客戶端唯一識別符號表示，MemoryPersistence設定clientid的儲存形式，預設為以記憶體儲存 client = new MqttClient(host, "test", new MemoryPersistence()); //MQTT的連線設定 options = new MqttConnectOptions(); //設定是否清空session,這裡如果設定為false表示伺服器會保留客戶端的連線記錄，這裡設定為true表示每次連線到伺服器都以新的身份連線 options.setCleanSession(true); //設定連線的使用者名稱 options.setUserName(userName); //設定連線的密碼 options.setPassword(passWord.toCharArray()); // 設定超時時間 單位為秒 options.setConnectionTimeout(10); // 設定會話心跳時間 單位為秒 伺服器會每隔1.5*20秒的時間向客戶端傳送個訊息判斷客戶端是否線上，但這個方法並沒有重連的機制 options.setKeepAliveInterval(20); //設定回撥 client.setCallback(new MqttCallback() { @Override public void connectionLost(Throwable cause) { //連線丟失後，一般在這裡面進行重連 System.out.println("connectionLost----------"); } @Override public void deliveryComplete(IMqttDeliveryToken token) { //publish後會執行到這裡 System.out.println("deliveryComplete---------" + token.isComplete()); } @Override public void messageArrived(String topicName, MqttMessage message) throws Exception { //subscribe後得到的訊息會執行到這裡面 System.out.println("messageArrived----------"); Message msg = new Message(); msg.what = 1; msg.obj = topicName+"---"+message.toString(); handler.sendMessage(msg); } }); // connect(); } catch (Exception e) { e.printStackTrace(); } } private void connect() { new Thread(new Runnable() { @Override public void run() { try { client.connect(options); Message msg = new Message(); msg.what = 2; handler.sendMessage(msg); } catch (Exception e) { e.printStackTrace(); Message msg = new Message(); msg.what = 3; handler.sendMessage(msg); } } }).start(); } @Override public boolean onKeyDown(int keyCode, KeyEvent event) { if(client != null && keyCode == KeyEvent.KEYCODE_BACK) { try { client.disconnect(); } catch (Exception e) { e.printStackTrace(); } } return super.onKeyDown(keyCode, event); } @Override protected void onDestroy() { super.onDestroy(); try { scheduler.shutdown(); client.disconnect(); } catch (MqttException e) { e.printStackTrace(); } } }
由於專案需要，我用到了心跳重連。根據這裡的解釋設定apollo.xml，主要有設定主機連線的地址。另外，options還有個setWill方法，如果專案中需要知道客戶端是否掉線可以呼叫該方法。
3、新建j2se工程MQTTServer
4、Server程式碼如下：
import java.awt.Container;
import java.awt.event.ActionEvent;
import java.awt.event.ActionListener;
import javax.swing.JButton;
import javax.swing.JFrame;
import javax.swing.JPanel;
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.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
public class Server extends JFrame {
private static final long serialVersionUID = 1L;
private JPanel panel;
private JButton button;
private MqttClient client;
private String host = "tcp://127.0.0.1:1883";
// private String host = "tcp://localhost:1883";
private String userName = "test";
private String passWord = "test";
private MqttTopic topic;
private MqttMessage message;
private String myTopic = "test/topic";
public Server() {
try {
client = new MqttClient(host, "Server",
new MemoryPersistence());
connect();
} catch (Exception e) {
e.printStackTrace();
}
Container container = this.getContentPane();
panel = new JPanel();
button = new JButton("釋出話題");
button.addActionListener(new ActionListener() {
@Override
public void actionPerformed(ActionEvent ae) {
try {
MqttDeliveryToken token = topic.publish(message);
token.waitForCompletion();
System.out.println(token.isComplete()+"========");
} catch (Exception e) {
e.printStackTrace();
}
}
});
panel.add(button);
container.add(panel, "North");
}
private void connect() {
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(false);
options.setUserName(userName);
options.setPassword(passWord.toCharArray());
// 設定超時時間
options.setConnectionTimeout(10);
// 設定會話心跳時間
options.setKeepAliveInterval(20);
try {
client.setCallback(new MqttCallback() {
@Override
public void connectionLost(Throwable cause) {
System.out.println("connectionLost-----------");
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
System.out.println("deliveryComplete---------"+token.isComplete());
}
@Override
public void messageArrived(String topic, MqttMessage arg1)
throws Exception {
System.out.println("messageArrived----------");
}
});
topic = client.getTopic(myTopic);
message = new MqttMessage();
message.setQos(1);
message.setRetained(true);
System.out.println(message.isRetained()+"------ratained狀態");
message.setPayload("eeeeeaaaaaawwwwww---".getBytes());
client.connect(options);
} catch (Exception e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
Server s = new Server();
s.setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);
s.setSize(600, 370);
s.setLocationRelativeTo(null);
s.setVisible(true);
}
}上面程式碼跟客戶端的程式碼差不多，這裡就不做解釋了。
沒什麼好說的，MQTT就是這麼簡單，但開始在使用的時候要注意一些引數的設定來適應專案的需求。
jar包下載地址：
https://repo.eclipse.org/content/repositories/paho/org/eclipse/paho/mqtt-client/0.4.0/