

上次看到有个社区提问我找不到了,就是问如何在主流程中强制结束无限循环的并发子流程。一直想这个问题来的,但一直没有找到答案。
我们知道并发子流程调用后会返回一个thread_instance,即一个线程实例。

翻阅了下python中线程相关,想粗暴地通过thread.kill()杀死线程,提示:没有kill属性或方法。其他杀死线程的方法需要在子流程中做对应的处理,这在影刀的并发调用子流程指令中无法实现。

今天偶然将thread_instance打印出来看了下,发现里面有串数字,很显然,那这个就是线程id喽。
下面为打印出来的实例,其中最后7404为线程id。
<Thread(Thread-2 (run), started daemon 7404)>上魔法指令:先从这个线程实例中截取线程id出来, 然后直接强制杀死线程id。
测试结果:
使用给出的线程id强制杀死线程(魔法指令上的代码)
# 使用此指令前,请确保安装必要的Python库,例如使用以下命令安装:
# pip install psutil
import ctypes
import sys
import psutil
from typing import *
try:
from xbot.app.logging import trace as print
except:
from xbot import print
def kill_thread_by_id(thread_id):
"""
title: 根据线程ID强制杀死线程
description: 根据给定的线程ID % thread_id %,强制终止该线程(仅支持Windows系统)。
inputs:
- thread_id (int): 要杀死的线程ID,eg: "5678"
outputs:
None
"""
if not isinstance(thread_id, int) or thread_id <= 0:
raise ValueError("线程ID必须为正整数")
if sys.platform != 'win32':
raise OSError("此功能仅支持Windows操作系统")
def _kill_thread_windows(tid):
"""
在Windows系统中强制杀死指定线程
"""
try:
# 获取Windows API函数
kernel32 = ctypes.windll.kernel32
# 打开线程句柄
THREAD_TERMINATE = 0x0001
thread_handle = kernel32.OpenThread(THREAD_TERMINATE, False, tid)
if thread_handle == 0:
error_code = kernel32.GetLastError()
if error_code == 87: # ERROR_INVALID_PARAMETER
raise ValueError(f"线程ID {tid} 不存在或已经结束")
elif error_code == 5: # ERROR_ACCESS_DENIED
raise PermissionError(f"没有权限终止线程ID {tid},请以管理员权限运行")
else:
raise OSError(f"无法打开线程 {tid},错误代码: {error_code}")
# 终止线程
result = kernel32.TerminateThread(thread_handle, 0)
if result == 0:
error_code = kernel32.GetLastError()
kernel32.CloseHandle(thread_handle)
raise OSError(f"终止线程失败,错误代码: {error_code}")
# 关闭线程句柄
kernel32.CloseHandle(thread_handle)
print(f"线程ID {tid} 已成功终止")
print("警告: 强制终止线程可能导致程序不稳定,建议重启相关应用程序")
except Exception as e:
raise RuntimeError(f"终止线程时发生错误: {str(e)}")
_kill_thread_windows(thread_id)
附测试代码:
A。子流程24
B。子流程25(为并发调用的子流程,此处没有用无限循环)
C。子流程30
注意:直接强制杀死进程中某一个线程将可能带来无法预料的风险!
Enjoy!