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

MicroPython 开发ESP32应用教程之线程介绍及实例分析

news 来源：原创 2025/5/29 15:01:40

MicroPython ESP32 线程（Thread）基础

MicroPython 在 ESP32 上支持线程（Thread）功能，通过 _thread 模块实现。线程允许程序并发执行多个任务，适合处理需要同时运行的场景，例如传感器数据采集和网络通信。

线程模块导入与基本方法

_thread 模块提供线程管理的核心功能。需注意 MicroPython 的线程实现可能因硬件和固件版本不同而有所差异。

import _thread
import timedef thread_function(name):for i in range(5):print("Thread {}: Count {}".format(name, i))time.sleep(1)_thread.start_new_thread(thread_function, ("Thread-1",))

运行结果：

Thread Thread-1: Count 0
Thread Thread-1: Count 1
Thread Thread-1: Count 2
Thread Thread-1: Count 3
Thread Thread-1: Count 4

线程同步与锁机制

多线程操作共享资源时需使用锁（Lock）避免竞争条件。_thread 模块提供简单的锁实现。

lock = _thread.allocate_lock()
shared_data = 0def safe_increment():global shared_datawith lock:shared_data += 1print("Shared data:", shared_data)_thread.start_new_thread(safe_increment, ())

运行结果：Shared data: 1

ESP32 线程实例分析

以下实例展示如何在 ESP32 上使用线程处理传感器数据上传和 LED 控制：

import _thread
import time
from machine import Pinled = Pin(2, Pin.OUT)  # ESP32 板载 LED
data_lock = _thread.allocate_lock()
sensor_data = []def sensor_thread():while True:simulated_data = time.ticks_ms()  # 模拟传感器数据with data_lock:sensor_data.append(simulated_data)time.sleep(0.5)def network_thread():while True:with data_lock:if sensor_data:print("Uploading data:", sensor_data.pop(0))time.sleep(1)def led_thread():while True:led.value(not led.value())time.sleep(0.2)_thread.start_new_thread(sensor_thread, ())
_thread.start_new_thread(network_thread, ())
_thread.start_new_thread(led_thread, ())

线程注意事项

ESP32 的 MicroPython 线程存在以下限制：

线程数量受内存限制
全局解释器锁（GIL）可能导致并非真正的并行
异常处理需谨慎，未捕获的异常可能导致线程崩溃

以下代码展示如何增强线程健壮性：

def robust_thread():try:while True:# 线程主要逻辑time.sleep(1)except Exception as e:print("Thread error:", e)finally:print("Thread exiting")_thread.start_new_thread(robust_thread, ())

线程优先级与调度

MicroPython 的线程调度通常是抢占式的，但具体行为取决于底层实现。开发者无法直接设置线程优先级，需通过延时控制来协调线程执行。

import _thread
import timedef high_freq_task(name,interval):while True:# 高频任务print(name)time.sleep(interval)def low_freq_task(name,interval):while True:print(name)time.sleep(interval)_thread.start_new_thread(high_freq_task, ("High_Freq_Thread",1))
_thread.start_new_thread(low_freq_task, ("Low_Freq_Thread",5))

运行结果：
High_Freq_Thread
High_Freq_Thread
High_Freq_Thread
High_Freq_Thread
Low_Freq_Thread
High_Freq_Thread
High_Freq_Thread
High_Freq_Thread
High_Freq_Thread
High_Freq_Thread
Low_Freq_Thread
参数说明：从上面例程及运行结果，我们很容易知道，第一个参数是线程名称，后面的参数就比较灵活，甚至可以不需要任何参数。

实例分析

下面代码是我们真实项目中的部分代码：

import _thread as threadlock = thread.allocate_lock()def configwifi_task(name):if wlan.ssid and wlan.password:with lock:wlansts = wlan.check_wifi()if not wlansts :with lock:tft.showstring("正在连接到 " + wlan.ssid,Chat_X_Posi,Chat_Hint_Y_Posi,color(0,0xff,0))try:with lock:wlan.wlan.connect(wlan.ssid,wlan.password)max_wait = 10while max_wait > 0:with lock:wlansts = wlan.check_wifi()if wlansts:breakmax_wait -= 1text = "等待连接……" + "{:02d}".format(max_wait)tft.showstring(text,Chat_X_Posi,Chat_Hint_Y_Posi + 20,color(0,0xff,0))
#                    print(text)time.sleep(1)with lock:wlansts = wlan.check_wifi()tft.fill(color(0,0,0))tft.battery_status(H_Pixel - 20,0,20,12,percent)tft.show()if not wlansts:tft.showstring("连接失败",Chat_X_Posi,Chat_Hint_Y_Posi,color(0,0xff,0))except Exception as e:
#                print(e)tft.showstring("连接失败",Chat_X_Posi,Chat_Hint_Y_Posi,color(0,0xff,0))thread.start_new_thread(configwifi_task,("wificonnect",))

先简单说明一下，启动这个线程的原因。我们的项目需要用到WIFI，开始的时候，我们程序启动后直接连接WIFI，但很快发现，WIFI连接有时需要较长时间，甚至有时会接连失败，这就导致我们需要等待较长时间，才能进行后续的操作，所以我们建立了一个专门用于连接WIFI的线程。

之所以把这个线程拿出来作实例讲解，是因为我们刚开始的时候犯了个大错，我们没有添加线程锁机制，导致程序崩溃，启动直接崩溃，最要命的是，我们做这个之前没有备份，差不多3天修改添加的代码全在板上，电脑上最新的代码是3天前的。所以特意拿这部分代码出来做实例，提醒大家线程锁机制的重要性。

启用线程时，多个线程中都有访问的硬件资源一定要添加锁机制。

该线程本身的功能非常简单，就是连接WIFI，并等待10秒，连接成功则立即结束线程，否则提示连接失败结束线程。