如何在使用 Python 時將 MQTT 訊息從我的裝置發佈至 AWS IoT Core？

4 分的閱讀內容

我無法在 AWS IoT Core 與我的裝置或 MQTT 用戶端之間傳送或接收 MQTT (MQ 遙測傳輸) 訊息。如何將 MQTT 訊息發佈至 AWS IoT Core？

簡短描述

確認已正確設定 AWS IoT 實物，並正確連接其憑證。若要測試您的設定，您可以使用 AWS IoT MQTT 用戶端和本文中提供的 Python 程式碼範例。

解決方法

設定目錄以測試 MQTT 發佈

1. 在開發環境中建立工作目錄。例如：iot-test-publish。

2. 在新的工作目錄中為憑證建立子目錄。例如：certificates。

3. 從命令列中，將目錄變更為新的工作目錄。

安裝適用於 Python 的 pip 和 AWS IoT SDK

1. 如果您尚未這樣做，請安裝適用於 Python 3 封裝的 pip。如需詳細資訊，請參閱 Python Packaging Authority (PyPA) 網站上的安裝。

2. 從命令列執行下列命令，以安裝適用於 Python v2 的 AWS IoT SDK：

pip install awsiotsdk

-或-

執行下列命令，以安裝適用於 Python 的 AWS IoT Device SDK (前一個 SDK 版本)：

pip install AWSIoTPythonSDK

如需詳細資訊，請參閱 GitHub 上的適用於 Python v2 的 AWS IoT SDK 或適用於 Python 的 AWS IoT Device SDK。

**注意：**雖然建議使用這些 SDK 連接至 AWS IoT Core，但其並非必要項目。您也可以使用任何合規第三方 MQTT 用戶端進行連接。

建立 AWS IoT Core 政策

1. 開啟 AWS IoT Core 主控台。

2. 在左側導覽窗格中，選擇安全。

3. 在安全下方，選擇政策。

4. 如果您有現有 AWS IoT Core 政策，請選擇建立以建立新政策。
-或-
如果您沒有任何現有政策，請在您還沒有任何政策頁面上，選擇建立政策。

5. 在建立政策頁面上，輸入政策的名稱。例如：admin。

6. 在新增陳述式下方，執行下列操作：
對於動作，請輸入 iot:*。
重要：允許所有 AWS IoT 動作 (iot:*) 對於測試很有用。但是，最佳實務是提高生產設定的安全。如需更多安全政策範例，請參閱 AWS IoT 政策範例。
對於資源 ARN，請輸入 *。
對於效果，請選取允許核取方塊。

7. 選擇建立。

如需詳細資訊，請參閱建立 AWS IoT Core 政策和 AWS IoT Core 政策。

建立 AWS IoT 實物

**注意：**您無需建立實物即可連接至 AWS IoT。但是，實物可讓您使用其他安全控制項及其他 AWS IoT 功能，例如機群索引、作業或裝置影子。

1. 在 AWS IoT Core 主控台的左側導覽窗格中，選擇管理。

2. 如果您有現有實物，請選擇建立以建立新實物。
-或-
如果您沒有任何現有實物，請在您還沒有任何實物頁面上，選擇註冊實物。

3. 在建立 AWS IoT 實物頁面上，選擇建立單一實物。

4. 在將您的裝置新增至實物登錄檔頁面上，執行下列操作：
輸入實物的名稱。例如：Test-Thing。
(選用) 在將類型新增至此實物下方，選擇或建立實物類型。
(選用) 在將此實物新增至群組下方，選擇或建立群組。如需有關群組的詳細資訊，請參閱靜態實物群組和動態實物群組。
(選用) 在設定可搜尋的實物屬性 (選用) 下方，將屬性新增為鍵值對。
選擇下一步。

5. 在新增實物的憑證頁面上，選擇建立憑證。您會看到通知，確認已為您建立實物和實物憑證。

6. 在已建立憑證頁面上，執行下列操作：
在若要連接裝置，您需要下載下列項目下方，針對憑證、公有金鑰和私有金鑰選擇下載。
將下載的每個檔案儲存至先前建立的憑證子目錄中。
在您還需要下載 AWS IoT 的根 CA 下方，選擇下載。伺服器驗證頁面會開啟至用於伺服器驗證的 CA 憑證。

7. 在 Amazon 信任服務端點 (偏好) 下方，選擇 Amazon 根 CA 1。憑證會在瀏覽器中開啟。

8. 複製憑證 (從 -----BEGIN CERTIFICATE----- 至 -----END CERTIFICATE----- 的所有內容)，然後將其貼到文字編輯器中。

9. 將憑證以名為 root.pem 的 .pem 檔案儲存至憑證子目錄。

10. 在 AWS IoT Core 主控台的已建立憑證頁面上，選擇啟用。按鈕會變更為停用。

11. 選擇連接政策。

12. 在新增實物的政策頁面上，執行下列操作：
選取先前建立的 AWS IoT Core 政策。例如：admin。
選擇註冊實物。

如需詳細資訊，請參閱建立實物物件。

複製 AWS IoT Core 端點 URL

1. 在 AWS IoT Core 主控台的左側導覽窗格中，選擇設定。

2. 在設定頁面的自訂端點下方，複製端點。此 AWS IoT Core 自訂端點 URL 專供您的 AWS 帳戶和區域使用。

建立 Python 程式檔案

將下列其中一個 Python 程式碼範例，儲存成名為 publish.py 的 Python 程式檔案。

如果您之前安裝了適用於 Python v2 的 AWS IoT SDK，請使用下列程式碼範例：

