如果不杀死线程会更好。 一种方法是在线程的循环中引入一个“try”块，并在想要停止线程时抛出一个异常(例如break/return/…)这会让你停下来/一会儿/…)。 我在我的应用程序上使用了这个，它是有效的…

这里还有另一种方法，但代码非常干净和简单，适用于2021年的Python 3.7:

import ctypes 

def kill_thread(thread):
    """
    thread: a threading.Thread object
    """
    thread_id = thread.ident
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id, ctypes.py_object(SystemExit))
    if res > 1:
        ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id, 0)
        print('Exception raise failure')

改编自:https://www.geeksforgeeks.org/python-different-ways-to-kill-a-thread/

正如@Kozyarchuk的回答中提到的，安装跟踪工作。由于这个答案不包含代码，下面是一个工作就绪的示例:

import sys, threading, time 

class TraceThread(threading.Thread): 
    def __init__(self, *args, **keywords): 
        threading.Thread.__init__(self, *args, **keywords) 
        self.killed = False
    def start(self): 
        self._run = self.run 
        self.run = self.settrace_and_run
        threading.Thread.start(self) 
    def settrace_and_run(self): 
        sys.settrace(self.globaltrace) 
        self._run()
    def globaltrace(self, frame, event, arg): 
        return self.localtrace if event == 'call' else None
    def localtrace(self, frame, event, arg): 
        if self.killed and event == 'line': 
            raise SystemExit() 
        return self.localtrace 

def f(): 
    while True: 
        print('1') 
        time.sleep(2)
        print('2') 
        time.sleep(2)
        print('3') 
        time.sleep(2)

t = TraceThread(target=f) 
t.start() 
time.sleep(2.5) 
t.killed = True

它在输出1和2之后停止。3不打印。

只是建立在@SCB的想法(这正是我所需要的)，创建一个KillableThread子类与自定义函数:

from threading import Thread, Event

class KillableThread(Thread):
    def __init__(self, sleep_interval=1, target=None, name=None, args=(), kwargs={}):
        super().__init__(None, target, name, args, kwargs)
        self._kill = Event()
        self._interval = sleep_interval
        print(self._target)

    def run(self):
        while True:
            # Call custom function with arguments
            self._target(*self._args)

            # If no kill signal is set, sleep for the interval,
            # If kill signal comes in while sleeping, immediately
            #  wake up and handle
            is_killed = self._kill.wait(self._interval)
            if is_killed:
                break

        print("Killing Thread")

    def kill(self):
        self._kill.set()

if __name__ == '__main__':

    def print_msg(msg):
        print(msg)

    t = KillableThread(10, print_msg, args=("hello world"))
    t.start()
    time.sleep(6)
    print("About to kill thread")
    t.kill()

自然地，就像@SBC一样，线程不会等待运行一个新的循环来停止。在这个例子中，你会看到“kill Thread”消息紧跟在“About to kill Thread”之后，而不是等待4秒钟线程完成(因为我们已经睡了6秒了)。

KillableThread构造函数中的第二个参数是您的自定义函数(print_msg)。Args参数是在调用函数(("hello world"))时使用的参数。

你不应该在没有与线程合作的情况下强行终止线程。

杀死一个线程消除了try/finally阻塞设置的任何保证，所以你可能会让锁锁定，文件打开等等。

唯一可以认为强制终止线程是一个好主意的情况是快速终止程序，但绝不是单个线程。

您可以在进程中执行命令，然后使用进程id终止它。 我需要在两个线程之间同步，其中一个线程本身不返回。

processIds = []

def executeRecord(command):
    print(command)

    process = subprocess.Popen(command, stdout=subprocess.PIPE)
    processIds.append(process.pid)
    print(processIds[0])

    #Command that doesn't return by itself
    process.stdout.read().decode("utf-8")
    return;


def recordThread(command, timeOut):

    thread = Thread(target=executeRecord, args=(command,))
    thread.start()
    thread.join(timeOut)

    os.kill(processIds.pop(), signal.SIGINT)

    return;