• 创建一个Event对象，默认为未触发状态
• 在线程函数中循环检查该事件是否被设置（event.is_set()）
• 当需要取消任务时，调用event.set()

import threading
import time
<p>def worker(stop_event):
while not stop_event.is_set():
print("任务正在运行...")</p><h1>模拟耗时操作（可拆分为小段）</h1><pre class='brush:python;toolbar:false;'>    for _ in range(10):
        if stop_event.is_set():
            print("收到取消信号，退出任务")
            return
        print(".", end="", flush=True)
        time.sleep(0.5)  # 每半秒检查一次
    print("\n一轮完成")

主程序

stop_event = threading.Event() thread = threading.Thread(target=worker, args=(stop_event,)) thread.start()

time.sleep(3) # 运行3秒后取消 stop_event.set() # 发送取消信号 thread.join() # 等待线程结束 print("主线程：任务已取消")

import threading
<p>class Task:
def <strong>init</strong>(self):
self.cancelled = False</p><pre class='brush:python;toolbar:false;'>def run(self):
    while not self.cancelled:
        if self.cancelled:
            break
        print("执行中...")
        time.sleep(1)
    print("任务已取消")

task = Task() t = threading.Thread(target=task.run) t.start()

time.sleep(3) task.cancelled = True t.join()

from queue import Queue
import threading
<p>def worker(q, stop_event):
while not stop_event.is_set():
try:
item = q.get(timeout=0.5)  # 设置超时以便定期检查事件
if item is None:
break  # 收到终止信号
print(f"处理 {item}")
time.sleep(1)
q.task_done()
except:
continue</p>