**重要：**請將 customEndpointUrl 取代為您的 AWS IoT Core 自訂端點 URL。將 certificates 取代為您的憑證子目錄名稱。將 a1b23cd45e-certificate.pem.crt 取代為您的用戶端 .crt 名稱。將 a1b23cd45e-private.pem.key 取代為您的私有金鑰名稱。

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: MIT-0

from awscrt import io, mqtt, auth, http
from awsiot import mqtt_connection_builder
import time as t
import json

# Define ENDPOINT, CLIENT_ID, PATH_TO_CERTIFICATE, PATH_TO_PRIVATE_KEY, PATH_TO_AMAZON_ROOT_CA_1, MESSAGE, TOPIC, and RANGE
ENDPOINT = "customEndpointUrl"
CLIENT_ID = "testDevice"
PATH_TO_CERTIFICATE = "certificates/a1b23cd45e-certificate.pem.crt"
PATH_TO_PRIVATE_KEY = "certificates/a1b23cd45e-private.pem.key"
PATH_TO_AMAZON_ROOT_CA_1 = "certificates/root.pem"
MESSAGE = "Hello World"
TOPIC = "test/testing"
RANGE = 20

# Spin up resources
event_loop_group = io.EventLoopGroup(1)
host_resolver = io.DefaultHostResolver(event_loop_group)
client_bootstrap = io.ClientBootstrap(event_loop_group, host_resolver)
mqtt_connection = mqtt_connection_builder.mtls_from_path(
            endpoint=ENDPOINT,
            cert_filepath=PATH_TO_CERTIFICATE,
            pri_key_filepath=PATH_TO_PRIVATE_KEY,
            client_bootstrap=client_bootstrap,
            ca_filepath=PATH_TO_AMAZON_ROOT_CA_1,
            client_id=CLIENT_ID,
            clean_session=False,
            keep_alive_secs=6
            )
print("Connecting to {} with client ID '{}'...".format(
        ENDPOINT, CLIENT_ID))
# Make the connect() call
connect_future = mqtt_connection.connect()
# Future.result() waits until a result is available
connect_future.result()
print("Connected!")
# Publish message to server desired number of times.
print('Begin Publish')
for i in range (RANGE):
    data = "{} [{}]".format(MESSAGE, i+1)
    message = {"message" : data}
    mqtt_connection.publish(topic=TOPIC, payload=json.dumps(message), qos=mqtt.QoS.AT_LEAST_ONCE)
    print("Published: '" + json.dumps(message) + "' to the topic: " + "'test/testing'")
    t.sleep(0.1)
print('Publish End')
disconnect_future = mqtt_connection.disconnect()
disconnect_future.result()

-或-

如果您安裝了適用於 Python 的 AWS IoT Device SDK (先前的 SDK 版本)，請使用下列範例程式碼：

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: MIT-0

import time as t
import json
import AWSIoTPythonSDK.MQTTLib as AWSIoTPyMQTT

# Define ENDPOINT, CLIENT_ID, PATH_TO_CERTIFICATE, PATH_TO_PRIVATE_KEY, PATH_TO_AMAZON_ROOT_CA_1, MESSAGE, TOPIC, and RANGE
ENDPOINT = "customEndpointUrl"
CLIENT_ID = "testDevice"
PATH_TO_CERTIFICATE = "certificates/a1b23cd45e-certificate.pem.crt"
PATH_TO_PRIVATE_KEY = "certificates/a1b23cd45e-private.pem.key"
PATH_TO_AMAZON_ROOT_CA_1 = "certificates/root.pem"
MESSAGE = "Hello World"
TOPIC = "test/testing"
RANGE = 20

myAWSIoTMQTTClient = AWSIoTPyMQTT.AWSIoTMQTTClient(CLIENT_ID)
myAWSIoTMQTTClient.configureEndpoint(ENDPOINT, 8883)
myAWSIoTMQTTClient.configureCredentials(PATH_TO_AMAZON_ROOT_CA_1, PATH_TO_PRIVATE_KEY, PATH_TO_CERTIFICATE)

myAWSIoTMQTTClient.connect()
print('Begin Publish')
for i in range (RANGE):
    data = "{} [{}]".format(MESSAGE, i+1)
    message = {"message" : data}
    myAWSIoTMQTTClient.publish(TOPIC, json.dumps(message), 1)
    print("Published: '" + json.dumps(message) + "' to the topic: " + "'test/testing'")
    t.sleep(0.1)
print('Publish End')
myAWSIoTMQTTClient.disconnect()

測試設定

1. 在 AWS IoT Core 主控台的左側導覽窗格中，選擇測試。

2. 在 MQTT 用戶端頁面上，對於訂閱主題，輸入 test/testing。

3. 選擇訂閱主題。名為 test/testing 的測試主題已準備就緒，可用於測試訊息發佈。如需詳細資訊，請參閱使用 AWS IoT MQTT 用戶端檢視裝置 MQTT 訊息。

4. 從命令列執行下列命令：

python3 publish.py

Python 程式會將 20 則測試訊息，發佈至您在 AWS IoT Core 主控台中建立的 test/testing 主題。在主控台中檢視此主題以查看已發佈的訊息。

**提示：**您也可以使用隨附的 pubsub 範例，測試 SDK 的其他功能，例如透過 WebSockets 訂閱和連接。如需詳細資訊，請參閱 GitHub 上的 pubsub (適用於 Python v2 的 AWS IoT SDK) 或 BasicPubSub (適用於 Python 的 AWS IoT Device SDK)。