ESP32-C6安全(SSL)MQTT服务点灯全套流程
2024-04-093543
AI 快速预览详细
本文介绍了如何在micropython中实现ESP32-C6的安全MQTT连接。首先，你需要申请一个14天免费的只能用SSL连接的MQTT服务，如EMQX云服务。然后，在micropython中配置SSL上下文，设置验证模式为ssl.CERT_NONE。接下来，通过选择阻塞模式或非阻塞模式来控制消息接收的方式。最后，使用MQTTX或其他客户端工具发送ON或OFF消息到定义的主题，即可控制LED的亮灭。
|
在IoT物联网开发中，MQTT是不可或缺的一个好东西。通常情况下，我们会直接使用MQTT的1883端口进行连接，在一些对安全要求较高的场合，会需要使用基于TLS/SSL的安全连接。 下面就说一下在micropython中使用具体的步骤。 一、申请一个14天免费的只能用SSL连接的MQTT服务 申请网址：https://www.emqx.com/zh/cloud  注册后，直接新建一个部署，选择ServerLess即可：  最终得到一个开箱即用的安全MQTT服务：  然后到访问控制中，添加一个账号密码：  现在，可以使用你熟悉的MQTT客户端工具，如MQTTX来进行连接测试：   二、micropython编码 在上面的代码中，核心是ssl的设置： 另外，使用BLOCK_MODE设置是否阻塞模式，实际作用如下：
三、测试 用MQTTX或者其他客户端工具，如手机端的mqttdashboard，发送ON或者OFF到定义的主题即可：  运行mpy后，会收到消息，并根据消息ON、OFF来亮灭LED了  |
[code]
import mqtt
import network
import ssl
import time
import os
from machine import Pin
from umqtt.simple import MQTTClient
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect('WiFi名称', '密码')
while not wlan.isconnected():
print("try connect...")
time.sleep(1)
print("connect ok")
CLIENT_ID = b'test_from_esp32c6'
MQTT_SERVER = 'p52a5af8.ala.cn-hangzhou.emqxsl.cn'
PORT = 8883
USERNAME = b'esp32c6'
PASSWORD = b'密码'
TOPIC = b'esp32c6/test'
BLOCK_MODE = True # 是否使用阻塞模式
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
context.verify_mode = ssl.CERT_NONE
try:
client = MQTTClient(client_id=CLIENT_ID, # something unique to identify device, ie machine.unique_id()
server= MQTT_SERVER,
port=PORT, # if port ==0 AND ssl=True -> port=8883 the secure mqtt port
user=USERNAME,
password=PASSWORD,
keepalive=7200,
ssl=context
)
client.connect()
print('Connected to %s MQTT Broker'%(MQTT_SERVER))
except OSError as e:
print('Failed to connect to the MQTT Broker.')
time.sleep(5)
machine.reset()
def get_msg(topic, msg):
print(msg)
if msg == b'ON':
led.value(True)
elif msg == b'OFF':
led.value(False)
led=Pin(15, Pin.OUT)
client.set_callback(get_msg)
client.subscribe(TOPIC)
counter = 0
while True:
try:
counter = counter + 1
print(counter)
if BLOCK_MODE:
client.wait_msg() # 阻塞
else:
client.check_msg() # 非租塞
time.sleep(1)
except Exception as e:
print(str(e))
time.sleep(2)
print('reconnect!')
client.connect()
# 断线之后要重新订阅
client.subscribe(TOPIC)
[/code]