当前位置： 首页 > news >正文

aiohttps异步HTTPS库：uPyPI+MicroPython一键安装

news 2026/4/16 19:35:07

引言

做 MicroPython 嵌入式开发的朋友，谁没被​内存溢出​、​HTTP/HTTPS 请求阻塞​、大文件传输崩掉这些破事折磨过？尤其是在 ESP32S3、树莓派 Pico W 这种资源有限的设备上，想对接云端 API、传个音频 / 图片，或者搞个 SSE 流式通信，稍不注意就 OOM（内存溢出），调试到心态爆炸。
我踩了无数坑才磨出 aiohttps 这个库 —— 它完全基于 MicroPython 原生的 asyncio、socket 和 ssl 模块打造，​零外部依赖​，专门为嵌入式设备的内存瓶颈量身定做。核心设计就两个关键点：

  • 一是​1KB 分块流式读写​，不管是下载大文件还是上传数据，内存峰值永远卡在 1KB，彻底告别 OOM；
  • 二是​异步非阻塞​，请求时每 5ms 就让出 CPU，绝不霸占资源，让设备能同时处理多任务。

更贴心的是，它直接解决了 MicroPython ure 正则解析的递归溢出问题，用纯字符串操作搞定 URL 解析；还封装了 GET/POST/PUT/DELETE 全量 HTTP 方法，流式下载、SSE 逐行读取、大 Body 分块发送这些高频场景都给你做好了。不管你是给 ESP32S3 做智能终端应用，还是在 Pico W 上对接 LLM、IoT 平台，aiohttps 都能让你少走 90% 的弯路，把精力放在业务逻辑上，而不是和底层网络死磕。

原理

说白了，aiohttps 就是一套给嵌入式设备量身定做的、低内存、非阻塞的 HTTP/HTTPS 协议实现。

MicroPython 的 asyncio 事件循环最怕阻塞：普通的 socket.recv() 是同步的，一遇到没数据的情况，就会一直卡着等，把 CPU 占死，其他协程（比如你的 uPyOS 里的 UI 刷新、传感器读取）全都会停摆。我是这么改的：

  • 先把 socket 设为​非阻塞模式​（sock.setblocking(False)）：这样 recv() 没数据的时候不会卡死，只会返回空值或者抛异常。
  • 然后套一层循环轮询：每次读不到数据，就 await asyncio.sleep_ms(5) 让出 CPU，给其他任务留时间片。
  • 这个节奏和官方的 async_websocket_client 完全对齐，哪怕你的 uPyOS 里跑着 10 个联网应用，也不会被单个请求拖垮。

嵌入式设备的内存太金贵了：Pico W 只有 100 多 KB 可用 RAM，ESP32S3 虽然强，但也架不住一次性读个 1MB 的文件进内存，直接就炸了。所以核心设计就是 **“边用边扔” 的 1KB 分块策略 **：

  • ​下载场景​：每次只从 socket 读 1KB 数据，立刻写到文件里，写完就释放这块内存。不管文件是 10KB 还是 10MB，内存里永远只有 1KB 的缓存，绝不可能因为文件太大 OOM。
  • ​上传场景​：如果是传文件，先读文件大小填到 Content-Length 里，然后每次读 1KB 文件数据，分块发送，每发一块就 sleep_ms(5) 让出 CPU，不会把 socket 缓冲区堵死；传大 JSON/base64 图片的时候，也是按 1KB 分块写，避免大 payload 把非阻塞缓冲区撑爆。

MicroPython 自带的 ure 正则引擎是递归实现的，遇到带一堆鉴权参数的长 URL（超过 200 字符），直接触发 maximum recursion depth exceeded 错误，连请求都发不出去。所以我干脆​弃用了所有正则​，用纯字符串方法解析 URL：

  • 先用 str.startswith 判断是 HTTP 还是 HTTPS 协议；
  • 用 str.find("//") 跳过协议头，再用 str.find("/") 分离 host 和 path；
  • 最后用 str.find(":") 分离 host 和 port，全程零递归、零正则，再长的 URL 都不会崩。

