Mosquitto: Python 和 MQTT
本文件是入門指南指南的一部分。請在此處查看完整指南：如何開始使用 Mosquitto。
👋 歡迎來到 Stackhero 文件中心！
Stackhero 為您提供一個全代管的 Mosquitto MQTT cloud 環境，專為高可靠性與彈性設計：
- 無限制的訊息吞吐量與資料傳輸，確保您的工作流程不會受到人為限制。
- 透過您自有的外部 API，實現無限制的裝置驗證，讓裝置註冊與存取管理變得輕鬆無縫。
- 進階 ACLs，可對主題（topics）、使用者及操作進行細緻權限控管。
- 提供自訂網域名稱並內建HTTPS，打造安全且具品牌識別的端點（例如：https://mqtt.your-company.com）。
- 零負擔更新：只需一鍵即可套用功能改進或安全修補。
- 穩定的效能與強化的安全性，每個實例皆運行於專屬的私有基礎架構。
加速您的 IoT 專案，同時降低營運負擔。您只需幾分鐘，即可擁有一個安全、可用於生產環境的 Mosquitto MQTT cloud hosting 實例。
要使用 Python 與 Mosquitto，建議使用 Paho MQTT Python 客戶端庫。使用以下命令安裝：
pip install paho-mqtt
以下是一個使用 TLS 加密和身份驗證連接到 Mosquitto 伺服器的 Python 腳本示例：
import paho.mqtt.client as mqtt
import random
import string
from paho.mqtt.client import CallbackAPIVersion
def on_connect(client, userdata, flags, reason_code, properties=None):
if reason_code == 0:
print("Connected successfully")
else:
print(f"Connection failed with reason {reason_code}")
def on_message(client, userdata, msg, properties=None):
print(f"Received message: {msg.payload.decode()} on topic {msg.topic}")
def generate_client_id(length=8):
characters = string.ascii_letters + string.digits
return "client_" + ''.join(random.choice(characters) for _ in range(length))
client_id = generate_client_id()
client = mqtt.Client(
client_id=client_id,
callback_api_version=CallbackAPIVersion.VERSION2
)
client.on_connect = on_connect
client.on_message = on_message
host = "<XXXXXX>.stackhero-network.com"
port = <PORT_TLS>
client.username_pw_set("<USER>", "<PASSWORD>")
client.tls_set()
try:
print(f"Connecting to {host} with client ID: {client_id}")
client.connect(host, port)
client.loop_start()
client.subscribe("$SYS/#")
try:
while True:
pass
except KeyboardInterrupt:
print("\nDisconnecting...")
client.loop_stop()
client.disconnect()
except Exception as e:
print(f"Error: {e}")
有了這些指引，您應該能夠順利開始在 Stackhero 上使用 Mosquitto。祝您愉快地探索 MQTT！