Python實現MQTT接收訂閱數據

1、背景

目前MQTT的標準組織官網：http://www.mqtt.org，裏面列出了不少支持的軟件相關資源。

一個輕量級的MQTT服務器是：http://www.mosquitto.org，能夠運行ARM／MIPS的嵌入式linux系統上。

物聯網常使用 「消息隊列遙測傳輸(Message Queuing Telemetry Transport, MQTT)」 協議訂閱數據，這裏用Python實現從MQTT服務器訂閱數據。

首先和TCP協議比較

首先TCP是傳輸層協議，實現了一個雙向的通訊鏈路。

MQTT是基於TCP的應用層協議。（固然中間可能多一層，websocket）

二者不在一個層級，比較誰比誰好是沒有意義的。

咱們用raw TCP 也能夠實現數據通訊，好比發送傳感器數據到服務器。爲何要用到MQTT呢 ？

假設如今有一個物聯網的應用，題主固然能夠直接用TCP socket 作通訊，實際上很多人也是這麼作的。而後你就會發現：

• 須要本身寫確認重傳的機制，由於TCP 鏈接說不定就斷了。

• 若是有不少個傳感器（生產者），又要寫代碼管理這麼多TCP鏈接呢。

• 若是同時又有多個地方須要用到這些數據，還得寫一個轉發的邏輯。

• 若是系統很複雜，參與人或公司不少，那通訊格式要怎麼定，怎麼改，溝通成本就很大了。

這些東西這麼麻煩，又不想加班寫代碼，那有沒有辦法簡便地解決呢？固然有，就是用現成的協議啦，好比MQTT。

MQTT 提供兩個核心功能：

• 三個級別的QOS

• 基於訂閱/發佈的消息轉發服務。

用了MQTT, 上面提到的這些問題就都被優雅地解決掉啦。

其實，同類應用的不少問題，都是有必定共性的。這時候就會有一些人提出通常性的解決方式，這樣你們就不用重複造輪子，同時又保證了互操做性。這就是協議存在的意義啦。

因此一句話總結，MQTT 和其餘的應用層協議，好比 HTPP, FTP, BitTorrent 協議同樣，都是爲了解決特定問題而生的一套方案，能夠幫咱們省好多事。

2、安裝部署

環境：Python 2.7.5

          mosquitto version 1.5.8

          mosquitto is an MQTT v3.1.1 broker.

安裝安裝mosquitto，這裏比較省事了

yum -y install mosquitto mosquitto-clients python-mosquitto   

安裝paho-mqtt，pypi上有這個庫，能夠自行安裝

3、測試

啓動命令: mosquitto
server：mosquitto_pub -t test -h 127.0.0.1 -m  '{"pin":1,"value":0}'
client：mosquitto_sub -v -t test -h 127.0.0.1 （先啓動）

　測試結果

[root@test ~]# mosquitto_sub -v -t test -h 127.0.0.1        
test {"pin":1,"value":0}　

使用腳本測試

client.py

#!/usr/bin/python

import sys
import datetime
import socket, sys

#======================================================        
try:
    import paho.mqtt.client as mqtt
except ImportError:
    print("MQTT client not find. Please install as follow:")
    print("git clone http://git.eclipse.org/gitroot/paho/org.eclipse.paho.mqtt.python.git")
    print("cd org.eclipse.paho.mqtt.python")
    print("sudo python setup.py install")

# 服務器地址
strBroker = "localhost"
# 通訊端口
port = 1883
# 用戶名
username = 'username'
# 密碼
password = 'password'
# 訂閱主題名
topic = 'topic'

#======================================================
def on_connect(mqttc, obj, rc):
    print("OnConnetc, rc: "+str(rc))

def on_publish(mqttc, obj, mid):
    print("OnPublish, mid: "+str(mid))

def on_subscribe(mqttc, obj, mid, granted_qos):
    print("Subscribed: "+str(mid)+" "+str(granted_qos))

def on_log(mqttc, obj, level, string):
    print("Log:"+string)

def on_message(mqttc, obj, msg):
    curtime = datetime.datetime.now()
    strcurtime = curtime.strftime("%Y-%m-%d %H:%M:%S")
    print(strcurtime + ": " + msg.topic+" "+str(msg.qos)+" "+str(msg.payload))
    on_exec(str(msg.payload))

def on_exec(strcmd):
    print "Exec:",strcmd

#=====================================================
if __name__ == '__main__':
    mqttc = mqtt.Client("test")
    mqttc.on_message = on_message
    mqttc.on_connect = on_connect
    mqttc.on_publish = on_publish
    mqttc.on_subscribe = on_subscribe
    mqttc.on_log = on_log

    # 設置帳號密碼（若是須要的話）
    #mqttc.username_pw_set(username, password=password)

    mqttc.connect(strBroker, port, 60)
    mqttc.subscribe(topic, 0)
    mqttc.loop_forever()

server.py

#!/usr/bin/python

import sys
import datetime
import socket, sys
import paho.mqtt.publish as publish

def transmitMQTT(strMsg):
    strMqttBroker = "localhost"
    strMqttChannel = "test"
    print(strMsg)
    publish.single(strMqttChannel, strMsg, hostname = strMqttBroker)

if __name__ == '__main__':
    transmitMQTT("Hello,MQTT")
    print "Send msg ok."

4、開發中須要注意的一些問題

• MQTT鏈接心跳時間

• MQTT單個發佈消息最大長度

• 離線消息最長緩存時間

• 單MQTT鏈接的最大訂閱數