是否存在一個與Java的FixedThreadPool相當的Python庫？

Question

Python中是否有一個簡單的ThreadPool庫，例如，pool.execute（function，args）方法，當池已滿時應該被阻塞（直到其中一個線程變爲空閒狀態）？

我已經嘗試了使用多處理程序包中的ThreadPool，但是它的pool.apply_async（）函數在池已滿時不會阻塞。其實，我根本沒有得到它的行爲。

Answer 1

這ActiveState Code Recipes page具有基於Python queue做阻塞的實現。您可以使用add_taskexecute。

## {{{ http://code.activestate.com/recipes/577187/ (r9) 
from Queue import Queue 
from threading import Thread 

class Worker(Thread): 
    """Thread executing tasks from a given tasks queue""" 
    def __init__(self, tasks): 
     Thread.__init__(self) 
     self.tasks = tasks 
     self.daemon = True 
     self.start() 

    def run(self): 
     while True: 
      func, args, kargs = self.tasks.get() 
      try: func(*args, **kargs) 
      except Exception, e: print e 
      self.tasks.task_done() 

class ThreadPool: 
    """Pool of threads consuming tasks from a queue""" 
    def __init__(self, num_threads): 
     self.tasks = Queue(num_threads) 
     for _ in range(num_threads): Worker(self.tasks) 

    def add_task(self, func, *args, **kargs): 
     """Add a task to the queue""" 
     self.tasks.put((func, args, kargs)) 

    def wait_completion(self): 
     """Wait for completion of all the tasks in the queue""" 
     self.tasks.join() 

if __name__ == '__main__': 
    from random import randrange 
    delays = [randrange(1, 10) for i in range(100)] 

    from time import sleep 
    def wait_delay(d): 
     print 'sleeping for (%d)sec' % d 
     sleep(d) 

    # 1) Init a Thread pool with the desired number of threads 
    pool = ThreadPool(20) 

    for i, d in enumerate(delays): 
     # print the percentage of tasks placed in the queue 
     print '%.2f%c' % ((float(i)/float(len(delays)))*100.0,'%') 

     # 2) Add the task to the queue 
     pool.add_task(wait_delay, d) 

    # 3) Wait for completion 
    pool.wait_completion() 
## end of http://code.activestate.com/recipes/577187/ }}} 

Answer 2

from multiprocessing.pool import ThreadPool 

Answer 3

我修改@ckhan答案並添加了回調功能。

#Custom ThreadPool implementation from http://code.activestate.com/recipes/577187/ 
# Usage: 
#  def worker_method(data): 
#   ''' do processing ''' 
#   return data 
#   
#  def on_complete_callback(data, isfailed): 
#   print ('on complete %s' % data) 
#   
#  pool = ThreadPool(5) 
#  for data in range(1,10) 
#   pool.add_task(worker_method, on_complete_callback, data)  
#  pool.wait_completion() 

from Queue import Queue 
from threading import Thread 
import thread 

class Worker(Thread): 
    """Thread executing tasks from a given tasks queue""" 
    def __init__(self, tasks, thread_id): 
     Thread.__init__(self) 
     self.tasks = tasks 
     self.daemon = True 
     self.start() 
     self.thread_id = thread_id 

    def run(self): 
     while True: 
      func, callback, args, kargs = self.tasks.get() 
      try: 
       data = func(*args, **kargs) 
       callback(data, False) 
      except Exception, e:     
       callback(e, True) 
      self.tasks.task_done() 

class ThreadPool: 
    """Pool of threads consuming tasks from a queue""" 
    def __init__(self, num_threads): 
     self.tasks = Queue(num_threads) 
     for i in range(num_threads): Worker(self.tasks, i) 

    def add_task(self, func, callback, *args, **kargs): 
     """Add a task to the queue""" 
     self.tasks.put((func, callback, args, kargs)) 

    def wait_completion(self): 
     """Wait for completion of all the tasks in the queue""" 
     self.tasks.join()