Mqtt搭建代理服務器進行通訊-淺析

時間  2019-11-17
標籤 mqtt 搭建 代理 服務器 進行 通訊 淺析 简体版
原文   原文鏈接

本文基於Windows系統操做

MQTT簡介:

MQTT(Message Queuing Telemetry Transport，消息隊列遙測傳輸)是IBM開發的一個即時通信協議！
複製代碼

MQTT消息的主要特色:

使用(publish/subscribe)消息模式,簡稱p/s模式,即發佈/訂閱！提供一對多的發送方式！
複製代碼
MQTT根據QoS定義的等級來傳輸消息：
  • level 0：最多一次的傳輸
消息是基於TCP/IP網絡傳輸的。沒有迴應，在協議中也沒有定義重傳的語義。消息可能到達服務器1次，也可能根本不會到達。
複製代碼
  • level 1：至少一次的傳輸
服務器接收到消息會被確認，經過傳輸一個PUBACK信息。若是有一個能夠辨認的傳輸失敗，不管是通信鏈接仍是發送設備，仍是過了一段時間確認信息沒有收到，發送方都會將消息頭的DUP位置1，而後再次發送消息。消息最少一次到達服務器。SUBSCRIBE和UNSUBSCRIBE都使用level 1 的QoS。
若是客戶端沒有接收到PUBACK信息（不管是應用定義的超時，仍是檢測到失敗而後通信session重啓），客戶端都會再次發送PUBLISH信息，而且將DUP位置1。
當它從客戶端接收到重複的數據，服務器從新發送消息給訂閱者，而且發送另外一個PUBACK消息。
複製代碼
  • level 2： 只有一次的傳輸
在QoS level 1上附加的協議流保證了重複的消息不會傳送到接收的應用。這是最高級別的傳輸，當重複的消息不被容許的狀況下使用。這樣增長了網絡流量，可是它一般是能夠接受的，由於消息內容很重要。
QoS level 2在消息頭有Message ID。
複製代碼

接下來開始咱們的表演：

下載代理服務器

  • 本文使用mqtt代理服務器是apache下的apollo代理服務器java

  • 下載地址：http://www.apache.org/dyn/closer.cgi?path=activemq/activemq-apollo/1.7.1/apache-apollo-1.7.1-unix-distro.tar.gzandroid

  • Paste_Image.png

建立代理服務器

  • 下載完成而後解壓目錄apache

  • 打開dos窗口進入到apache-apollo-1.7.1\bin目錄下瀏覽器

  • 執行apollo create testbroker命令建立一個名稱爲testbroker的代理服務器服務器

  • Paste_Image.png

  • 下面就是咱們建立的代理服務器 網絡

    Paste_Image.png

啓動代理服務器

  • 使用dos進入testbroker目錄中的bin目錄下session

  • 執行apollo-broker run命令啓動代理服務器app

  • Paste_Image.png

經過HTTP訪問代理服務器

如今咱們能夠打開瀏覽器看下咱們的代理服務器 輸入網址http://127.0.0.1:61680/eclipse

  • Paste_Image.png

  • 用戶名密碼可到配置文件中查看tcp

  • 進入testbroker目錄下的etc目錄

Paste_Image.png

  • users.properties中配置的用戶名和密碼

Paste_Image.png

  • 默認有個用戶名爲admin，密碼爲password的用戶

  • 咱們也能夠本身配置用戶

  • 如今就用默認用戶登錄

  • Paste_Image.png

OK登錄成功

接下來咱們編寫Android客戶端

  • 首先準備mqtt jar包
  • 不容易呀,mqtt這個jar真難找
  • https://repo.eclipse.org/content/repositories/paho/org/eclipse/paho/org.eclipse.paho.client.mqttv3/1.0.2/

Paste_Image.png

必定保證客戶端和服務端以及代理服務器所在的電腦在同一網段下
  • 能夠在電腦上生成wifi熱點,手機客戶端鏈接熱點便可
  • 接下來直接貼Android客戶端代碼
MqttService.java
package com.example.jingwc.mqtt_demo;

