MQTT 是一種輕量級發布/訂閱消息的協議，通常用于具有小型的物聯網設備。消息中通常不會包含太多數據，只是傳感器值。
但是大多數情況下，MQTT 消息負載是文本，可能是少量文本或 JSON 數據負載。不過，設備如何在 MQTT 消息中發送文件，例如Image圖片.jpg格式文件呢？
這期我們通過整理網上的資料，把具體的方式分享給大家！
使用 MQTT 協議發布圖像
使用 MQTT 協議發布圖像是一種非常直接的方法。下面的圖像解釋了我們在本期中將要使用的流程。
Step1：首先我們需要在 Python 中讀取我們的圖像文件或其他類型的文件。
with open("./test.jpg",'rb') as file: filecontent = file.read()
Step2：讀取文件后，我們將圖片轉換為字節數組。然后我們將字節數組發布到我們想要發送圖片的主題。 byteArr = bytearray(filecontent)
Step3：在這種情況下，我們使用以下代碼將圖片發送到名為 photos 的主題。 result = client.publish(topic,byteArr,2)msg_status = result[0]if msg_status == 0: print(f"message : Message sent to topic {topic}")else: print(f"Failed to send message to topic {topic}")
字節數組將發布到主題。以下是一段完整的代碼，它將幫助您使用 mqtt 協議在 python 編程語言中發布任何圖片：import timefrom paho.mqtt import client as mqtt_clientbroker = 'broker.hivemq.com'port = 1883topic = "photos"client_id = 'your client id'def connect_mqtt(): def on_connect(client, userdata, flags, rc): if rc == 0: print("Successfully connected to MQTT broker") else: print("Failed to connect, return code %d", rc) client = mqtt_client.Client(client_id) client.on_connect = on_connect client.connect(broker, port) return clientdef publish(client): with open("./test.jpg",'rb') as file: filecontent = file.read() byteArr = bytearray(filecontent) print(byteArr) result = client.publish(topic,byteArr,2) msg_status = result[0] if msg_status == 0: print(f"message sent to topic {topic}") else: print(f"Failed to send message to topic {topic}")def main(): client = connect_mqtt() client.loop_start() publish(client) time.sleep(5) client.loop_stop()if __name__ == '__main__': main()
上述代碼使用了 paho-mqtt 庫，所以在終端中輸入以下命令確保已經安裝了該庫。
對于本期教程，我使用的是免費的 hivemq 公共 MQTT 代理。代碼中的 broker 地址、端口憑證是 hivemq 公共 mqtt 代理的憑證。主題“photos”是我們將發布圖像或文本文件的主題，客戶端 ID 必須是唯一的。
當然也可以使用其他的代理，例如EMQX：
broker = "broker.emqx.io"port = 1883timelive=60image_name="capture.jpg"
Top 5 免費開源MQTT Brokers代理！！！
另一種方式不同將圖像編碼為字節數組，而是將圖像編碼為 base64。要將圖像轉換為 base64，可以使用以下發布函數。
import base64def publish(client): with open("./test.jpg",'rb') as file: filecontent = file.read() base64_bytes = base64.b64encode(filecontent) base64_message = base64_bytes.decode('ascii') result = client.publish(topic,base64_message,0) msg_status = result[0] if msg_status == 0: print(f"message sent to topic {topic}") else: print(f"Failed to send message to topic {topic}")
這兩種方式都是可行的。
使用 MQTT 協議接收圖像
接收圖像的概念是相同的，只是順序相反。我們需要訂閱“photos”主題以接收字節數組或 base64 消息。當圖像或任何文件發布到該主題時，我們將接收到消息。首先，我們需要創建一個 jpg 文件。f = open('receive.jpg', 'wb')
然后，我們將接收到的字節數組或 base64 消息寫入文件。f.write(msg.payload)f.close()
如果你使用了 base64，那么可以使用以下代碼將其轉換回圖像。但在執行此操作之前，請確保已經導入了 base64 庫。
f = open('receive.jpg', 'wb')msg = str(message.payload.decode('utf-8'))img = msg.encode('ascii')final_img = base64.b64decode(img)f.write(final_img)f.close()
完整的代碼如下所示：
from paho.mqtt import client as mqtt_clientbroker = 'broker.hivemq.com'port = 1883topic = "photos"topic_sub = "photos"client_id = 'xzcfghjt123'def connect_mqtt(): def on_connect(client, userdata, flags, rc): if rc == 0: print("Successfully connected to MQTT broker") else: print("Failed to connect, return code %d", rc) client = mqtt_client.Client(client_id) client.on_connect = on_connect client.connect(broker, port) return clientdef subscribe(client: mqtt_client): def on_message(client, userdata, msg): f = open('receive.jpg', 'wb') f.write(msg.payload) f.close() print ('image received') client.subscribe(topic_sub) client.on_message = on_messagedef main(): client = connect_mqtt() subscribe(client) client.loop_forever()
if __name__ == '__main__': main()
代碼與發布代碼非常相似，但我們現在使用的是 subscribe 函數而不是 publish 函數。首先，我們連接到 MQTT 代理，連接后我們訂閱了“photos”主題。在 subscribe 函數中，我們還定義了 client.on_Message 函數，這意味著每當接收到消息時，該函數將被調用。在這個代碼中，我們使用了 loop_forever 函數，因為我們想一直監聽數據。
參考鏈接：
https://boardor.com/blog/detect-objects-and-transmit-images-using-mqtt
https://davidmac.pro/posts/2021-07-21-files-over-mqtt/
https://highvoltages.co/iot-internet-of-things/mqtt/image-using-mqtt-protocol/
閱讀原文：原文鏈接
該文章在 2025/6/23 12:59:24 編輯過
400 186 1886
