Как завершить поток в Python?

Question

Я знаю, что topi c рассматривался несколько раз, однако я пытался или, по крайней мере, пробовал практически все решения, однако, будучи довольно новым python basher, я не смог получить ни одно из предыдущих решений

Основная предпосылка c сценария заключается в том, что он подписан на брокера MQTT и ожидает команд, команды одиночного действия работают на 100%, однако одна из команд требует al oop для работать бесконечно, пока не будет получена другая команда, поэтому наиболее подходящим решением было запустить «l oop» в отдельном потоке, в то время как основной подписчик l oop продолжает «слушать» следующую команду.

Все работает на 95%, команды "stati c" проходят и задачи выполняются нормально, затем, когда проходит команда "mtg", он выполняет действия с потоком, и l oop выполняется на 100%, однако это где он падает, когда получена следующая команда, я могу подтвердить, что оператор "if" обрабатывает команду, поскольку он печатает сообщение в c onsole, но thread.stop () не запущен, или он может быть запущен, но не завершает поток --- Я выдергиваю волосы, пытаясь понять это.

Некоторый код:

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()

Answer 1

Ваша переменная t1 является локальной, поэтому она теряется при выходе из функции on_message. Кроме того, вы объединяете класс task и экземпляр t1 (task.stop() не будет работать).

Для быстрого исправления объявите t1 = None как глобальный, затем добавьте global t1 к вашей on_message функции ...

Однако я бы подумал о рефакторинге, чтобы всегда был запущенный поток для управления Blinkt !, а обработчик сообщений MQTT просто устанавливает свое состояние соответственно - что-то вроде этот. Dry -кодирование, очевидно, так что может быть какая-то глупость.

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 BlinktManager(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.stop_event = threading.Event()
        self.mode = None

    def run(self):
        while not self.stop_event.isSet():
            self.tick()
            self.stop_event.wait(0.1)  # instead of sleep

    def tick(self):
        if self.mode == "reds":
            self._tick_reds()

    def _tick_reds(self):
        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()

    def clear(self):
        self.mode = None
        blinkt.clear()
        blinkt.show()

    def set_all_pixels(self, r, g, b):
        self.mode = None
        for x in range(blinkt.NUM_PIXELS):
            blinkt.set_pixel(x, r, g, b)
        blinkt.show()

    def set_pixel(self, x, r, g, b):
        self.mode = None
        blinkt.set_pixel(x, r, g, b)
        blinkt.show()

    def begin_reds(self):
        self.mode = "reds"


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_manager.clear()

    if command == "rgb" and len(data) == 4:
        x = data[0]
        r, g, b = [int(x) & 0xFF for x in data[1:]]
        if x == "*":
            blinkt_manager.set_all_pixels(r, g, b)
        else:
            # TODO: error handling
            blinkt_manager.set_pixel(int(x), r, g, b)

    if command == "mtg" and len(data) == 0:
        blinkt_manager.begin_reds()


blinkt.set_clear_on_exit()

blinkt_manager = BlinktManager()
blinkt_manager.start()

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

Answer 2

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()