需要安裝的python庫

 使用python編寫程式進行測試MQTT的釋出和訂閱功能。首先要安裝:pip install paho-mqtt

測試釋出（pub）

 我的MQTT部署在阿里雲的伺服器上面，所以我在本機上編寫了python程式進行測試。

然後在shell裡面重新開啟一個終端，訂閱一個主題為“chat” mosquitto_sub -t chat

 在本機上測試遠端的MQTT的釋出功能就是把自己作為一個傳送資訊的人，當自己傳送資訊的時候，所有訂閱過該主題（topic）的物件都將收到自己傳送的資訊。 
mqtt_client.py

# encoding: utf-8

import paho.mqtt.client as
 
 mqtt

HOST = "101.200.46.138"
PORT = 1883

def test():
    client = mqtt.Client()
    client.connect(HOST, PORT, 60)
    client.publish("chat","hello liefyuan",2) # 釋出一個主題為'chat',內容為‘hello liefyuan’的資訊
    client.loop_forever()

if __name__ == '__main__':
    test()

釋出/訂閱測試

# -*- coding: utf-8 -*-

import
 
 paho.mqtt.client as mqtt

MQTTHOST = "101.200.46.138"
MQTTPORT = 1883
mqttClient = mqtt.Client()


# 連線MQTT伺服器
def on_mqtt_connect():
    mqttClient.connect(MQTTHOST, MQTTPORT, 60)
    mqttClient.loop_start()


# publish 訊息
def on_publish(topic, payload, qos):
    mqttClient.publish(topic, payload, qos)

# 訊息處理函式
 

def on_message_come(lient, userdata, msg):

    print(msg.topic + " " + ":" + str(msg.payload))


# subscribe 訊息
def on_subscribe():
    mqttClient.subscribe("/server", 1)
    mqttClient.on_message = on_message_come # 訊息到來處理函式


def main():
    on_mqtt_connect()
    on_publish("/test/server", "Hello Python!", 1)
    on_subscribe()
    while True:
        pass



if __name__ == '__main__':
    main()

註解函式：

client.connect(self, host, port, keepalive, bind_address)

client.publish(self, topic, payload, qos, retain)

client.subscribe(self, topic, qos)

測試訂閱（sub）

 在本機上編寫程式測試訂閱功能，就是讓自己的程式作為一個接收者，同一個主題沒有釋出（pub）資訊的時候，就自己一直等候。

# encoding: utf-8


import paho.mqtt.client as mqtt


def on_connect(client, userdata, flags, rc):
    print("Connected with result code "+str(rc))
    client.subscribe("chat")


def on_message(client, userdata, msg):
    print(msg.topic+" " + ":" + str(msg.payload))

client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect("www.liefyuan.top", 1883, 60)
client.loop_forever()