引言

做 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（MicroPython Package Repository）中搜索aiohttps：

点击复制安装命令：

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 = 15
  

  

   
   while not wlan.isconnected() and timeout > 0:
  

  

   
   time.sleep(1)
  

  

   
   timeout -= 1
  

  

   
   print("Connecting... {}s remaining".format(timeout))
  

  

   
   
  

  

   
   if wlan.isconnected():
  

  

   
   print("WiFi connected, IP: {}".format(wlan.ifconfig()[0]))
  

  

   
   else:
  

  

   
   print("WiFi connection failed")
  

  

   
   return None
  

  

   
   else:
  

  

   
   print("WiFi already connected")
  

  

   
   
  

  

   
   return wlan
  

  

   
   
  

  

   
   
  

  

   
   def sync_ntp():
  

  

   
   """
  

  

   
   通过 NTP 同步系统时间。
  

  

   
   
  

  

   
   ==========================================
  

  

   
   
  

  

   
   Sync system time via NTP.
  

  

   
   """
  

  

   
   for host in ("ntp.aliyun.com", "ntp.tencent.com", "pool.ntp.org"):
  

  

   
   try:
  

  

   
   ntptime.host = host
  

  

   
   ntptime.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]))
  

  

   
   return
  

  

   
   except 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. 连接 WiFi
  

  

   
   if not connect_wifi():
  

  

   
   return
  

  

   
   
  

  

   
   # 2. 同步 NTP
  

  

   
   sync_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 download
  

  

   
   print("[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 property
  

  

   
   print("[5/8] GET https://httpbin.org/encoding/utf8 -> text")
  

  

   
   try:
  

  

   
   resp = await aiohttps.get("https://httpbin.org/encoding/utf8")
  

  

   
   t = await resp.text
  

  

   
   print(" 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 body
  

  

   
   print("[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 == 200
  

  

   
   print(" [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 SSE
  

  

   
   print("[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 = 0
  

  

   
   async for line in resp.iter_lines():
  

  

   
   line = line.strip()
  

  

   
   if not line:
  

  

   
   # 跳过空行（HTTP 分隔行）
  

  

   
   continue
  

  

   
   count += 1
  

  

   
   print(" 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")
  

  

   
   
  

  

   
   # ======================================== 主程序 ===========================================
  

  

   
   
  

 

 
  

        # https
      
        # 网络协议
      
        # http
      
  
 
 Logo 
AI硬件创业社区
 
智能硬件社区聚焦AI智能硬件技术生态，汇聚嵌入式AI、物联网硬件开发者，打造交流分享平台，同步全国赛事资讯、开展 OPC 核心人才招募，助力技术落地与开发者成长。
 
 
  
更多推荐
  
 
【SpringBoot集成Redis】Spring Boot 集成 Redis 完整指南：从自动配置原理到序列化乱码彻底剖析
 
本文深入解析了SpringBoot集成Redis的核心机制，主要包含五个关键点：1）SpringBoot通过类路径扫描和条件注解实现Redis自动配置；2）生产环境应避免使用默认JDK序列化（存在可读性差、存储膨胀等问题），推荐JSON序列化方案；3）Redis默认序列化的"乱码"实质是Java序列化协议头；4）Redis基于TCP文本协议（RESP）通信，支持telnet等工
 
avatar AI硬件创业社区
 
cover
 
OBD-II/UDS 渗透测试（三）：安全机制 — NRC 情报解读、Session 控制与 SecAccess 破解
 
 
avatar AI硬件创业社区
 
cover
 
STM32F103C8 + FreeRTOS + ESP32 学习记录（二）：ESP32连接WiFi热点获取时间数据
 
 
avatar AI硬件创业社区