Python的線程池實現

 ?1?#coding:utf-8
?2?
?3?#Python的線程池實現
?4?
?5?import?Queue
?6?import?threading
?7?import?sys
?8?import?time
?9?import?urllib
10?
11?#替我們工作的線程池中的線程
12?class?MyThread(threading.Thread):
13??def?__init__(self,?workQueue,?resultQueue,timeout=30,?**kwargs):
14???threading.Thread.__init__(self,?kwargs=kwargs)
15???#線程在結束前等待任務隊列多長時間
16???self.timeout?=?timeout
17???self.setDaemon(True)
18???self.workQueue?=?workQueue
19???self.resultQueue?=?resultQueue
20???self.start()
21?
22??def?run(self):
23???while?True:
24????try:
25?????#從工作隊列中獲取一個任務
26?????callable,?args,?kwargs?=?self.workQueue.get(timeout=self.timeout)
27?????#我們要執行的任務
28?????res?=?callable(args,?kwargs)
29?????#報任務返回的結果放在結果隊列中
30?????self.resultQueue.put(res+"?|?"+self.getName())????
31????except?Queue.Empty:?#任務隊列空的時候結束此線程
32?????break
33????except?:
34?????print?sys.exc_info()
35?????raise
36?????
37?class?ThreadPool:
38??def?__init__(?self,?num_of_threads=10):
39???self.workQueue?=?Queue.Queue()
40???self.resultQueue?=?Queue.Queue()
41???self.threads?=?[]
42???self.__createThreadPool(?num_of_threads?)
43?
44??def?__createThreadPool(?self,?num_of_threads?):
45???for?i?in?range(?num_of_threads?):
46????thread?=?MyThread(?self.workQueue,?self.resultQueue?)
47????self.threads.append(thread)
48?
49??def?wait_for_complete(self):
50???#等待所有線程完成。
51???while?len(self.threads):
52????thread?=?self.threads.pop()
53????#等待線程結束
54????if?thread.isAlive():#判斷線程是否還存活來決定是否調用join
55?????thread.join()
56?????
57??def?add_job(?self,?callable,?*args,?**kwargs?):
58???self.workQueue.put(?(callable,args,kwargs)?)
59?
60?def?test_job(id,?sleep?=?0.001?):
61??html?=?""
62??try:
63???time.sleep(1)
64???conn?=?urllib.urlopen('http://www.google.com/')
65???html?=?conn.read(20)
66??except:
67???print??sys.exc_info()
68??return??html
69?
70?def?test():
71??print?'start?testing'
72??tp?=?ThreadPool(10)
73??for?i?in?range(50):
74???time.sleep(0.2)
75???tp.add_job(?test_job,?i,?i*0.001?)
76??tp.wait_for_complete()
77??#處理結果
78??print?'result?Queue\'s?length?==?%d?'%?tp.resultQueue.qsize()
79??while?tp.resultQueue.qsize():
80???print?tp.resultQueue.get()
81??print?'end?testing'
82?if?__name__?==?'__main__':
83??test() 

?