robots-auto-action/tools/thread_pool.py

from concurrent.futures import ThreadPoolExecutor


# 创建公用线程池
class ThreadPoolSingleton:
    _instance = None
    _executor = None

    # 获取公用线程池类方法
    # max_workers 默认的工作线程数为2
    # thread_name_prefix 默认的工作线程名称前缀为  major-thread-pool-
    # queue_size 默认的等待队列大小为 不限制
    # reject_handler 默认的拒绝策略为空
    @classmethod
    def get_executor(cls, max_workers=2, thread_name_prefix='major-thread-pool-'):
        if cls._executor is None:
            cls._executor = ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix=thread_name_prefix)
        return cls._executor

    def __new__(cls):
        if not cls._instance:
            cls._instance = super().__new__(cls)
        return cls._instance