一個有界任務隊列的python thradpoolexcutor， 直接捕獲錯誤日誌

基於官方的須要改版

一、改成有界，官方是吧全部任務添加到線程池的queue隊列中，這樣內存會變大，也不符合分佈式的邏輯(會把中間件的全部任務一次性取完，放到本地的queue隊列中，致使分佈式變差)

二、直接打印錯誤。官方的threadpolexcutor執行的函數，若是不設置回調，即便函數中出錯了，本身都不會知道。

 

# coding=utf-8
"""
一個有界任務隊列的thradpoolexcutor
直接捕獲錯誤日誌
"""
from functools import wraps
import queue
from concurrent.futures import ThreadPoolExecutor, Future
# noinspection PyProtectedMember
from concurrent.futures.thread import _WorkItem
from app.utils_ydf import LoggerMixin, LogManager

logger = LogManager('BoundedThreadPoolExecutor').get_logger_and_add_handlers()


def _deco(f):
    @wraps(f)
    def __deco(*args, **kwargs):
        try:
            return f(*args, **kwargs)
        except Exception as e:
            logger.exception(e)

    return __deco


class BoundedThreadPoolExecutor(ThreadPoolExecutor, ):
    def __init__(self, max_workers=None, thread_name_prefix=''):
        ThreadPoolExecutor.__init__(self, max_workers, thread_name_prefix)
        self._work_queue = queue.Queue(max_workers * 2)

    def submit(self, fn, *args, **kwargs):
        with self._shutdown_lock:
            if self._shutdown:
                raise RuntimeError('cannot schedule new futures after shutdown')
            f = Future()
            fn_deco = _deco(fn)
            w = _WorkItem(f, fn_deco, args, kwargs)
            self._work_queue.put(w)
            self._adjust_thread_count()
            return f


if __name__ == '__main__':
    def fun():
        print(1 / 0)

    pool = BoundedThreadPoolExecutor(10)
    pool.submit(fun)