Как прекратить работу скрипта на Python?

Question

Как прекратить работу скрипта на Python?

BrightlySlayer Ученик (142), на голосовании 3 месяца назад

import threading
import time

def mf():
global a
time.sleep(2)
if a=='1':
print('Успел')
elif a==0:
print('Не успел')
quit()
a=0
thr=threading.Thread(target=mf).start()
a=input(':')
print('основной поток')

Не пойму, почему когда создаю простой код из буквально 2х строк, то функция "quit" могла остановить код дай перед "input()", но когда я сделал код со вторым потоком то код не останавливается. Как сделать чтоб останавливался код даже с несколькими потоками, в условии где "Не успел". Я так понимаю что "quit" останавливает свой потом, да?

Answer 1

 import threading 
import time 
 
def mf(): 
    global a 
    global should_exit 
    time.sleep(2) 
    if a == '1': 
        print('Успел') 
    elif a == 0: 
        print('Не успел') 
        should_exit = True  # Устанавливаем флаг завершения 
 
should_exit = False  # Флаг завершения 
a = 0 
thr = threading.Thread(target=mf) 
thr.start() 
 
while not should_exit: 
    try: 
        a = input(':') 
        break  # Выходим из цикла, если пользователь что-то ввел 
    except EOFError: 
        # Handle Ctrl+D on Linux/macOS 
        break 
    except KeyboardInterrupt: 
        # Handle Ctrl+C 
        should_exit = True 
        break 
 
if not should_exit: 
    print('основной поток') 
 
if should_exit: 
   print("Завершение по команде из дочернего потока или Ctrl+C")

Answer 2

Использование флага остановки:
Создайте глобальную переменную-флаг, которая будет указывать на необходимость остановки потока. Поток будет периодически проверять значение этого флага и завершать свою работу, если флаг установлен.

 import threading 
import time 
 
stop_event = threading.Event()  # Флаг остановки 
 
def mf(): 
    global a 
    while not stop_event.is_set():  # Проверка флага остановки 
        time.sleep(0.1)  # Проверка с небольшой задержкой 
        if a == '1': 
            print('Успел') 
            break  # Выход из цикла, если успел 
        elif a == '0': 
            print('Не успел') 
            break # Выход из цикла, если не успел 
 
a = '0'  #Инициализируем переменную как строку. 
thr = threading.Thread(target=mf) 
thr.start() 
 
a = input(':') 
stop_event.set()  # Установка флага остановки 
thr.join()       # Ожидание завершения потока 
print('основной поток')

Использование threading.Thread.join():

 import threading 
import time 
 
def mf(): 
    time.sleep(2) 
    if a == '1': 
        print('Успел') 
    else: 
        print('Не успел') 
 
 
a = '0' 
thr = threading.Thread(target=mf) 
thr.start() 
 
a = input(':') 
thr.join()  # Ждем завершения дочернего потока 
print('основной поток')

В большинстве случаев, использование флага остановки (threading.Event) — самый чистый и надежный способ управлять завершением потоков. Он позволяет плавно остановить поток без необходимости резкого прерывания. threading.Thread.join() хорош, если вам просто нужно дождаться завершения потока.