線程池python

原創博文，轉載請注明出處

今天在學習python進程與線程時，無意間發現了線程池threadpool模塊，見官方文檔。

模塊使用非常簡單，前提是得需要熟悉線程池的工作原理。

我們知道系統處理任務時，需要為每個請求創建和銷毀對象。當有大量並發任務需要處理時，再使用傳統的多線程就會造成大量的資源創建銷毀導致服務器效率的下降。這時候，線程池就派上用場了。線程池技術為線程創建、銷毀的開銷問題和系統資源不足問題提供了很好的解決方案。

優點：

       （1）可以控制產生線程的數量。通過預先創建一定數量的工作線程並限制其數量，控制線程對象的內存消耗。（2）降低系統開銷和資源消耗。通過對多個請求重用線程，線程創建、銷毀的開銷被分攤到了多個請求上。另外通過限制線程數量，降低虛擬機在垃圾回收方面的開銷。（3）提高系統響應速度。線程事先已被創建，請求到達時可直接進行處理，消除了因線程創建所帶來的延遲，另外多個線程可並發處理。

 線程池的基本實現方法：
　　（1）線程池管理器。創建並維護線程池，根據需要調整池的大小，並監控線程泄漏現象。
　　（2）工作線程。它是一個可以循環執行任務的線程，沒有任務時處於 Wait 狀態，新任務到達時可被喚醒。
　　（3）任務隊列。它提供一種緩沖機制，用以臨時存放待處理的任務，同時作為並發線程的 monitor 對象。
　　（4）任務接口。它是每個任務必須實現的接口，工作線程通過該接口調度任務的執行。
      構建線程池管理器時，首先初始化任務隊列（Queue），運行時通過調用添加任務的方法將任務添加到任務隊列中。之后創建並啟動一定數量的工作線程，將這些線程保存在線程隊列中。線程池管理器在運行時可根據需要增加或減少工作線程數量。工作線程運行時首先鎖定任務隊列，以保證多線程對任務隊列的正確並發訪問，如隊列中有待處理的任務，工作線程取走一個任務並釋放對任務隊列的鎖定，以便其他線程實現對任務隊列的訪問和處理。在獲取任務之后工作線程調用任務接口完成對任務的處理。當任務隊列為空時，工作線程加入到任務隊列的等待線程列表中，此時工作線程處於 Wait 狀態，幾乎不占 CPU 資源。一旦新的任務到達，通過調用任務列表對象的notify方法，從等待線程列表中喚醒一個工作線程以對任務進行處理。通過這種協作模式，既節省了線程創建、銷毀的開銷，又保證了對任務的並發處理，提高了系統的響應速度。

      簡而言之：就是把並發執行的任務傳遞給一個線程池，來替代為每個並發執行的任務都啟動一個新的線程。只要池里有空閑的線程，任務就會分配給一個線程執行。

pool = ThreadPool(poolsize)
requests = makeRequests(some_callable,list_of_args,callback)
[pool.putRequest(req) for req in requests]
pool.wait()

第一行的意思是創建一個可存放poolsize個數目的線程的線程池。

第二行的意思是調用makeRequests創建請求。 some_callable是需要開啟多線程處理的函數，list_of_args是函數參數，callback是可選參數回調，默認是無。

第三行的意思是把運行多線程的函數放入線程池中。

最后一行的意思是等待所有的線程完成工作后退出。

通過分析源代碼，其實發現里面的內容很簡單。

import sys
import threading
import Queue
import traceback


# exceptions
class NoResultsPending(Exception):
    """All work requests have been processed."""
    pass

class NoWorkersAvailable(Exception):
    """No worker threads available to process remaining requests."""
    pass


# internal module helper functions
def _handle_thread_exception(request, exc_info):
    """Default exception handler callback function.

    This just prints the exception info via ``traceback.print_exception``.

    """
    traceback.print_exception(*exc_info)


