Python-如何实现“可停止"线程? [英] Python - How can I implement a 'stoppable' thread?

问题描述

此处创建一个可停止的线程.但是，我在理解如何实施此解决方案时遇到了一些问题.

There is a solution posted here to create a stoppable thread. However, I am having some problems understanding how to implement this solution.

使用代码...

import threading

class StoppableThread(threading.Thread):
    """Thread class with a stop() method. The thread itself has to check
    regularly for the stopped() condition."""

    def __init__(self):
        super(StoppableThread, self).__init__()
        self._stop_event = threading.Event()

    def stop(self):
        self._stop_event.set()

    def stopped(self):
        return self._stop_event.is_set()

如何创建一个线程，该线程运行每1秒钟将"Hello"打印到终端的函数. 5秒后，我使用.stop()停止循环功能/线程.

How can I create a thread that runs a function that prints "Hello" to the terminal every 1 second. After 5 seconds I use the .stop() to stop the looping function/thread.

再次，我在理解如何实施此停止解决方案时遇到了麻烦，这是我到目前为止所掌握的.

Again I am having troubles understanding how to implement this stopping solution, here is what I have so far.

import threading
import time

class StoppableThread(threading.Thread):
    """Thread class with a stop() method. The thread itself has to check
    regularly for the stopped() condition."""

    def __init__(self):
        super(StoppableThread, self).__init__()
        self._stop_event = threading.Event()

    def stop(self):
        self._stop_event.set()

    def stopped(self):
        return self._stop_event.is_set()

def funct():
    while not testthread.stopped():
        time.sleep(1)
        print("Hello")

testthread = StoppableThread()
testthread.start()
time.sleep(5)
testthread.stop()

上面的

代码创建线程testthread，可以通过testthread.stop()命令将其停止.据我了解，这只是创建一个空线程...有没有一种方法可以创建一个运行funct()的线程，并且当我使用.stop()时该线程将结束.基本上，我不知道如何实现StoppableThread类来将funct()函数作为线程运行.

Code above creates the thread testthread which can be stopped by the testthread.stop() command. From what I understand this is just creating an empty thread... Is there a way I can create a thread that runs funct() and the thread will end when I use .stop(). Basically I do not know how to implement the StoppableThread class to run the funct() function as a thread.

常规线程函数的示例...

Example of a regular threaded function...

import threading
import time

def example():
    x = 0
    while x < 5:
        time.sleep(1)
        print("Hello")
        x = x + 1

t = threading.Thread(target=example)
t.start()
t.join()
#example of a regular threaded function.

推荐答案

在原始示例中，如何使用代码有两个问题.首先，您没有将任何构造函数参数传递给基本构造函数.这是一个问题，因为如您在纯线程示例中所看到的，构造函数参数通常是必需的.您应按以下方式重写StoppableThread.__init__:

There are a couple of problems with how you are using the code in your original example. First of all, you are not passing any constructor arguments to the base constructor. This is a problem because, as you can see in the plain-Thread example, constructor arguments are often necessary. You should rewrite StoppableThread.__init__ as follows:

def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self._stop_event = threading.Event()

由于使用的是Python 3，因此不需要为super提供参数.现在你可以做

Since you are using Python 3, you do not need to provide arguments to super. Now you can do

testthread = StoppableThread(target=funct)

这仍然不是最佳解决方案，因为funct使用外部变量testthread来停止自身.尽管对于像您这样的小例子来说这可以接受，但使用这样的全局变量通常会导致巨大的维护负担，而您却不想这样做.更好的解决方案是为特定任务扩展通用StoppableThread类，以便您可以正确访问self:

This is still not an optimal solution, because funct uses an external variable, testthread to stop itself. While this is OK-ish for a tiny example like yours, using global variables like that normally causes a huge maintenance burden and you don't want to do it. A much better solution would be to extend the generic StoppableThread class for your particular task, so you can access self properly:

class MyTask(StoppableThread):
    def run(self):
        while not self.stopped():
            time.sleep(1)
            print("Hello")

testthread = MyTask()
testthread.start()
time.sleep(5)
testthread.stop()

如果您绝对不想扩展StoppableThread，则可以使用

If you absolutely do not want to extend StoppableThread, you can use the current_thread function in your task in preference to reading a global variable:

def funct():
    while not current_thread().stopped():
        time.sleep(1)
        print("Hello")

testthread = StoppableThread(target=funct)
testthread.start()
sleep(5)
testthread.stop()

这篇关于Python-如何实现“可停止"线程?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！