直接进入正题。这几天在折腾 ESP32 做数据上报,把 HTTP GET 和 POST 请求用 MicroPython 完整跑通了一遍。这东西网上教程不少,但大多是零散片段,要么只讲 GET 不提 POST,要么代码一跑就报错。我把自己踩过的坑、验证过的写法整理成这篇东西,目标是让一个刚拿到 ESP32 的开发者,照着敲完就能正常收发请求。
先说结论:MicroPython 里做 HTTP 请求,官方推荐的是urequests这个库,它从 CPython 的requests库移植而来,API 基本一致,能覆盖绝大多数物联网场景。但受限于 ESP32 的内存和固件裁剪,用起来有几个关键差异,不搞清楚会走很多弯路。
1. 项目整体设计与思路拆解
1.1 为什么用 MicroPython 做 HTTP 请求
很多人会纠结到底用 Arduino 还是 MicroPython 来写 ESP32。我的看法是:如果你要快速验证一个物联网原型,或者你在 Python 生态里比较熟练,MicroPython 的开发效率非常高。它自带交互式解释器,改完代码直接运行,不需要编译烧录,调试体验接近脚本语言。
HTTP 请求这块,MicroPython 的优势更明显。urequests封装了 TCP 连接、请求头构造、响应解析这些底层细节,几行代码就能发请求。换做 Arduino 的HTTPClient库,虽然也能做,但代码量明显更大,字符串处理也更繁琐。
但反过来,MicroPython 解释执行带来的性能开销和内存占用是实实在在的代价。一个 GET 请求返回几 KB 的 JSON 数据,解析完可能占掉几十 KB 的 RAM,这在内存只有 320KB 的 ESP32 上需要精打细算。所以实际项目中,我一般只在数据量小、请求频率低(比如几秒一次)的场景用 MicroPython,高频大批量数据传输还是得上 C 语言方案。
1.2 核心需求解析:GET 与 POST 各解决什么问题
HTTP 协议里的 GET 和 POST,本质上是语义不同的两种请求方式。GET 用于从服务端获取资源,请求参数拼在 URL 后面,比如查询某个传感器的当前值;POST 用于向服务端提交数据,数据放在请求体里,比如上报温湿度记录。
在 ESP32 的实际场景里,两者的分工非常明确:
- GET 请求:设备主动拉取配置、获取指令、查询状态。比如设备开机后向服务器请求最新的运行参数,或者定时轮询是否有新的控制命令。
- POST 请求:设备上报数据、提交日志、发送告警。比如温度传感器每分钟上报一次读数,或者门磁触发时发送告警事件。
选择哪种方式,不仅是后端接口设计的问题,也影响设备端的处理和容错。GET 的优点是幂等,重复请求没有副作用,适合轮询场景;POST 的优点是数据容量大、不会把敏感参数暴露在 URL 里,适合上行数据。
1.3 技术选型:urequests 库与网络连接基础
MicroPython 做 HTTP 请求,核心依赖两个东西:WiFi 连接和urequests库。前者通过network模块实现,后者直接 import 使用。
import network import urequests有个细节很多人不知道:urequests默认不校验 HTTPS 证书。这跟 CPython 的requests完全不同。如果请求的是 HTTPS 接口,它其实是在做裸的 TLS 传输,不验证服务端身份,这在安全上是有隐患的。但在实际物联网场景里,设备端受限于算力和存储,很多项目就妥协了。如果业务要求必须验证证书,MicroPython 不是不能做,但需要提前把 CA 证书以文件形式烧录到设备里,写起来会比较繁琐,而且不同固件版本的 API 有差异。
所以我的建议是:如果不是强安全场景,先用 HTTP 把功能跑通,再考虑升级到 HTTPS。很多内网 IoT 项目直接 HTTP 完全够用。
2. 核心细节解析与实操要点
2.1 连接 WiFi 的完整流程与状态检查
发起 HTTP 请求的前提是 ESP32 已经接入局域网或互联网。WiFi 连接这块踩过坑的人不少,核心问题是:不能发完连接指令就立刻发起请求,必须等连接状态稳定。
import network import time wlan = network.WLAN(network.STA_IF) wlan.active(True) wlan.connect('你的WiFi名称', '你的WiFi密码') # 等待连接成功,超时设为10秒 timeout = 10 while timeout > 0: if wlan.isconnected(): break time.sleep(1) timeout -= 1 if wlan.isconnected(): print('WiFi已连接,IP地址:', wlan.ifconfig()[0]) else: print('连接失败')这里有几个要点:
- 必须调用
wlan.active(True)把网卡激活,否则 connect 不会生效。 - 连接是异步过程的,
connect()方法会立即返回,但此时不一定已经连上,需要轮询isconnected()或等待STAT_GOT_IP事件。 - 连接失败后重试的次数和间隔要控制好,不要死循环卡住整个程序。我一般重试 3 次,每次间隔 5 秒,还不行就重启设备。
有个很实用的小技巧:把 WiFi 连接封装成一个函数,返回布尔值表示是否成功。这样主程序里可以清晰判断网络状态,避免请求发出去才发现没网。
def connect_wifi(ssid, password, timeout=10): wlan = network.WLAN(network.STA_IF) wlan.active(True) wlan.connect(ssid, password) while timeout > 0: if wlan.isconnected(): return True time.sleep(1) timeout -= 1 return False注意:ESP32 的 WiFi 模块在某些固件版本上有个已知问题,
wlan.disconnect()后立刻重新connect()可能会失败。稳妥做法是wlan.active(False)再active(True),相当于软复位网卡。
2.2 理解 GET 请求的 URL 参数拼接规则
GET 请求的参数以键值对的形式拼在 URL 末尾,用?分隔多个参数,参数之间用&连接。比如:
http://192.168.1.100:8080/api/sensor?temp=25.6&humidity=60手工拼字符串的风险在于:如果参数值里包含特殊字符(比如中文、空格、&、=),拼出来的 URL 可能被服务端解析错误。MicroPython 的标准库里没有urllib.parse.urlencode这样现成的函数,需要自己写一个简单的 URL 编码。
from urllib.parse import quote def build_query(params): return '&'.join([f'{quote(str(k))}={quote(str(v))}' for k, v in params.items()])quote函数会做百分比编码,把空格变成%20,把中文字符变成 UTF-8 字节的百分号形式。这样服务端收到的参数就是标准格式,不会出解析问题。
我实际测试过,如果只是传数字和简单的 ASCII 字母,手工拼接完全没问题。但如果要传中文描述或者带特殊符号的文本,一定要做编码。某次上报设备备注信息时就因为没做编码,服务端收到的字符串被截断,排查了很久才发现问题。
2.3 POST 请求的 Content-Type 选择与数据处理
POST 请求的数据放在请求体(body)里,服务端需要知道怎么解析这个 body,这就靠Content-Type头来约定。最常见的两种格式:
application/json:JSON 格式字符串,适合结构化数据,后端用request.json()解析。application/x-www-form-urlencoded:键值对形式,类似 GET 的查询字符串,后端用表单方式解析。
在urequests里,通过json参数或data参数指定不同类型的数据:
import urequests # JSON 格式 payload = {'temperature': 25.6, 'humidity': 60} response = urequests.post('http://192.168.1.100:8080/api/data', json=payload) # 表单格式 payload = 'temperature=25.6&humidity=60' response = urequests.post('http://192.168.1.100:8080/api/data', data=payload)这里有个关键差异:传json参数时,urequests会自动把字典转成 JSON 字符串,并设置Content-Type: application/json;传data参数时,它直接把字符串作为 body 发送,默认 Content-Type 不是application/x-www-form-urlencoded,需要手动指定。
response = urequests.post( 'http://192.168.1.100:8080/api/data', data='temperature=25.6&humidity=60', headers={'Content-Type': 'application/x-www-form-urlencoded'} )我在对接某些后端框架时发现,FastAPI 和 Flask 对这两种格式的处理方式完全不同。Flask 的request.form只解析表单格式,request.get_json()只解析 JSON。如果 Content-Type 设置错了,后端要么收不到参数,要么直接报 415 错误。
2.4 超时设置与响应数据的正确关闭
urequests的所有请求方法都支持timeout参数,单位是秒。这个参数极其重要,因为 ESP32 的 TCP 栈在连接不上时默认会等很久,如果没有超时机制,程序可能卡在请求那里几十秒甚至更久。
try: response = urequests.get('http://192.168.1.100:8080/api/ping', timeout=5) print(response.status_code) print(response.text) response.close() except Exception as e: print('请求异常:', e)关于response.close(),网上很多教程都漏了这一行。MicroPython 设备内存有限,urequests的响应对象底层持有一个 socket,不 close 的话 socket 不会被释放,连续请求几十次就可能把内存耗尽,导致设备崩溃重启。
我的习惯是:
- 每次请求完成立即调用
response.close()。 - 如果只需要 JSON 数据,直接
response.json()拿到结果后立刻 close。 - 把请求逻辑写进
try...except里,确保异常时也能关闭连接。
响应对象常用属性:
response.status_code:HTTP 状态码,200 表示成功,404 表示资源不存在,500 表示服务端错误。response.text:响应体字符串形式。response.json():尝试把响应体解析成字典或列表,失败会抛异常。response.headers:响应头字典(注意包含原始头信息和 HTTPHeaderDict 包装的差异)。
我踩过的一个坑:调用response.json()后不要再用response.text,因为json()内部会读取并缓存内容,如果顺序反了或者反复读取,有些固件版本会返回空字符串。
3. 实操过程与核心环节实现
3.1 最小可用代码:一次完整的 GET 请求
下面这段代码是我验证过的最小可运行示例,完成"连接 WiFi → 发起 GET 请求 → 解析响应 → 关闭连接"的完整链路。
import network import urequests import time def connect_wifi(ssid, password, timeout=10): wlan = network.WLAN(network.STA_IF) wlan.active(True) wlan.connect(ssid, password) while timeout > 0 and not wlan.isconnected(): time.sleep(1) timeout -= 1 return wlan.isconnected() # 配置 WiFi SSID = '你的WiFi名称' PASSWORD = '你的WiFi密码' if not connect_wifi(SSID, PASSWORD): print('WiFi 连接失败,程序退出') raise SystemExit print('WiFi 已连接') # 发起 GET 请求 url = 'http://192.168.1.100:8080/api/sensor?temp=25.6' try: response = urequests.get(url, timeout=5) print('状态码:', response.status_code) print('响应内容:', response.text) except Exception as e: print('请求出错:', e) finally: response.close()运行这段代码,串口输出类似:
WiFi 已连接 状态码: 200 响应内容: {"status": "ok", "message": "数据已接收"}注意几个细节:
raise SystemExit用来终止脚本,在 MicroPython 的 REPL 环境里比sys.exit()更干脆。finally块里调用response.close(),确保即使发生异常,连接也能被释放。这比只在成功路径里 close 更稳妥。- 超时设为 5 秒,实测如果 WiFi 刚连上就立刻发请求,偶尔会因为 TCP 栈还没准备好而超时,这时候重试一次基本能成功。
3.2 带重试机制的 GET 请求:解决弱网环境
日常使用中,WiFi 信号不稳定、服务端暂时不可用等情况都会导致请求失败。直接让程序崩溃显然不可取,加一个简单的重试机制就稳很多。
import time import urequests def http_get_with_retry(url, retries=3, timeout=5): for i in range(retries): try: response = urequests.get(url, timeout=timeout) if response.status_code == 200: data = response.text response.close() return data else: print(f'第 {i+1} 次请求,状态码: {response.status_code}') response.close() except Exception as e: print(f'第 {i+1} 次请求异常: {e}') time.sleep(2) # 退避间隔 return None这个函数的设计逻辑:
- 最多重试 3 次,避免无限重试造成资源浪费。
- 每次重试间隔 2 秒,给网络和服务端一个恢复的时间窗口。
- 只有收到 200 才认为成功,其他状态码都算失败需要重试。
- 返回
None表示最终失败,调用方可以做降级处理,比如暂存数据等待下次上报。
在弱网环境实测下来,重试机制能把请求成功率从 70% 提升到 95% 以上。当然,如果服务端明确返回 404 这种"永远不可能成功"的状态码,重试也没有意义,这时候可以根据状态码分类处理:5xx 和网络异常做重试,4xx 直接放弃。
3.3 POST 请求完整实现:JSON 数据上报
POST 请求最常见的场景是数据上报。下面的代码演示了如何构造 JSON 数据并发送:
import network import urequests import json import time def post_json_data(url, payload, timeout=5): try: headers = {'Content-Type': 'application/json'} response = urequests.post(url, json=payload, headers=headers, timeout=timeout) print('状态码:', response.status_code) print('响应:', response.text) response.close() return response.status_code == 200 or response.status_code == 201 except Exception as e: print('POST 请求异常:', e) return False # 模拟传感器数据 payload = { 'device_id': 'esp32_001', 'temperature': 25.6, 'humidity': 60.2, 'battery': 3.7, 'timestamp': time.time() } url = 'http://192.168.1.100:8080/api/upload' success = post_json_data(url, payload) if success: print('数据上报成功') else: print('数据上报失败,准备本地缓存')我在实际项目里做的是,采集一次数据立刻上报,失败的话把数据追加到本地文件里,等下次开机或者网络恢复后再批量补报。这个"离线缓存"的思路对物联网设备非常重要,毕竟 WiFi 不可能永远稳定。
关于时间戳,time.time()在 MicroPython 里返回的是 2000 年 1 月 1 日以来的秒数,跟 CPython 的 Unix 时间戳差了 30 年左右。如果服务端用标准 Unix 时间戳解析,记得加上偏移量(31536000 * 30 或直接用time.mktime转换到本地时间后再算),否则后端看到的时间会非常奇怪。
3.4 处理二进制数据与文件上传场景
有些场景要上报的不是 JSON,而是图片、音频或者其他二进制文件。比如带摄像头的 ESP32 抓拍一张图片,POST 到服务器保存。这时候 body 直接塞二进制数据就行。
def upload_binary(url, filepath): with open(filepath, 'rb') as f: data = f.read() headers = {'Content-Type': 'application/octet-stream'} response = urequests.post(url, data=data, headers=headers, timeout=30) print('状态码:', response.status_code) response.close()注意几个点:
data参数接受 bytes 类型,用open('rb')读取文件避免文本编码问题。- 大文件(超过 100KB)在 ESP32 上要小心,因为
data会被完整加载到内存。如果文件大到内存装不下,需要用分段传输(chunked)或者流式读取,但urequests对 chunked 的支持有限,真要传大文件不如用 FTP 或者 MQTT 的二进制消息。 - 超时要适当调大,因为传输几十 KB 的数据在慢速 WiFi 下可能要好几秒。
3.5 服务端地址选择:IP 直连与域名解析
开发阶段我们通常直接用局域网 IP 访问,比如http://192.168.1.100:8080。生产环境一般用域名。这两者的区别在于:域名需要 DNS 解析,而 MicroPython 的 DNS 解析有时不太稳定。
如果遇到"局域网 IP 能通,域名不通"的情况,排查思路是:
- 先用
socket.getaddrinfo(host, port)测试 DNS 解析是否正常。 - 检查路由器是否配置了正确的 DNS 服务器。
- 尝试在代码里手动指定 DNS。
import socket addr_info = socket.getaddrinfo('example.com', 80) print(addr_info)如果输出类似[(2, 1, 0, '', ('93.184.216.34', 80))],说明解析成功。如果卡住不动或者报错,那就是 DNS 的问题。
另外我建议把 URL 的 base 和路径分开管理,这样切换环境和接口版本时不用改所有请求代码:
BASE_URL = 'http://192.168.1.100:8080' API_PATH = '/api/v2/upload' url = BASE_URL + API_PATH3.6 解析服务端返回的 JSON 数据
请求发出去只是第一步,更重要的是处理好响应。urequests的response.json()方法可以直接把 JSON 字符串转成 Python 对象。
response = urequests.get('http://192.168.1.100:8080/api/status', timeout=5) if response.status_code == 200: data = response.json() print('风扇状态:', data.get('fan', {}).get('speed')) print('设备模式:', data.get('mode', 'auto')) response.close()这里有个细节:data.get('fan', {}).get('speed')这种写法,在嵌套 JSON 结构里很实用,避免因为某个字段缺失直接抛KeyError。我见到很多新手用data['fan']['speed']访问,一旦服务端返回的 JSON 结构有变动,代码就崩了。
还有一个 MemoryError 的问题。如果服务端返回的 JSON 很大(几 KB 以上),json()解析过程中会产生临时对象,对 ESP32 的 RAM 压力很大。实测下来,超过 20KB 的 JSON 响应,MicroPython 基本就会因为内存不足抛异常。
try: data = response.json() except MemoryError: print('响应过大,内存不足')面对这种情况,方案是与后端协商精简返回内容,只返回必要的字段;或者改用流式解析方式,一次只处理一部分数据。物联网设备端还是越精简越好。
4. 常见问题与排查技巧实录
4.1 连接超时:为什么我的请求经常卡住
现象:程序执行到urequests.get()就卡在那里,过很久才报超时错误,或者直接没有任何响应。
核心原因有两种。第一种是 WiFi 连接不稳定。ESP32 的 WiFi 模块在信号弱的时候,TCP 建连会反复尝试,耗时很长。第二种是服务端没有正确响应,比如端口没监听、IP 地址错误、防火墙拦截等。
排查顺序:
- 用
ping命令在电脑上测试目标 IP 是否可达。 - 检查 ESP32 是否真的连上了 WiFi,打印
wlan.isconnected()和wlan.ifconfig()确认拿到 IP。 - 在代码里用
socket.connect()测试目标端口是否开放。如果 socket 层能连上,那就是应用层的问题。
实际上有一个很常见的坑:ESP32 连的 WiFi 是 5GHz 频段,或者路由器开了 AP 隔离,导致设备之间不能互相访问。这时候 ESP32 显示 WiFi 已连接、也有 IP,但就是无法访问局域网内的其他设备。排查时需要确认 ESP32 支持 2.4GHz 频段,以及路由器没有隔离客户端。
4.2 内存不足(MemoryError)的诱因与对策
ESP32 的内存是有限的,MicroPython 运行时本身占一部分,连接 WiFi、创建 socket 都要消耗内存。请求越多、响应越大,内存碎片化越严重。
我踩过的一个典型问题:连续运行几个小时后,突然在某次请求时报 MemoryError,重启后又好了。这是因为内存碎片化导致可用连续内存块变小,虽然总剩余内存还有,但不足以分配一个大的 buffer。
对策:
- 降低请求频率,给系统一个回收内存的时间。
- 尽量用小的 buffer 传输数据,不要一次性读入大文件。
- 定期检查
gc.mem_free()监控内存余量,在程序中定期调用gc.collect()手动回收。 - 如果确认是内存碎片化,可以考虑在请求密集时先断开 WiFi 再重连,因为重连会释放旧的 socket 和相关对象,相当于一次软清理。
import gc gc.collect() print('剩余内存:', gc.mem_free())4.3 服务端收不到 POST 数据的排查思路
POST 请求发出去了,状态码也返回 200,但服务端日志里显示请求参数为空。这个问题我排查过很多次,大部分是两种原因:
Content-Type 不匹配。前面说过,传
json参数时urequests会自动设Content-Type: application/json,但如果你传的是data字符串,需要手动指定Content-Type: application/x-www-form-urlencoded。我见过有人混用了json和data两个参数,导致服务端解析失败。数据格式不符合预期。有些后端框架要求 POST 的 body 是特定格式,比如特殊网关要求的 XML 格式必须带
application/xml类型。用data参数传递时,务必确认对方能解析你的格式。
排查技巧:在服务端打印原始请求头和 body,看收到的到底是什么。如果服务端是简单的 Flask 应用,临时加一段打印即可;如果用的是现成平台,看请求日志一般也能看出来。
@app.route('/api/upload', methods=['POST']) def upload(): print('Headers:', request.headers) print('Body:', request.get_data()) return 'ok'4.4 中文和特殊字符导致的编码问题
向服务端 POST 中文内容时,MicroPython 的urequests默认发送 UTF-8 编码的字节流,这本身没问题。但有些老系统的后端默认按 Latin-1 或 GBK 解析,就会产生乱码。
处理方式是在请求头里显式声明字符集:
headers = {'Content-Type': 'application/json; charset=utf-8'}同时,发送前把字符串手动编码成 UTF-8 字节:
data = json.dumps(payload).encode('utf-8') response = urequests.post(url, data=data, headers=headers)这样能最大程度避免编码不一致的混乱。因为urequests在传json参数时,内部调用的json.dumps生成的字符串然后在传输时按 UTF-8 编码,但显式写出来能让双方都明确约定,排查问题也方便。
4.5 判断请求是否成功的几个指标
开发过程中,怎么知道请求到底通没通?不同层面的信号提供不同精度的判断:
| 判断方式 | 可靠程度 | 说明 |
|---|---|---|
打印response.status_code | 高 | 服务端有响应,200-299 表示成功 |
打印response.text | 中 | 能看到返回内容,但需要代码执行到这一步 |
| 串口监视器的异常堆栈 | 中 | 能定位是连接、超时还是解析出了问题 |
| 服务端日志 | 极高 | 后端是否收到请求一目了然 |
建议是:开发阶段把状态码、响应头和响应体全部打印出来,运行稳定后再删除调试代码。
4.6 固件版本与模块兼容性排查
urequests在大多数官方 MicroPython 固件里都存在,但个别精简版固件(比如某些 ESP32-C3 专用固件)可能没有集成这个模块。如果 import 时报错ModuleNotFoundError,有两个解决方向:
- 换用官方标准固件,确认
urequests.py在modules目录下。 - 手动从 MicroPython 源码仓库下载
urequests.py,用 ampy 或 thonny 上传到设备/lib目录。
另外,不同固件版本对urequests内部实现的细节有差异。比如有些版本支持response.json()直接解析,有些版本需要先用ujson.loads(response.text)。统一的做法是先用response.text拿到字符串,再用ujson.loads()手动解析,兼容性最好。
import ujson text = response.text data = ujson.loads(text) # 手动解析,兼容性优先5. 进阶优化与稳定运行技巧
5.1 用定时器和主循环结合控制请求节奏
很多应用需要周期性上报数据,比如每 30 秒上报一次温湿度。最自然的方式是在主循环里加time.sleep(30),但这样有个问题:循环卡住时无法响应其他事件(比如按键、消息)。
如果应用同时有其他任务,可以把定时上报封装成函数,用time.time()检查时间差:
last_report_time = 0 REPORT_INTERVAL = 30 # 秒 def should_report(): global last_report_time now = time.time() if now - last_report_time >= REPORT_INTERVAL: last_report_time = now return True return False while True: if should_report(): # 执行上报逻辑 pass # 执行其他任务 pass这种方式不会阻塞主循环,而且在处理耗时操作时更可控。对于更精确的定时,可以考虑用machine.Timer,但注意不要在中断回调里执行网络请求,因为中断上下文不适合做阻塞操作,只适合设置标志位。
5.2 请求失败的数据本地缓存与补报机制
物联网设备不能保证网络永远畅通。我的做法是在设备上把上报数据先写入本地文件,上报成功后删除;下轮上报时先补发缓存数据。
import os import ujson CACHE_FILE = '/cache_data.json' def save_to_cache(data): try: with open(CACHE_FILE, 'r') as f: cache = ujson.load(f) except: cache = [] cache.append(data) with open(CACHE_FILE, 'w') as f: ujson.dump(cache, f) def report_with_cache(url, data): if post_json_data(url, data): # 上报成功后清理缓存 try: os.remove(CACHE_FILE) except: pass return True else: save_to_cache(data) return False这个方案在断电重启后依然有效,因为数据持久化在 Flash 文件系统里。注意 Flash 写入寿命问题,频繁擦写会影响 Flash 寿命,所以缓存数据要控制大小,定时批量写入而不是每条单独写。
5.3 降低功耗:请求完成后的 WiFi 关闭策略
如果设备靠电池供电,WiFi 是耗电大户。策略是无请求时关闭 WiFi,上报时再打开。实测下来,ESP32 开启 WiFi 时电流约 100mA,关闭后深睡时电流能降到 10uA 以下,差距相当可观。
def report_and_sleep(data, sleep_seconds): wlan = network.WLAN(network.STA_IF) wlan.active(True) wlan.connect(SSID, PSK) # 等待连接后上报 post_json_data('http://...', data) # 关闭 WiFi 再深睡 wlan.active(False) machine.deepsleep(sleep_seconds * 1000)注意:深睡后重新唤醒,程序会从开头执行(类似复位),需要设计一个启动标志判断是否要恢复现场。通常用 RTC 内存保存上次的计数状态,在启动时读取。
5.4 区分局域网调试与公网生产的配置管理
开发时用局域网 IP + 无加密 HTTP,部署到公网后要切到域名 + HTTPS,这个切换如果硬编码在代码里很麻烦。我用一个简单的配置文件解决:
# config.py SERVER_HOST = '192.168.1.100' # 开发环境 SERVER_PORT = 8080 # SERVER_HOST = 'iot.example.com' # 生产环境 # SERVER_PORT = 443 API_BASE = f'http://{SERVER_HOST}:{SERVER_PORT}'把配置集中在文件顶部,切换环境时只改一处,其他代码不动。这是个很小的习惯,但能节省大量来回改代码的时间。
关于 HTTPS 再说一句:Micropython 的urequests请求 HTTPS 时底层用的是ssl.wrap_socket(),它默认不验证服务端证书,中间人攻击其实是有可能发生的。如果数据敏感,至少要在应用层做加密签名,或者用简单的 token 鉴权。
5.5 多请求并发需求:用非阻塞的方式实现
MicroPython 是多线程支持有限的(受限于 GIL 和内存)。有段时间我以为可以用_thread并发发请求,实际测下来问题很多:线程切换开销大、内存占用高、容易触发死锁。后来改为在单线程内用状态机方式管理多个请求:
class RequestScheduler: def __init__(self): self.tasks = [] def add_task(self, url, callback, interval): self.tasks.append({ 'url': url, 'callback': callback, 'interval': interval, 'last_run': 0 })这种方案的本质是把并发改为时间片轮转,一个时刻只处理一个请求,但通过合理的调度策略让多个任务看起来在并行执行。对大多数 IoT 场景够用,还能避开多线程的坑。
6. 实战案例:一个完整的上报通道
结合前面所有内容,这里给出一个完整的示例工程骨架,把 GET 拉取配置、POST 上报数据、缓存重试和状态打印整合在一起。
import network import urequests import ujson import time import gc # ============ 配置区 ============ SSID = 'your_wifi_ssid' PASSWORD = 'your_wifi_password' CONFIG_URL = 'http://192.168.1.100:8080/api/device/config' REPORT_URL = 'http://192.168.1.100:8080/api/device/report' REPORT_INTERVAL = 60 # 上报间隔(秒) # ================================ wlan = network.WLAN(network.STA_IF) def connect_wifi(): wlan.active(True) if not wlan.isconnected(): wlan.connect(SSID, PASSWORD) timeout = 10 while timeout > 0 and not wlan.isconnected(): time.sleep(1) timeout -= 1 return wlan.isconnected() def fetch_config(): try: response = urequests.get(CONFIG_URL, timeout=5) if response.status_code == 200: config = response.json() response.close() return config response.close() except Exception as e: print('获取配置失败:', e) return None def report_data(data): try: response = urequests.post(REPORT_URL, json=data, timeout=5) status = response.status_code response.close() return status in (200, 201) except Exception as e: print('上报失败:', e) return False def main(): gc.collect() if not connect_wifi(): print('WiFi 连接失败,进入低功耗模式') return print('WiFi 已连接,开始执行任务') config = fetch_config() if config: print('远程配置:', config) last_report = 0 cache = [] while True: if time.time() - last_report >= REPORT_INTERVAL: sensor_data = { 'device_id': 'esp32_demo', 'temp': 26.5, 'humidity': 58, 'ts': time.time() } if report_data(sensor_data): print('上报成功') if cache: # 补发缓存数据 pass else: cache.append(sensor_data) print('上报失败,数据已缓存') last_report = time.time() gc.collect() # 处理其他任务... time.sleep(1) if __name__ == '__main__': main()这个骨架把前面提到的模块连接、超时处理、JSON 解析、定时上报和缓存逻辑都串起来了。实际使用中,根据自己的接口调整 URL 和数据格式即可。
最后再分享一个小技巧:调试 HTTP 请求时,在电脑上跑一个简单的本地 HTTP 服务(比如用某知名 Python 框架写个十几行的临时接口),打印出所有请求头和 body。把 ESP32 的请求指向这个本地服务,就能第一时间看清设备实际发出去的内容长什么样,比在设备端反复打印日志排查效率高得多。我在调试 Content-Type 和编码问题时,这招帮了大忙。