Global outage event

If you're experiencing issues with your AWS services, then please refer to the AWS Health Dashboard. You can find the overall status of ongoing outages, the health of AWS services, and the latest updates from AWS engineers.

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

5 分的閱讀內容

我無法在 AWS IoT Core 與我的裝置或 MQTT 用戶端之間傳送或接收 MQTT (訊息佇列遙測傳輸) 訊息。

簡短說明

請確認您的 AWS IoT 實物已正確設定，且其憑證已正確附加。若要測試您的設定，您可以使用 AWS IoT MQTT 用戶端與本文提供的 Python 範例程式碼。

解決方法

設定目錄以測試 MQTT 發佈

在您的開發環境中建立一個工作目錄。例如，建立名為 iot-test-publish 的目錄。
在新的工作目錄中建立憑證子目錄。例如，建立名為 certificates 的子目錄。
從命令列將目錄變更為新的工作目錄。

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

注意：在開始之前，請為 Python 3 封裝安裝 pip。如需更多資訊，請參閱 Python Packaging Authority (PyPA) 網站上的安裝。

若要安裝 AWS IoT SDK for Python v2，請使用以下命令：
```
pip install awsiotsdk
```
-或-

若要安裝 AWS IoT Device SDK for Python (先前的 SDK 版本 v1)，請使用以下命令：
```
pip install AWSIoTPythonSDK
```

**注意：**最佳實務是使用 AWS IoT SDK for Python 版本的最新更新。如需更多資訊，請參閱 GitHub 上的 AWS IoT SDK for Python v2 或 AWS IoT Device SDK for Python。最佳實務也建議您使用這些 SDK 連線至 AWS IoT Core，但這些 SDK 並非必要，您可以使用任何其他符合規範的第三方 MQTT 用戶端連線至 AWS IoT Core。

建立 AWS IoT Core 政策

開啟 AWS IoT Core console (AWS IoT Core 主控台)。
在左側導覽窗格中，選擇 Secure (安全)。
在 Secure (安全) 下，選擇 Policies (政策)。
如果您已有現有的 AWS IoT Core 政策，請選擇 Create (建立) 以建立新政策。
-或-
如果您沒有現有政策，請在 You don't have any policies yet page (您尚未建立任何政策頁面) 上，選擇 Create a policy (建立政策)。
在 Create a policy (建立政策) 頁面上，為您的政策輸入 Name (名稱)。例如，輸入名稱 admin。
在 Add statements (新增陳述式) 下，完成以下動作：
在 Action (動作) 中，輸入 iot:*。
**重要：**您可以在測試期間允許所有 AWS IoT 動作 (iot:*)。然而，在生產環境設定中，最佳實務是提高安全性。如需更安全的政策範例，請參閱 AWS IoT Core 政策範例。
在 Resource ARN (資源 ARN) 中，輸入 *。
在 Effect (效果) 中，選取 Allow (允許) 核取方塊。
選擇 Create (建立)。

如需更多資訊，請參閱建立 AWS IoT 政策與 AWS IoT Core 政策。

建立 AWS IoT 實物

**注意：**您不需要建立實物即可連線至 AWS IoT。然而，實物可讓您使用 additional security controls (其他安全控制) 與其他 AWS IoT 功能，例如 Fleet Indexing (機群索引)、Jobs (作業) 或 Device Shadow (裝置影子)。

