python - How to terminate a thread in Python?

Question

I know the topic has been covered a few times however I have attempted, or at least tried virtually all solutions however being a fairly new python basher I've not been able to get any of the previous solutions to work.

The basic premise of the script is that its subscribed to a MQTT broker and waiting for commands, the single action commands work 100%, however one of the commands required a loop to run indefinitely until another command is received, thus the most appropriate solution was to run the "loop" in a separate thread while the main subscriber loop continues to "listen" for the next command.

Everything is working 95%, the "static" commands come through and the tasks is run fine, then when the "mtg" command comes through it actions the thread and the loop runs 100%, however this is where it falls down, when the next command is received I can confirm the "if" statement processes the command as it prints the message to the console, but the thread.stop() is not run, or it may be run but it does not terminate the thread --- I'm pulling my hair out trying to figure it out.

Some code:

from sys import exit
import blinkt
import threading
import time

MQTT_SERVER = '192.168.x.x'
MQTT_PORT = 1883
MQTT_TOPIC = 'mytopic'

REDS = [0, 0, 0, 0, 0, 16, 64, 255, 64, 16, 0, 0, 0, 0, 0, 0]

start_time = time.time()

class task(threading.Thread):

     def __init__(self):
         threading.Thread.__init__(self)
         self.kill = threading.Event()
         self.event = threading.Event()
         self._stop = threading.Event()

     def run(self):
#        while not self.kill.is_set(): 
         while True:
             if self.stopped():
                return
             self.start_run()

     def stop(self):
#        self.event.set()
         self._stop.set()

     def stopped(self):
         return self._stop.isSet()

     def start_run(self):
#         while True: <-- no longer needed as the run method loops the process. 
             delta = (time.time() - start_time) * 16
             offset = int(abs((delta % len(REDS)) - blinkt.NUM_PIXELS))

             for i in range(blinkt.NUM_PIXELS):
                 blinkt.set_pixel(i, REDS[offset + i], 0, 0)

             blinkt.show()
             time.sleep(0.1)

def on_connect(client, userdata, flags, rc):
    print('Connected with result code ' + str(rc))
    client.subscribe(MQTT_TOPIC)

def on_message(client, userdata, msg):

    data = msg.payload
    if type(data) is bytes:
        data = data.decode('utf-8')
    data = data.split(',')
    command = data.pop(0)

    if command == 'clr' and len(data) == 0:
        blinkt.clear()
        blinkt.show()
        t1.stop()      #<--- I've tried a few ways to get the task to stop when the "clr" command is recieved
        task.stop()
        return

    if command == 'rgb' and len(data) == 4: #<-- This code block works fine, msg arrives and LEDs are set correctly
        try:
            pixel = data.pop(0)

            if pixel == '*':
                pixel = None
            else:
                pixel = int(pixel)
                if pixel > 7:
                    print('Pixel out of range: ' + str(pixel))
                    return

            r, g, b = [int(x) & 0xff for x in data]

            print(command, pixel, r, g, b)

        except ValueError:
            print('Malformed command: ' + str(msg.payload))
            return
        if pixel is None:
            for x in range(blinkt.NUM_PIXELS):
                blinkt.set_pixel(x, r, g, b)
        else:
            blinkt.set_pixel(pixel, r, g, b)
        blinkt.show()
        return


    if command == 'mtg' and len(data) == 0:
        print(command)
        t1 = task()
        t1.start()   #<-- Here is where the Thread is called to start and seems to run ok
        return

blinkt.set_clear_on_exit()

client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect(MQTT_SERVER, MQTT_PORT, 60)
client.loop_forever()

标签： pythonmultithreadingmqttpaho

Answer 1

Python程序在python线程中引发异常

import threading 
import ctypes 
import time 

class thread_with_exception(threading.Thread):

    def __init__(self, name): 
    threading.Thread.__init__(self) 
    self.name = name 

def run(self): 

    # target function of the thread class 
    try: 
        while True: 
            print('running ' + self.name) 
    finally: 
        print('ended') 

def get_id(self): 

    # returns id of the respective thread 
    if hasattr(self, '_thread_id'): 
        return self._thread_id 
    for id, thread in threading._active.items(): 
        if thread is self: 
            return id

def raise_exception(self): 
    thread_id = self.get_id() 
    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') 

t1 = thread_with_exception('Thread 1') 
t1.start() 
time.sleep(2) 
t1.raise_exception() 
t1.join()