micropython _thread 库 - 代码无法访问错误答案

【问题标题】：micropython _thread library - Code is unreachable errormicropython _thread 库 - 代码无法访问错误
【发布时间】：2022-12-03 05:16:26
【问题描述】：

我正在使用 micropython 的 _thread 库，在一个类中，我正在编写代码来处理线程中的任务，当我试图了解线程是否正在工作时，我收到了错误代码无法访问错误

这是我的代码：

import _thread
from time import sleep

def task_0():
    print('c')
    sleep(0.5)
    print('o')
    sleep(0.5)
    print('r')
    sleep(0.5)
    print('e')
    sleep(0.5)
    print('0')
    sleep(0.5)

def task_1():
    print('C')
    sleep(0.5)
    print('O')
    sleep(0.5)
    print('R')
    sleep(0.5)
    print('E')
    sleep(0.5)
    print('1')
    sleep(0.5)

class ThreadWorker:
    def __init__(self):
        self.threadId = _thread.get_ident()
        self.lock = _thread.allocate_lock()
        self.queue = []

    def add(self, task):
        self.queue.append(task())

    def status(self):
        if _thread.get_ident() is None:
            print('Not thread working')
        else:
            print('Thread ID: {}'.format(_thread.get_ident()))

    def start(self):
        while len(self.queue) > 0:
            fn = self.queue.pop(0)
            if fn is None:
                break

            self.lock.acquire()
            print("Queue: {}\n Next Task: {}".format(self.queue, fn))
            self.threadId = _thread.start_new_thread(fn, ())
            self.lock.release()

    def stop(self):
        print('Trying to stop Thread with ID: {}'.format(self.threadId))
        _thread.exit()


thread = ThreadWorker()
thread.start()
thread.add(task_0)
thread.add(task_1)
thread.stop()
thread.status() <--- (method) status: () -> None | Code is unreachablePylance

我试图在所有任务完成后停止线程并显示胎面信息。

【问题讨论】：

标签： python multithreading oop micropython

【解决方案1】：

看起来 status 方法是在 stop 方法之后调用的，这导致了“Code is unreachable”错误。这是因为stop方法退出线程，之后的代码永远不会执行。

要修复此错误，您可以在调用 stop 方法之前移动对 status 方法的调用。这样，status方法将在线程停止之前被调用。

这是具有此更改的代码的更新版本：

import _thread
from time import sleep

def task_0():
    print('c')
    sleep(0.5)
    print('o')
    sleep(0.5)
    print('r')
    sleep(0.5)
    print('e')
    sleep(0.5)
    print('0')
    sleep(0.5)

def task_1():
    print('C')
    sleep(0.5)
    print('O')
    sleep(0.5)
    print('R')
    sleep(0.5)
    print('E')
    sleep(0.5)
    print('1')
    sleep(0.5)

class ThreadWorker:
    def __init__(self):
        self.threadId = _thread.get_ident()
        self.lock = _thread.allocate_lock()
        self.queue = []

    def add(self, task):
        self.queue.append(task())

    def status(self):
        if _thread.get_ident() is None:
            print('Not thread working')
        else:
            print('Thread ID: {}'.format(_thread.get_ident()))

    def start(self):
        while len(self.queue) > 0:
            fn = self.queue.pop(0)
            if fn is None:
                break

            self.lock.acquire()
            print("Queue: {}
 Next Task: {}".format(self.queue, fn))
            self.threadId = _thread.start_new_thread(fn, ())
            self.lock.release()

    def stop(self):
        print('Trying to stop Thread with ID: {}'.format(self.threadId))
        _thread.exit()


thread = ThreadWorker()
thread.start()
thread.add(task_0)
thread.add(task_1)
thread.status()  # <--- moved this line here
thread.stop()

希望这可以帮助！

【讨论】：