在 AWS IoT Core 主控台中，於左側導覽窗格選擇 Manage (管理)。
如果您已有現有實物，請選擇 Create (建立) 以建立新的實物。
-或-
如果您沒有任何現有實物，請在 You don't have any things yet (您尚未建立任何實物) 頁面上，選擇 Register a thing (註冊實物)。
在 Creating AWS IoT things (建立 AWS IoT 實物) 頁面上，選擇 Create a single thing (建立單一實物)。
在 Add your device to the thing registry (將您的裝置新增至實物登錄) 頁面上，完成以下動作：
輸入您實物的名稱。例如，輸入 Test-Thing。
(選用) 在 Add a type to this thing (為此實物新增類型) 下，選擇或建立 thing type (實物類型)。
(選用) 在 Add this thing to a group (將此實物新增至群組) 下，選擇或建立群組。如需群組的更多資訊，請參閱靜態實物群組與動態實物群組。
(選用) 在 Set searchable thing attributes (optional) (設定可搜尋的實物屬性 (選用)) 下，將屬性新增為鍵值組。
選擇 Next (下一步)。
**注意：**最佳實務是讓實物名稱與 clientId 相符。若名稱相符，您即可使用 AWS IoT Core 功能，例如機群索引與實物政策變數。
在 Add a certificate for your thing (為您的實物新增憑證) 頁面上，選擇 Create certificate (建立憑證)。您會看到通知，確認您的實物與其憑證已建立。
在 Certificate created (已建立憑證) 頁面上，完成以下動作：
在 In order to connect a device, you need to download the following (若要連線裝置，您需要下載以下項目) 下，為憑證、公有金鑰與私有金鑰選擇 Download (下載)。
將每個下載的檔案儲存至您先前建立的憑證子目錄。
在 You also need to download a root CA for AWS IoT (您也需要為 AWS IoT 下載根 CA) 下，選擇 Download (下載)。Server authentication (伺服器驗證) 頁面會顯示 CA certificates for server authentication (用於伺服器驗證的 CA 憑證)。
在 Amazon Trust Services Endpoints (preferred) (Amazon 信任服務誤端點 (建議)) 下，選擇 Amazon Root CA 1 (Amazon 根 CA 1)。憑證會在您的瀏覽器中開啟。
複製憑證 (從 -----BEGIN CERTIFICATE----- 到 -----END CERTIFICATE----- 的所有內容) 並貼到文字編輯器中。
將憑證儲存為名為 root.pem 的 .pem 檔案，並儲存至憑證子目錄。
在 AWS IoT Core 主控台的 Certificate created (已建立憑證) 頁面上，選擇 Activate (啟用)。按鈕會變更為 Deactivate (停用)。
選擇 Attach a policy (附加政策)。
在 Add a policy for your thing (為您的實物新增政策) 頁面上，完成以下動作：
選取您先前建立的 AWS IoT Core 政策。例如，選取 admin (管理員)。
選擇 Register Thing (註冊實物)。

如需更多資訊，請參閱建立實物物件。

複製 AWS IoT Core 端點網址

在 AWS IoT Core 主控台中，於導覽窗格選擇 Settings (設定)。
在 Settings (設定) 頁面上，在 Custom endpoint (自訂端點) 下，複製 Endpoint (端點)。此 AWS IoT Core 自訂端點網址對您的 AWS 帳戶與區域而言是唯一的。

建立 Python 程式檔案

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

若使用 AWS IoT SDK for Python v2，請使用以下範例程式碼：

**重要：**將 customEndpointUrl 替換為您的 AWS IoT Core 自訂端點網址。將 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()

-或-

若使用 AWS IoT Device SDK for Python v1，請使用以下範例程式碼：

# 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()

**注意：**舊版 AWS IoT Device SDK for Python 不支援 Python 3.12 或更新版本。請確保您使用 AWS IoT Device SDK for Python 的最新版本。

測試設定

在 AWS IoT Core 主控台中，於左側導覽窗格選擇 Test (測試)。
在 MQTT client (MQTT 用戶端) 頁面上，於 Subscription topic (訂閱主題) 輸入 test/testing。
選擇 Subscribe to topic (訂閱主題)。名為 test/testing 的測試主題已準備就緒，可用於測試訊息發佈。如需更多資訊，請參閱使用 AWS IoT MQTT 用戶端檢視 MQTT 訊息。
從您的命令列執行以下命令：
```
python3 publish.py
```

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

**提示：**您也可以使用內含的 pubsub 範例測試 SDK 的其他功能，例如 WebSockets 訂閱與連線。如需更多資訊，請參閱 GitHub 網站上的 pubsub (AWS IoT SDK for Python v2) 或 BasicPubSub (AWS IoT Device SDK for Python)。

(選用) 啟用 AWS IoT 記錄至 Amazon CloudWatch

您可以監控您發佈至 AWS IoT Core 的 MQTT 訊息事件日誌。如需設定指示，請參閱設定 AWS IoT 記錄與使用 CloudWatch Logs 監控 AWS IoT。