Python實現MQTT接收訂閱數據

一、背景

目前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 協議一樣，都是為了解決特定問題而生的一套方案，可以幫我們省好多事。

二、安裝部署

環境：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上有這個庫，可以自行安裝

三、測試

啟動命令: 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."

四、開發中需要注意的一些問題

• MQTT連接心跳時間

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

• 離線消息最長緩存時間

• 單MQTT連接的最大訂閱數