使用 Python 时，如何从我的设备向 AWS IoT Core 发布 MQTT 消息？

4 分钟阅读

我无法在 AWS IoT Core 和我的设备或 MQTT 客户端之间发送或接收 MQTT（消息队列遥测传输）消息。如何向 AWS IoT Core 发布 MQTT 消息？

简短描述

确认您的 AWS IoT 事物是否配置正确并且是否已相应地附上其证书。要测试您的设置，可使用 AWS IoT MQTT 客户端和本文中提供的 Python 代码示例。

解决方法

设置一个目录来测试 MQTT 发布

1.在开发环境中创建工作目录。例如：iot-test-publish。

2.在新的工作目录中为证书创建子目录。例如：certificates。

3.在命令行中，将该目录更改为新的工作目录。

安装 Pip 和 AWS IoT SDK for Python

1.如果您还未安装，请为 Python 3 包安装 Pip。有关更多信息，请在 Python Packaging Authority（PyPA）网站上参阅 Installation。

2.通过从命令行运行以下命令安装 AWS IoT SDK for Python v2：

pip install awsiotsdk

-或-

通过运行以下命令安装 AWS IoT Device SDK for Python（之前的 SDK 版本）：

pip install AWSIoTPythonSDK

有关更多信息，请在 GitHub 上参阅 AWS IoT SDK for Python v2 或AWS IoT Device SDK for Python。

**注意：**建议将这些 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.选择创建。

有关更多信息，请参阅 Create an AWS IoT Core policy和 AWS IoT Core policies。

创建 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 Root CA 1。证书将在浏览器中打开。

8.复制证书（从**-----BEGIN CERTIFICATE-----** 到**-----END CERTIFICATE-----**的所有内容）并将其粘贴到文本编辑器中。

9.将证书以名为 root.pem 的 .pem 文件保存到证书子目录中。

10.在 AWS IoT Core 控制台的已创建证书页面上，选择启用。该按钮变为停用。

11.选择附加策略。

12.在添加事物的策略页面上，执行以下操作：
选择您之前创建的 AWS IoT Core 策略。例如：admin。
选择注册事物。

有关更多信息，请参阅 Create a thing object。

复制 AWS IoT Core 端点 URL

1.在 AWS IoT Core 控制台左侧的导航窗格中，选择设置。

2.在设置页面的自定义端点下，复制端点。此 AWS IoT Core 自定义端点 URL 是专门针对您的 AWS 账户和区域的。

创建 Python 程序文件

将以下 Python 代码示例之一保存为名为 publish.py 的 Python 程序文件。

如果您之前安装了 AWS IoT SDK for Python v2，请使用以下示例代码：

**重要事项：**将 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()

-或-

如果您安装了 AWS IoT Device SDK for Python（之前的 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 的测试主题可用于测试消息发布。有关更多信息，请参阅 View device MQTT messages with the AWS IoT MQTT client。

4.从命令行运行以下命令：

python3 publish.py

Python 程序在 AWS IoT Core 控制台中向您创建的主题 test/testing 发布了 20 条测试消息。在控制台中查看该主题，以了解发布的消息。

**提示：**您还可以使用随附的 Pubsub 示例测试 SDK 的其他功能，例如通过 WebSockets 进行订阅和连接。有关更多信息，请在 GitHub 上参阅 pubsub（AWS IoT SDK for Python v2）或 BasicPubSub（AWS IoT Device SDK for Python）。

（可选）激活对 Amazon CloudWatch 的 AWS IoT 日志记录

您可以监控发布到 AWS IoT Core 的 MQTT 消息的事件日志。有关设置的相关说明，请参阅 Configure AWS IoT logging 和 Monitor AWS IoT using CloudWatch Logs。