42. 使用线程池

素颜马尾好姑娘i 2023-07-04 04:54 3阅读 0赞

前面实现了一个多线程的web视频监控服务器，由于服务器资源有限（CPU，内存，带宽），需要对请求连接数（线程数）做限制，避免因资源耗尽而瘫痪。

要求：使用线程池，替代原来的每次请求创建线程。

解决方案：

使用标准库汇总`concurrent.futures`下的`ThreadPoolExecutor`类，对象的`submit()`和`map()`方法可以用来启动线程池中的线程执行任务。

--------------------

*  对于`ThreadPoolExecutor`类：

class concurrent.futures.ThreadPoolExecutor(max_workers=None, thread_name_prefix='', initializer=None, initargs=())

Executor的一个子类，使用最多max\_workers个线程的线程池来异步执行调用。

initializer是在每个工作者线程开始处调用的一个可选可调用对象。initargs是传递给初始化器的元组参数。任何向池提交更多工作的尝试，initializer都将引发一个异常，当前所有等待的工作都会引发一个BrokenThreadPool。

submit(fn, *args, **kwargs)

调度可调用对象fn，以`fn(*args **kwargs)`方式执行并返回Future对象代表可调用对象的执行。

map(func, *iterables, timeout=None, chunksize=1)

类似于`map(func, *iterables)`，不过立即收集iterables而不是延迟再收集，另外func是异步执行的且对func的调用可以并发执行。

>>> import threading, time, random
    
    >>> def f(a, b):
    ...     print(threading.current_thread().name, ':', a, b)
    ...     time.sleep(random.randint(5, 10))
    ...     return a * b
    ...
    >>> from concurrent.futures import ThreadPoolExecutor
    
    >>> executor = ThreadPoolExecutor(3)                #创建3个线程的线程池
    
    >>> executor.submit(f, 2, 3)
    ThreadPoolExecutor-0_0 : 2 3
    <Future at 0x7f831190a4e0 state=running>
    
    >>> future = executor.submit(f, 2, 3)
    ThreadPoolExecutor-0_0 : 2 3
    
    >>> future.result()
    6

>>> executor.map(f, range(1, 6), range(2, 7))
    ThreadPoolExecutor-0_1 : 1 2
    <generator object Executor.map.<locals>.result_iterator at 0x7f830ef736d8>
    >>> ThreadPoolExecutor-0_0 : 2 3
    ThreadPoolExecutor-0_2 : 3 4
    ThreadPoolExecutor-0_1 : 4 5
    ThreadPoolExecutor-0_0 : 5 6
    
    >>> list(executor.map(f, range(1, 6), range(2, 7)))
    ThreadPoolExecutor-0_2 : 1 2
    ThreadPoolExecutor-0_1 : 2 3
    ThreadPoolExecutor-0_0 : 3 4
    ThreadPoolExecutor-0_2 : 4 5
    ThreadPoolExecutor-0_1 : 5 6
    [2, 6, 12, 20, 30]

这里执行map()方法时，首先3个线程执行任务，执行完毕后返回线程池，然后再次得到2个线程执行任务，直到所有任务全部执行完毕。

调用`summit()`是执行一个任务，而调用`map()`是对所有任务依次执行。

--------------------

*  方案示例：

import os, cv2, time, struct, threading
    from http.server import HTTPServer, BaseHTTPRequestHandler
    from socketserver import TCPServer, ThreadingTCPServer
    from threading import Thread, RLock
    from select import select
    from concurrent.futures import ThreadPoolExecutor
    
    class JpegStreamer(Thread):
        def __init__(self, camrea):
            super().__init__()
            self.cap = cv2.VideoCapture(camrea)
            self.lock = RLock()
            self.pipes = { }
    
        def register(self):
            pr, pw = os.pipe()
            self.lock.acquire()
            self.pipes[pr] = pw
            self.lock.release()
            return pr
    
        def unregister(self, pr):
            self.lock.acquire()
            pw = self.pipes.pop(pr)
            self.lock.release()
            os.close(pr)
            os.close(pw)
    
        def capture(self):
            cap = self.cap
            while cap.isOpened():
                ret, frame = cap.read()
                if ret:
                    ret, data = cv2.imencode('.jpg', frame, (cv2.IMWRITE_JPEG_QUALITY, 40))
                    yield data.tostring()
    
        def send_frame(self, frame):
            n = struct.pack('1', len(frame))
            self.lock.acquire()
            if len(self.pipes):
                _, pipes, _ = select([], self.pipes.values(), [], 1)
                for pipe in pipes:
                    os.write(pipe, n)
                    os.write(pipe, frame)
            self.lock.release()
    
        def run(self):
            for frame in self.capture():
                self.send_frame(frame)
    
    
    class JpegRetriever:
        def __init__(self, streamer):
            self.streamer = streamer
            self.local = threading.local()
    
        def retrieve(self):
            while True:
                ns = os.read(self.local.pipe, 8)
                n = struct.unpack('1', ns)[0]
                data = os.read(self.local.pipe, n)
                yield data
    
        def __enter__(self):
            if hasattr(self.local, 'pipe'):
                raise RuntimeError()
    
            self.local.pipe = streamer.register()
            return self.retrieve()
    
        def __exit__(self, *args):
            self.streamer.unregister(self.local.pipe)
            del self.local.pipe
            return True
    
    
    class WebHandler(BaseHTTPRequestHandler):
        retriever = None
    
        @staticmethod
        def set_retriever(retriever):
            WebHandler.retriever = retriever
    
        def do_GET(self):
            if self.retriever is None:
                raise RuntimeError('no retriever')
    
            if self.path != '/':
                return
    
            self.send_response(200)
            self.send_header('Content-type', 'multipart/x-mixed-replace;boundary=jpeg_frame')
            self.end_headers()
    
            with self.retriever as frames:
                for frame in frames:
                    self.send_frame(frame)
    
        def send_frame(self, frame):
            sh = b'--jpeg_frame\r\n'
            sh += b'Content-Type: image/jpeg\r\n'
            sh += b'Content-Length: %d\r\n\r\n' % len(frame)
            self.wfile.write(sh)
            self.wfile.write(frame)
    
    
    class ThreadingPoolTCPServer(ThreadingTCPServer):
        def __init__(self, server_address, RequestHandlerClass, bind_and_activate=True, thread_n=100):
            super().__init__(server_address, RequestHandlerClass, bind_and_activate=True)
            self.executor = ThreadPoolExecutor(thread_n)
        
        def process_request(self, request, client_address):
            self.executor.submit(self.process_request_thread, request, client_address)
    
    
    if __name__ == '__main__':
        # 创建Streamer，开启摄像头采集
        streamer = JpegStreamer(0)
        streamer.start()
    
        # http服务器创建Retriever
        retriever = JpegRetriever(streamer)
        WebHandler.set_retriever(retriever)
    
        # 开启http服务器
        HOST = '192.168.30.128'
        POST = 9000
        print('Start server...(http://%s:%d)' % (HOST, POST))
        httpd = ThreadingPoolTCPServer((HOST, POST), WebHandler, thread_n=3)                #线程池线程数量为3
        httpd.serve_forever()

此时，通过浏览器访问会发现，至多3个窗口可以同时访问，即至多产生3条访问记录，更多访问无法得到响应，因为线程池中的线程数量只为3。

--------------------