import android.app.Service;
import android.content.Intent;
import android.os.Binder;
import android.os.IBinder;
import android.util.Log;

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.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class MqttService extends Service {

    /** * 代理服務器ip地址 */
    public static final String MQTT_BROKER_HOST = "tcp://192.168.1.107:61613";

    /** * 客戶端惟一標識 */
    public static final String MQTT_CLIENT_ID = "android-jingwc";

    /** * 訂閱標識 */
    public static final String MQTT_TOPIC = "jingwc";

    /** * 用戶名 */
    public static final String USERNAME = "admin";
    /** * 密碼 */
    public static final String PASSWORD = "password";

    private MqttClient mqttClient;

    public MqttService() {
    }

    @Override
    public IBinder onBind(Intent intent) {
        return binder;
    }

    /** * 鏈接mqtt */
    public void connect(){
        try {
            // host爲主機名，clientid即鏈接MQTT的客戶端ID，通常以客戶端惟一標識符表示，
            // MemoryPersistence設置clientid的保存形式，默認爲之內存保存
            mqttClient = new MqttClient(MQTT_BROKER_HOST,MQTT_CLIENT_ID,new MemoryPersistence());
            // 配置參數信息
            MqttConnectOptions 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);
            // 鏈接
            mqttClient.connect(options);

            // 訂閱
            mqttClient.subscribe(MQTT_TOPIC);

            // 設置回調
            mqttClient.setCallback(new MqttCallback() {
                //鏈接丟失後，通常在這裏面進行重連
                @Override
                public void connectionLost(Throwable throwable) {
                    Log.d("test","connectionLost");
                }


                //subscribe後獲得的消息會執行到這裏面
                @Override
                public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                    Log.d("test","messageArrived"+mqttMessage.toString());
                }

                //publish後會執行到這裏
                @Override
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                    Log.d("test","deliveryComplete");
                }
            });

        } catch (MqttException e) {
            e.printStackTrace();
        }catch (Exception e) {
            e.printStackTrace();
        }
    }

    /** * 斷開鏈接 */
    public void disconnect(){
        if(mqttClient != null){
            if(mqttClient.isConnected()){
                try {
                    mqttClient.disconnect();
                    mqttClient = null;
                } catch (MqttException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    private final Binder binder = new MyBinder();

    class MyBinder extends Binder{

        public MqttService getService(){
            return MqttService.this;
        }
    }
}

複製代碼
MainActivity.java
package com.example.jingwc.mqtt_demo;

import android.content.ComponentName;
import android.content.Intent;
import android.content.ServiceConnection;
import android.os.IBinder;
import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.view.View;
import android.widget.Button;
import android.widget.EditText;

public class MainActivity extends AppCompatActivity {

    MqttService service = null;

    private ServiceConnection mConnection = new ServiceConnection() {
        @Override
        public void onServiceConnected(ComponentName componentName, IBinder iBinder) {
            service = ((MqttService.MyBinder)iBinder).getService();
        }

        @Override
        public void onServiceDisconnected(ComponentName componentName) {
            service = null;
        }
    };

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        Button bt_connect = (Button) findViewById(R.id.bt_connect);
        Button bt_disconnect = (Button) findViewById(R.id.bt_disconnect);

        bt_connect.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                // 鏈接
                service.connect();
            }
        });

        bt_disconnect.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                // 斷開鏈接
                service.disconnect();
            }
        });

        bindService(new Intent(this,MqttService.class),mConnection,BIND_AUTO_CREATE);
    }
}
複製代碼

服務端代碼

  • 也能夠在寫一個android程序看成服務端
  • 我這裏寫的是java項目
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
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.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class MqttServer {
	
	// 代理服務器ip地址
	private static String host = "tcp://192.168.1.107:61613";
	
	private static String userName = "admin";
	private static String password = "password";
	
	private static MqttClient client;
	
	// 主題
	private static MqttTopic topic;
	
	private static MqttMessage message;
	
	// 訂閱標識
	private static String topicStr = "jingwc";
	
	public static void main(String[] args) throws MqttException{
		client = new MqttClient(host,"java-server-jingwc",new MemoryPersistence());
		MqttConnectOptions options = new MqttConnectOptions();
		options.setCleanSession(true);
		options.setUserName(userName);
		options.setPassword(password.toCharArray());
		options.setConnectionTimeout(10);
		options.setKeepAliveInterval(20);
		
		topic = client.getTopic(topicStr);
		
		message = new MqttMessage();
		message.setQos(1);
		message.setRetained(true);
		message.setPayload("from server message".getBytes());
		client.connect(options);
		MqttDeliveryToken token = topic.publish(message);
		token.waitForCompletion();
		System.out.println("token:"+token.isComplete());
		
	}

}
複製代碼

服務端經過代理服務器發送客戶端訂閱的消息圖 ( 我的理解 )

  • Paste_Image.png

Paste_Image.png
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程