MQTT協議的簡單介紹和服務器的安裝

轉：http://blog.csdn.net/djun100/article/details/25752491

　　最近公司作的項目中有用到消息推送，通過多方面的篩選以後肯定了使用MQTT協議，相對於XMPP，MQTT更加輕量級，而且佔用用戶不多的帶寬。

MQTT是IBM推出的一種針對移動終端設備的基於TCP/IP的發佈/預訂協議，能夠鏈接大量的遠程傳感器和控制設備。

　　MQTT的官網見：http://mqtt.org/。其中http://mqtt.org/software裏面提供了官方推薦的各類服務器和客戶端使用的各類語言版本的API。

　　下面以服務器Apollo 1.6爲例，以前嘗試過使用ActiveMQ，效果很不理想，只能實現服務器和客戶端一對一的通訊，從官網上了解到Apollo屬於activemq的一個子工程。先無論這些了，言歸正傳，如下在windows環境下。

　　一、在這裏下載Apollo服務器，下載後解壓，而後運行apache-apollo-1.6\bin\apollo.cmd，輸入create mybroker（名字任意取，這裏是根據官網介紹的來取的）建立服務器實例，服務器實例包含了全部的配置，運行時數據等，而且和一個服務器進程關聯。

　　二、create mybroker以後會在bin目錄下生成mybroker文件夾，裏面包含有不少信息，其中etc\apollo.xml文件下是配置服務器信息的文件，etc\users.properties文件包含鏈接MQTT服務器時用到的用戶名和密碼，後面會介紹，能夠修改原始的admin=password，能夠接着換行添加新的用戶名密碼。

　　三、打開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兩個工程：

　　一、新建android工程MQTTClient

　　二、MainActivity代碼以下：

[java] view plaincopyprint?

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方法，若是項目中須要知道客戶端是否掉線能夠調用該方法。

　　三、新建j2se工程MQTTServer

　　四、Server代碼以下：

[java] view plaincopyprint?

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/

　　轉自：http://www.longdw.com/mqtt-server-client-android/