learning at 20191014

vodaka · vodaka · commit f9306d819541 · 2019-10-14T13:33:16.000+08:00
diff --git a/python3_programming_tricks/ch08/8-5.py b/python3_programming_tricks/ch08/8-5.py
@@ -0,0 +1,134 @@
+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
+
+class JpegStream(Thread):
+    def __init__(self, camera):
+        super().__init__()
+        self.cap =cv2.VideoCapture(camera)
+        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('l', 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('l', ns)[0]
+            data = os.read(self.local.pipe, n)
+            yield data
+
+    def __enter__(self):
+        if hasattr(self.local, 'pipe'):
+            raise RuntimeError()
+
+        # 将所有pipe变成线程本地数据
+        # self.pipe = streamer.register()
+        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_fame(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)
+
+from concurrent.futures import ThreadPoolExecutor
+
+class ThreadingPoolTCPServer(ThreadingTCPServer):
+    def __init__(self, server_address, RequestHandlerClass, bind_and_active=True, thread_n=100):
+        super.__init__(server_address, RequestHandlerClass, bind_and_active=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 = 'localhost'
+    PORT = 9000
+    print('Start Server ... (http://%s:%s)' % (HOST, PORT))
+    # 为了支持多线程，需要将TCPServer改成ThreadingTCPServer,多线程版本
+    # 每个线程需要独立的管道
+    # httpd = TCPServer((HOST, PORT), WebHandler)
+    # httpd = ThreadingTCPServer((HOST, PORT), WebHandler)
+    httpd = ThreadingPoolTCPServer((HOST, PORT), WebHandler, thread_n=3)
+    httpd.serve_forever()
+
diff --git a/python3_programming_tricks/ch08/8-6.py b/python3_programming_tricks/ch08/8-6.py
diff --git a/python3_programming_tricks/ch08/8_multithread_and_multiprocess.md b/python3_programming_tricks/ch08/8_multithread_and_multiprocess.md
@@ -74,3 +74,20 @@ GIL属于进程，不同进程中的GIL是相互独立的
 之前实现了一个多线程web视频监控服务器，由于服务器资源有限(CPU,内存,带宽),需要对请求连接数(线程数)做限制，避免因资源耗尽而瘫痪。
 
 可以使用线程池，替代原来的每次请求创建线程。
+线程池：提前创建固定数量的线程，用的时候去池子中取，用完在放回
+
+**解决方案**
+
+使用标准库中concurrent.futures下的ThreadPoolExecutor对象的submit和map方法可以用来启动线程池中线程执行任务。
+
+## 8.6 如何使用多进程
+
+**实际案例**
+
+由于python中全局解释器锁(GIL)的存在，在任意时刻只允许一个线程在解释器中运行。因此python的多线程不适合处理CPU密集型任务
+
+想要处理CPU密集型任务，可以使用多进程模型。
+
+**解决方案**
+
+使用标准库中multiprocessing.Process, 它可以启动子进程执行任务操作接口，进程间通信，进程间同步等都与Threading.Thread类似
