iOSAI/Utils/ThreadManager.py

import ctypes
import threading
from typing import Dict, Tuple, List

from Utils.LogManager import LogManager


def _async_raise(tid: int, exc_type=KeyboardInterrupt):
    """向指定线程抛异常，强制跳出"""
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid), ctypes.py_object(exc_type))
    if res == 0:
        raise ValueError("线程不存在")
    elif res > 1:
        ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid), 0)

class ThreadManager:
    _tasks: Dict[str, Dict] = {}
    _lock = threading.Lock()

    @classmethod
    def add(cls, udid: str, thread: threading.Thread, event: threading.Event) -> Tuple[int, str]:
        LogManager.method_info(f"准备创建任务:{udid}", "task")
        with cls._lock:
            # 判断当前设备是否有任务
            if cls._tasks.get(udid, None) is not None:
                return 1001, "当前设备已存在任务"
            thread.start()
            print(thread.ident)

            cls._tasks[udid] = {
                "id": thread.ident,
                "thread": thread,
                "event": event
            }
        return 200, "创建成功"

    @classmethod
    def stop(cls, udid: str) -> Tuple[int, str]:
        try:
            print(cls._tasks)
            obj = cls._tasks.get(udid, None)
            obj["event"].set()
            r = cls._kill_thread(obj.get("id"))
            if r:
                cls._tasks.pop(udid, None)
            else:
                print("好像有问题")
        except Exception as e:
            print(e)
        return 200, "操作成功"

    @classmethod
    def batch_stop(cls, ids: List[str]) -> Tuple[int, str]:
        try:
            for udid in ids:
                cls.stop(udid)
        except Exception as e:
            print(e)
        return 200, "停止成功."

    @classmethod
    def _kill_thread(cls, tid: int) -> bool:
        """向原生线程 ID 抛 KeyboardInterrupt，强制跳出"""
        res = ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid),
                                                         ctypes.py_object(KeyboardInterrupt))
        if res == 0:  # 线程已不存在
            print("线程不存在")
            return False
        if res > 1:  # 命中多个线程，重置
            print("命中了多个线程")
            ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid), 0)
        print("杀死线程成功")
        return True