對Python線程池進行詳細說明
下面給大家總結一下自己總結的對Python線程池經(jīng)驗之談,對于那些沒有接觸學習過編程語言或者多開發(fā)語言略懂的用戶而言,Python語言絕對是***的選擇之一,并建議初學的程序員先從Python開始學習編程。
- import Queue, threading, sys
- from threading import Thread
- import time,urllib
- # working thread
- class Worker(Thread):
- worker_count = 0
- def __init__( self, workQueue, resultQueue, timeout = 0, **kwds):
- Thread.__init__( self, **kwds )
- self.id = Worker.worker_count
- Worker.worker_count += 1
- self.setDaemon( True )
- self.workQueue = workQueue
- self.resultQueue = resultQueue
- self.timeout = timeout
- self.start( )
- def run( self ):
- ''' the get-some-work, do-some-work main loop of worker threads '''
- while True:
- try:
- callable, args, kwds = self.workQueue.get(timeout=self.timeout)
- res = callable(*args, **kwds)
- print "worker[%2d]: %s" % (self.id, str(res) )
- self.resultQueue.put( res )
- except Queue.Empty:
- break
- except :
- print 'worker[%2d]' % self.id, sys.exc_info()[:2]
- class WorkerManager:
- def __init__( self, num_of_workers=10, timeout = 1):
- self.workQueue = Queue.Queue()
- self.resultQueue = Queue.Queue()
- self.workers = []
- self.timeout = timeout
- self._recruitThreads( num_of_workers )
- def _recruitThreads( self, num_of_workers ):
- for i in range( num_of_workers ):
- worker = Worker( self.workQueue, self.resultQueue, self.timeout )
- self.workers.append(worker)
- def wait_for_complete( self):
- # ...then, wait for each of them to terminate:
- while len(self.workers):
- worker = self.workers.pop()
- worker.join( )
- if worker.isAlive() and not self.workQueue.empty():
- self.workers.append( worker )
- print "All jobs are are completed."
- def add_job( self, callable, *args, **kwds ):
- self.workQueue.put( (callable, args, kwds) )
- def get_result( self, *args, **kwds ):
- return self.resultQueue.get( *args, **kwds )
Worker類是一個Python線程池,不斷地從workQueue隊列中獲取需要執(zhí)行的任務,執(zhí)行之,并將結果寫入到resultQueue中。這里的workQueue和resultQueue都是現(xiàn)成安全的,其內(nèi)部對各個線程的操作做了互斥。當從workQueue中獲取任務超時,則線程結束。
WorkerManager負責初始化Python線程池,提供將任務加入隊列和獲取結果的接口,并能等待所有任務完成。一個典型的測試例子如下,它用10個線程去下載一個固定頁面的內(nèi)容,實際應用時應該是執(zhí)行不同的任務。
- def test_job(id, sleep = 0.001 ):
- try:
- urllib.urlopen('[url]https://www.gmail.com/[/url]').read()
- except:
- print '[%4d]' % id, sys.exc_info()[:2]
- return id
- def test():
- import socket
- socket.setdefaulttimeout(10)
- print 'start testing'
- wm = WorkerManager(10)
- for i in range(500):
- wm.add_job( test_job, i, i*0.001 )
- wm.wait_for_complete()
- print 'end testing'
【編輯推薦】