并且，这里我也做了一些优化：

  • 自动根据 URL 的 scheme 选协议：HTTP 直接明文连接，HTTPS 自动走 TLS 加密，不用用户手动切换。
  • TLS 握手用原生 ssl.wrap_socket，默认用 cert_reqs=ssl.CERT_NONE（不验证服务端证书），避免加载证书占内存 —— 毕竟嵌入式设备存根证书太麻烦，用户要生产环境验证的话，自己改参数就行，默认给个低内存方案。
  • 也解决了嵌入式 TLS 握手慢的问题：用 lwIP 软件 TLS 第一次握手要 2-4 秒，这个是硬件限制，我在文档里也写清楚了，避免用户误以为是库的 bug。

整个库只依赖 MicroPython 内置的 socket、ssl、asyncio 模块，没有任何第三方包。

测试

这里，大家在uPyPI（https://upypi.net/）中搜索aiohttps：

image.png

点击复制安装命令：

image.png

粘贴到终端运行下载即可，然后将下面的代码烧录到测试的单片机上：

# Python env   : MicroPython v1.23.0
# -*- coding: utf-8 -*-
# @Time    : 2026/04/15
# @Author  : leeqingsui
# @File    : main.py
# @Description : aiohttps async HTTPS client test for MicroPython on Raspberry Pi Pico 2W# ======================================== 导入相关模块 =========================================import network
import asyncio
import time
import json
import ntptime
import aiohttps# ======================================== 全局变量 ============================================WIFI_SSID     = "Y/OURSPACE"
WIFI_PASSWORD = "qc123456789"# ======================================== 功能函数 ============================================def connect_wifi():"""连接 WiFi 并返回网络对象。Returns:network.WLAN: 已连接的 WLAN 对象；连接失败时返回 None。==========================================Connect to WiFi and return the network object.Returns:network.WLAN: Connected WLAN object; None if connection fails."""# 初始化 WiFi 为 STA 模式wlan = network.WLAN(network.STA_IF)wlan.active(True)# 避免重复连接if not wlan.isconnected():print("Connecting to WiFi: {}".format(WIFI_SSID))wlan.connect(WIFI_SSID, WIFI_PASSWORD)timeout = 15while not wlan.isconnected() and timeout > 0:time.sleep(1)timeout -= 1print("Connecting... {}s remaining".format(timeout))if wlan.isconnected():print("WiFi connected, IP: {}".format(wlan.ifconfig()[0]))else:print("WiFi connection failed")return Noneelse:print("WiFi already connected")return wlandef sync_ntp():"""通过 NTP 同步系统时间。==========================================Sync system time via NTP."""for host in ("ntp.aliyun.com", "ntp.tencent.com", "pool.ntp.org"):try:ntptime.host = hostntptime.settime()t = time.gmtime()print("NTP synced via {}: {}-{:02d}-{:02d} {:02d}:{:02d}:{:02d} UTC".format(host, t[0], t[1], t[2], t[3], t[4], t[5]))returnexcept Exception as e:print("NTP failed ({}): {}".format(host, e))print("NTP sync unavailable.")async def test_aiohttps():"""aiohttps 库全量测试，使用 httpbin.org 作为公开测试服务器。Test 1: GET  -> json()         HTTPS 握手 + 全量读取 + JSON 解析Test 2: POST -> json()         str 请求体上传 + 服务端回显Test 3: GET  -> save()         流式下载写文件（4096 字节）Test 4: GET  -> status 404     非 200 状态码不抛异常Test 5: GET  -> text           resp.text 属性读取Test 6: POST -> bytes body     bytes 类型请求体Test 7: PUT  -> request()      非 GET/POST 方法Test 8: GET  -> http://        HTTP 明文连接Test 9: GET  -> iter_lines()   SSE 流式逐行读取==========================================Full test for aiohttps using httpbin.org as the public test server."""# 1. 连接 WiFiif not connect_wifi():return# 2. 同步 NTPsync_ntp()print("--- aiohttps Test Start ---")# Test 1: HTTPS GET -> json()print("[1/8] GET https://httpbin.org/get")try:resp = await aiohttps.get("https://httpbin.org/get")print("  status:", resp.status)data = await resp.json()print("  origin IP:", data.get("origin", "?"))print("  [1/8] PASS" if resp.status == 200 else "  [1/8] FAIL")except Exception as e:print("  [1/8] ERROR:", e)# Test 2: HTTPS POST str body -> json()print("[2/8] POST https://httpbin.org/post (str body)")try:body = json.dumps({"device": "pico2w", "lib": "aiohttps"})resp = await aiohttps.post("https://httpbin.org/post",headers={"Content-Type": "application/json"},data=body,)data = await resp.json()echoed = data.get("json", {})ok = echoed.get("device") == "pico2w" and echoed.get("lib") == "aiohttps"print("  [2/8] PASS" if ok else "  [2/8] FAIL (echo mismatch)")except Exception as e:print("  [2/8] ERROR:", e)# Test 3: HTTPS GET -> save() streaming downloadprint("[3/8] GET https://httpbin.org/bytes/4096 -> save test.bin")try:resp = await aiohttps.get("https://httpbin.org/bytes/4096")n = await resp.save("test.bin")print("  saved:", n, "bytes")print("  [3/8] PASS" if n == 4096 else "  [3/8] FAIL (expected 4096, got {})".format(n))except Exception as e:print("  [3/8] ERROR:", e)# Test 4: non-200 status code (404)print("[4/8] GET https://httpbin.org/status/404")try:resp = await aiohttps.get("https://httpbin.org/status/404")resp.close()print("  status:", resp.status)print("  [4/8] PASS" if resp.status == 404 else "  [4/8] FAIL (expected 404)")except Exception as e:print("  [4/8] ERROR:", e)# Test 5: resp.text propertyprint("[5/8] GET https://httpbin.org/encoding/utf8 -> text")try:resp = await aiohttps.get("https://httpbin.org/encoding/utf8")t = await resp.textprint("  text length:", len(t))print("  [5/8] PASS" if len(t) > 0 else "  [5/8] FAIL (empty text)")except Exception as e:print("  [5/8] ERROR:", e)# Test 6: POST bytes bodyprint("[6/8] POST https://httpbin.org/post (bytes body)")try:body = b"\x00\x01\x02\x03\xff"resp = await aiohttps.post("https://httpbin.org/post",headers={"Content-Type": "application/octet-stream"},data=body,)await resp.text  # 消费响应体（二进制回显不能 json()）ok = resp.status == 200print("  [6/8] PASS" if ok else "  [6/8] FAIL")except Exception as e:print("  [6/8] ERROR:", e)# Test 7: PUT method via request()print("[7/8] PUT https://httpbin.org/put")try:resp = await aiohttps.request("PUT", "https://httpbin.org/put",headers={"Content-Type": "application/json"},data=json.dumps({"action": "put_test"}),)data = await resp.json()ok = resp.status == 200 and data.get("json", {}).get("action") == "put_test"print("  [7/8] PASS" if ok else "  [7/8] FAIL")except Exception as e:print("  [7/8] ERROR:", e)# Test 8: HTTP (plain, non-HTTPS)print("[8/9] GET http://httpbin.org/get (plain HTTP)")try:resp = await aiohttps.get("http://httpbin.org/get")print("  status:", resp.status)data = await resp.json()print("  origin IP:", data.get("origin", "?"))print("  [8/9] PASS" if resp.status == 200 else "  [8/9] FAIL")except Exception as e:print("  [8/9] ERROR:", e)# Test 9: iter_lines() streaming SSEprint("[9/9] GET https://httpbin.org/stream/3 -> iter_lines()")try:resp = await aiohttps.get("https://httpbin.org/stream/3")print("  status:", resp.status)count = 0async for line in resp.iter_lines():line = line.strip()if not line:# 跳过空行（HTTP 分隔行）continuecount += 1print("  line {}: {} bytes".format(count, len(line)))print("  [9/9] PASS" if count >= 3 else "  [9/9] FAIL (expected >=3 lines, got {})".format(count))except Exception as e:print("  [9/9] ERROR:", e)print("--- aiohttps Test Done ---")# ======================================== 自定义类 ============================================# ======================================== 初始化配置 ===========================================time.sleep(3)
print("FreakStudio: aiohttps async HTTPS client test")# ========================================  主程序  ===========================================if __name__ == "__main__":asyncio.run(test_aiohttps())

