Ngắt một luồng trong Python là một yêu cầu phổ biến trong lập trình đa luồng, nơi việc thực thi của một luồng cần được kết thúc dưới những điều kiện nhất định. Trong một chương trình đa luồng, một tác vụ trong một luồng mới có thể cần được dừng lại. Điều này có thể do nhiều lý do, chẳng hạn như: hoàn thành tác vụ, tắt ứng dụng hoặc các điều kiện bên ngoài khác.
Trong Python, việc ngắt các luồng có thể được thực hiện bằng cách sử dụng threading.Event hoặc bằng cách thiết lập cờ kết thúc bên trong chính luồng đó. Những phương pháp này cho phép bạn ngắt các luồng một cách hiệu quả, đảm bảo rằng tài nguyên được giải phóng đúng cách và các luồng thoát một cách sạch sẽ.
Một trong những cách đơn giản để ngắt một luồng là sử dụng lớp threading.Event . Lớp này cho phép một luồng thông báo cho luồng khác rằng một sự kiện cụ thể đã xảy ra. Dưới đây là cách bạn có thể triển khai việc ngắt luồng bằng cách sử dụng threading.Event.
Trong ví dụ này, chúng ta có một lớp MyThread. Đối tượng của nó bắt đầu thực thi phương thức run(). Luồng chính sẽ ngủ trong một khoảng thời gian nhất định và sau đó thiết lập một sự kiện. Cho đến khi sự kiện được phát hiện, vòng lặp trong phương thức run() sẽ tiếp tục. Ngay khi sự kiện được phát hiện, vòng lặp sẽ kết thúc.
from time import sleep
from threading import Thread
from threading import Event
class MyThread(Thread):
def __init__(self, event):
super(MyThread, self).__init__()
self.event = event
def run(self):
i=0
while True:
i+=1
print ('Child thread running...',i)
sleep(0.5)
if self.event.is_set():
break
print()
print('Child Thread Interrupted')
event = Event()
thread1 = MyThread(event)
thread1.start()
sleep(3)
print('Main thread stopping child thread')
event.set()
thread1.join()
Khi bạn thực thi đoạn mã này, nó sẽ tạo ra output −
Child thread running... 1 Child thread running... 2 Child thread running... 3 Child thread running... 4 Child thread running... 5 Child thread running... 6 Main thread stopping child thread Child Thread Interrupted
Một cách tiếp cận khác để ngắt các luồng là sử dụng một flag mà luồng kiểm tra định kỳ. Phương pháp này liên quan đến việc đặt một thuộc tính cờ trong đối tượng luồng và thường xuyên kiểm tra giá trị của nó trong vòng lặp thực thi của luồng.
Ví dụ này minh họa cách sử dụng một cờ để điều khiển và dừng một luồng đang chạy trong chương trình đa luồng Python.
import threading
import time
def foo():
t = threading.current_thread()
while getattr(t, "do_run", True):
print("working on a task")
time.sleep(1)
print("Stopping the Thread after some time.")
# Create a thread
t = threading.Thread(target=foo)
t.start()
# Allow the thread to run for 5 seconds
time.sleep(5)
# Set the termination flag to stop the thread
t.do_run = False
Khi bạn thực thi đoạn mã này, nó sẽ tạo ra output −
working on a task working on a task working on a task working on a task working on a task Stopping the Thread after some time.