Python之線程池

版本一：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import Queue
import threading
 
 
class ThreadPool(object):
 
    def __init__(self, max_num=20):
        self.queue = Queue.Queue(max_num)
        for i in xrange(max_num):
            self.queue.put(threading.Thread)
 
    def get_thread(self):
        return self.queue.get()
 
    def add_thread(self):
        self.queue.put(threading.Thread)
 
"""
pool = ThreadPool(10)
 
def func(arg, p):
    print arg
    import time
    time.sleep(2)
    p.add_thread()
 
 
for i in xrange(30):
    thread = pool.get_thread()
    t = thread(target=func, args=(i, pool))
    t.start()
"""

版本二：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
 
"""
custom ThreadPool
 
How to use:
 
pool = ThreadPool(1)
 
def callback(status, result):
    # status, execute action status
    # result, execute action return value
 
    pass
def action(i):
    pass
 
for i in range(20):
    if pool.stop:
        pool.terminal()
        break
    ret = pool.run(action, (i,), callback)
 
print 'end'
 
 
"""
 
 
import Queue
import threading
import contextlib
 
StopEvent = object()
 
 
class ThreadPool(object):
 
    def __init__(self, max_num):
        self.q = Queue.Queue(max_num)
        self.max_num = max_num
        self.cancel = False
        self.generate_list = []
        self.free_list = []
 
    def run(self, func, args, callback=None):
        """
        線程池執行一個任務
        :param func: 任務函數
        :param args: 任務函數所需參數
        :param callback: 任務執行失敗或成功後執行的回調函數，回調函數有兩個參數一、任務函數執行狀態；二、任務函數返回值（默認爲None，即：不執行回調函數）
        :return: 若是線程池已經終止，則返回True不然None
        """
        if self.cancel:
            return True
        if len(self.free_list) == 0 and len(self.generate_list) < self.max_num:
            self.generate_thread()
        w = (func, args, callback,)
        self.q.put(w)
 
    def generate_thread(self):
        """
        建立一個線程
        """
        t = threading.Thread(target=self.call)
        t.start()
 
    def call(self):
        """
        循環去獲取任務函數並執行任務函數
        """
        current_thread = threading.currentThread
        self.generate_list.append(current_thread)
 
        event = self.q.get()
        while event != StopEvent:
            func, arguments, callback = event
            try:
                result = func(*arguments)
                success = True
            except Exception, e:
                success = False
                result = None
 
            if callback is not None:
                try:
                    callback(success, result)
                except Exception, e:
                    pass
 
            with self.worker_state(self.free_list, current_thread):
                event = self.q.get()
        else:
 
            self.generate_list.remove(current_thread)
 
    def terminal(self):
        """
        終止線程池中的全部線程
        """
        self.cancel = True
        full_size = len(self.generate_list)
        while full_size:
            self.q.put(StopEvent)
            full_size -= 1
 
    @contextlib.contextmanager
    def worker_state(self, state_list, worker_thread):
        """
        用於記錄線程中正在等待的線程數
        """
        state_list.append(worker_thread)
        try:
            yield
        finally:
            state_list.remove(worker_thread)

更多參見：twisted.python.threadpool

上下文管理：https://docs.python.org/2/library/contextlib.html