上面案例中，在树莓派 Pico 2W 硬件平台上完成 aiohttps 异步 HTTPS 客户端全功能验证测试，程序内置 WiFi 自动连接、多源 NTP 时间同步功能，并通过 9 项标准化测试用例，全面验证了库的核心能力：支持 HTTPS/HTTP 双协议、GET/POST/PUT 全 HTTP 方法，可实现 JSON / 二进制数据收发、大文件流式下载、404 等非 200 状态码兼容、SSE 流式逐行读取等高频场景，全程异步非阻塞运行、内存占用极低，无任何第三方依赖。

eb6b459ccb3f99726a2fd06d98170352.png

e56a916b375ed771aab3187baee81773.png

http://www.jsqmd.com/news/651628/

相关文章：

  • 搭建知睿 STM32MP135 的交叉编译环境
  • 智能驾驶ISP优化：低延迟与高保真图像处理的架构设计
  • 2026广西学历提升机构对比评测：5大热门机构全方位横评，谁更值得托付？ - 商业科技观察
  • 从ENIAC到物联网：用5个生活案例讲透信息技术发展史（教资考点速记版）
  • Scrcpy-iOS终极指南：3步实现iOS设备无线控制Android手机
  • 别再死磕随机种子了！聊聊IC验证中那些被忽略的覆盖率提升技巧（附VCS -cm_hier实战）
  • 天梯赛L3部分
  • 3步搞定《经济研究》期刊论文排版：Chinese-ERJ LaTeX模板终极指南
  • 创维E900V21E盒子有线网卡终极解决方案：深入剖析S905L2芯片Armbian兼容性难题
  • 3大核心功能：Arduino IDE如何让你轻松调试嵌入式项目？
  • QOJ5017 相等树链
  • FPGA新手必看：手把手教你用Verilog实现SPI主从通信（附完整代码与仿真波形）
  • 树图中的层次分解与结构优化
  • 怎么修改jpg创建时间和日期？6个实操方法，新手秒上手
  • AI建站工具选型指南：五个标准帮你找到真正靠谱的智能建站方案
  • FPGA开发环境搭建实战：从零部署Quartus Prime 20与ModelSim SE 10
  • 2025终极指南：如何在Apple Silicon Mac上使用PlayCover畅玩iOS游戏
  • 关于Cruise混动仿真模型及P2并联混动仿真模型的详细介绍
  • 基于二阶自抗扰ADRC的车辆轨迹跟踪控制：抗干扰性仿真研究及复现资料
  • 5分钟掌握RePKG：Wallpaper Engine资源提取与转换完整指南
  • 算法训练营第三天| 滑动窗口算法
  • 3步解决OneNote数据孤岛：OneNote Md Exporter迁移最佳实践
  • 3步轻松搞定：DS4Windows终极PS手柄PC兼容指南
  • 一个问题，GPT-6是否值得期待？？？
  • WebPShop插件终极指南：在Photoshop中完美处理WebP格式图像
  • lanqiao498 回文日期
  • 嵌入式开发避坑：RTL8211 PHY芯片在U-Boot中识别但Ping不通的排查与修复实录
  • 从卫星信号到你的位置：用MATLAB拆解GNSS软件接收机核心算法链
  • 跨平台文本复制实战：从网页到微信小程序的实现技巧
  • 通达信大资金流向监测公式实战解析