# utility functions
def makeRequests(callable_, args_list, callback=None,  #用來創建多個任務請求 callback是回調函數處理結果，exc_callback是用來處理發生的異常 
        exc_callback=_handle_thread_exception):
    """Create several work requests for same callable with different arguments.

    Convenience function for creating several work requests for the same
    callable where each invocation of the callable receives different values
    for its arguments.

    ``args_list`` contains the parameters for each invocation of callable.
    Each item in ``args_list`` should be either a 2-item tuple of the list of
    positional arguments and a dictionary of keyword arguments or a single,
    non-tuple argument.

    See docstring for ``WorkRequest`` for info on ``callback`` and
    ``exc_callback``.

    """
    requests = []
    for item in args_list:
        if isinstance(item, tuple):
            requests.append(
                WorkRequest(callable_, item[0], item[1], callback=callback,
                    exc_callback=exc_callback)
            )
        else:
            requests.append(
                WorkRequest(callable_, [item], None, callback=callback,
                    exc_callback=exc_callback)
            )
    return requests


# classes
class WorkerThread(threading.Thread):     #工作線程
    """Background thread connected to the requests/results queues.

    A worker thread sits in the background and picks up work requests from
    one queue and puts the results in another until it is dismissed.

    """

    def __init__(self, requests_queue, results_queue, poll_timeout=5, **kwds):
        """Set up thread in daemonic mode and start it immediatedly.

        ``requests_queue`` and ``results_queue`` are instances of
        ``Queue.Queue`` passed by the ``ThreadPool`` class when it creates a new
        worker thread.

        """
        threading.Thread.__init__(self, **kwds)
        self.setDaemon(1)
        self._requests_queue = requests_queue     #任務隊列
        self._results_queue = results_queue       #結果隊列    
        self._poll_timeout = poll_timeout
        self._dismissed = threading.Event()       
        self.start()

    def run(self):
        """Repeatedly process the job queue until told to exit."""
        while True:
            if self._dismissed.isSet():  #如果標識位設為True，則表示線程非阻塞
                # we are dismissed, break out of loop
                break
            # get next work request. If we don't get a new request from the
            # queue after self._poll_timout seconds, we jump to the start of
            # the while loop again, to give the thread a chance to exit.
            try:
                request = self._requests_queue.get(True, self._poll_timeout)#獲取待處理任務，block設為True，標識線程同步 ，並設置超時時間 
            except Queue.Empty:
                continue
            else:
                if self._dismissed.isSet():再次判斷，因為在取任務期間，線程有可能被掛起
                    # we are dismissed, put back request in queue and exit loop
                    self._requests_queue.put(request) #添加任務到任務隊列
                    break
                try:
                    result = request.callable(*request.args, **request.kwds)
                    self._results_queue.put((request, result))
                except:
                    request.exception = True
                    self._results_queue.put((request, sys.exc_info()))

    def dismiss(self):
        """Sets a flag to tell the thread to exit when done with current job."""
        self._dismissed.set()


class WorkRequest:      #創建單個任務請求
    """A request to execute a callable for putting in the request queue later.

    See the module function ``makeRequests`` for the common case
    where you want to build several ``WorkRequest`` objects for the same
    callable but with different arguments for each call.

    """

    def __init__(self, callable_, args=None, kwds=None, requestID=None,
            callback=None, exc_callback=_handle_thread_exception):
        """Create a work request for a callable and attach callbacks.

        A work request consists of the a callable to be executed by a
        worker thread, a list of positional arguments, a dictionary
        of keyword arguments.

        A ``callback`` function can be specified, that is called when the
        results of the request are picked up from the result queue. It must
        accept two anonymous arguments, the ``WorkRequest`` object and the
        results of the callable, in that order. If you want to pass additional
        information to the callback, just stick it on the request object.

        You can also give custom callback for when an exception occurs with
        the ``exc_callback`` keyword parameter. It should also accept two
        anonymous arguments, the ``WorkRequest`` and a tuple with the exception
        details as returned by ``sys.exc_info()``. The default implementation
        of this callback just prints the exception info via
        ``traceback.print_exception``. If you want no exception handler
        callback, just pass in ``None``.

        ``requestID``, if given, must be hashable since it is used by
        ``ThreadPool`` object to store the results of that work request in a
        dictionary. It defaults to the return value of ``id(self)``.

        """
        if requestID is None:
            self.requestID = id(self) #id返回對象的內存地址
        else:
            try:
                self.requestID = hash(requestID) #哈希處理 
            except TypeError:
                raise TypeError("requestID must be hashable.")
        self.exception = False
        self.callback = callback
        self.exc_callback = exc_callback
        self.callable = callable_
        self.args = args or []
        self.kwds = kwds or {}

    def __str__(self):
        return "<WorkRequest id=%s args=%r kwargs=%r exception=%s>" % \
            (self.requestID, self.args, self.kwds, self.exception)

class ThreadPool:  #線程池管理器
    """A thread pool, distributing work requests and collecting results.

    See the module docstring for more information.

    """

    def __init__(self, num_workers, q_size=0, resq_size=0, poll_timeout=5):
        """Set up the thread pool and start num_workers worker threads.

        ``num_workers`` is the number of worker threads to start initially.

        If ``q_size > 0`` the size of the work *request queue* is limited and
        the thread pool blocks when the queue is full and it tries to put
        more work requests in it (see ``putRequest`` method), unless you also
        use a positive ``timeout`` value for ``putRequest``.

        If ``resq_size > 0`` the size of the *results queue* is limited and the
        worker threads will block when the queue is full and they try to put
        new results in it.

        .. warning:
            If you set both ``q_size`` and ``resq_size`` to ``!= 0`` there is
            the possibilty of a deadlock, when the results queue is not pulled
            regularly and too many jobs are put in the work requests queue.
            To prevent this, always set ``timeout > 0`` when calling
            ``ThreadPool.putRequest()`` and catch ``Queue.Full`` exceptions.

        """
        self._requests_queue = Queue.Queue(q_size)  #任務隊列
        self._results_queue = Queue.Queue(resq_size) #結果隊列
        self.workers = []  #工作線程
        self.dismissedWorkers = [] #睡眠線程
        self.workRequests = {}  #一個字典 鍵是id 值是request
        self.createWorkers(num_workers, poll_timeout)

    def createWorkers(self, num_workers, poll_timeout=5):
        """Add num_workers worker threads to the pool.

        ``poll_timout`` sets the interval in seconds (int or float) for how
        ofte threads should check whether they are dismissed, while waiting for
        requests.

        """
        for i in range(num_workers):
            self.workers.append(WorkerThread(self._requests_queue,
                self._results_queue, poll_timeout=poll_timeout))

    def dismissWorkers(self, num_workers, do_join=False):
        """Tell num_workers worker threads to quit after their current task."""
        dismiss_list = []
        for i in range(min(num_workers, len(self.workers))):
            worker = self.workers.pop()
            worker.dismiss()
            dismiss_list.append(worker)

        if do_join:
            for worker in dismiss_list:
                worker.join()
        else:
            self.dismissedWorkers.extend(dismiss_list)

    def joinAllDismissedWorkers(self):
        """Perform Thread.join() on all worker threads that have been dismissed.
        """
        for worker in self.dismissedWorkers:
            worker.join()
        self.dismissedWorkers = []

    def putRequest(self, request, block=True, timeout=None):
        """Put work request into work queue and save its id for later."""
        assert isinstance(request, WorkRequest)
        # don't reuse old work requests
        assert not getattr(request, 'exception', None)
        self._requests_queue.put(request, block, timeout)
        self.workRequests[request.requestID] = request  #確立一對一對應關系 一個id對應一個request 

    def poll(self, block=False):#處理任務,
        """Process any new results in the queue."""
        while True:
            # still results pending?
            if not self.workRequests: #沒有任務
                raise NoResultsPending
            # are there still workers to process remaining requests?
            elif block and not self.workers:#無工作線程
                raise NoWorkersAvailable
            try:
                # get back next results
                request, result = self._results_queue.get(block=block)
                # has an exception occured?
                if request.exception and request.exc_callback:
                    request.exc_callback(request, result)
                # hand results to callback, if any
                if request.callback and not \
                       (request.exception and request.exc_callback):
                    request.callback(request, result)
                del self.workRequests[request.requestID]
            except Queue.Empty:
                break

    def wait(self):
        """Wait for results, blocking until all have arrived."""
        while 1:
            try:
                self.poll(True)
            except NoResultsPending:
                break

 有三個類 ThreadPool，workRequest，workThread，

第一步我們需要建立一個線程池調度ThreadPool實例（根據參數而產生多個線程works），然后再通過makeRequests創建具有多個不同參數的任務請求workRequest，然后把任務請求用putRequest放入線程池中的任務隊列中，此時線程workThread就會得到任務callable，然后進行處理后得到結果，存入結果隊列。如果存在callback就對結果調用函數。

注意：結果隊列中的元素是元組(request,result)這樣就一一對應了。 

在我的下一篇文章關於爬蟲方面的，我將嘗試使用線程池來加強爬蟲的爬取效率 。